From 99575b8c8b75234957ac56383d328b18da81e56a Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Wed, 14 Oct 2026 01:46:06 +0800 Subject: [PATCH 1/3] Activate shared SpanOp journaling with independent route consumers --- bt-daemon/src/delivery_ledger.rs | 4 +- bt-daemon/src/dispatch.rs | 706 ++++-------------------- bt-daemon/src/journal.rs | 68 ++- bt-daemon/src/pipeline/consumer.rs | 214 +++++++ bt-daemon/src/pipeline/delivery.rs | 7 +- bt-daemon/src/pipeline/journal.rs | 126 ++++- bt-daemon/src/pipeline/mod.rs | 7 +- bt-daemon/src/pipeline/routing.rs | 2 +- bt-daemon/src/pipeline/translation.rs | 501 +++++++++++++++++ bt-daemon/src/server/ingress.rs | 5 + bt-daemon/src/server/maintenance.rs | 47 +- bt-daemon/src/server/mod.rs | 4 +- bt-daemon/src/server/recovery.rs | 30 +- bt-daemon/src/server/rpc.rs | 11 +- bt-daemon/src/server/sessions.rs | 70 ++- bt-daemon/src/span_processor.rs | 3 + bt-daemon/src/translate/codex.rs | 37 +- bt-daemon/src/translate/mod.rs | 16 + bt-daemon/src/translate/turn_lineage.rs | 15 + bt-daemon/tests/cursor_pipeline.rs | 31 +- bt-daemon/tests/distributed_tracing.rs | 15 + bt-daemon/tests/pipeline.rs | 123 +++-- bt-daemon/tests/root_provenance.rs | 48 +- 23 files changed, 1360 insertions(+), 730 deletions(-) create mode 100644 bt-daemon/src/pipeline/consumer.rs create mode 100644 bt-daemon/src/pipeline/translation.rs diff --git a/bt-daemon/src/delivery_ledger.rs b/bt-daemon/src/delivery_ledger.rs index ace9b825..b389cdf8 100644 --- a/bt-daemon/src/delivery_ledger.rs +++ b/bt-daemon/src/delivery_ledger.rs @@ -336,11 +336,11 @@ impl Sink for LedgerSink { if let std::collections::hash_map::Entry::Vacant(entry) = self.span_origins.entry(row.span_id.clone()) { - let origin = OriginSnapshot { + let origin = row.origin.clone().unwrap_or_else(|| OriginSnapshot { plugin_version: self.plugin_version.clone(), bt_version: self.bt_version.clone(), source_version: self.source_version.clone(), - }; + }); if self.ledger.is_some() { self.pending_origins .insert(row.span_id.clone(), origin.clone()); diff --git a/bt-daemon/src/dispatch.rs b/bt-daemon/src/dispatch.rs index d1123eef..42cf3796 100644 --- a/bt-daemon/src/dispatch.rs +++ b/bt-daemon/src/dispatch.rs @@ -7,14 +7,14 @@ //! event is durably journaled. Translation, correlation, and delivery here are //! entirely out-of-band from hook execution. -use crate::delivery_ledger::LedgerSink; -use crate::journal::JournalWriter; -use crate::pipeline::routing::RoutingSink; -use crate::pipeline::{Completion, Delivery}; +use crate::pipeline::{ + consumer::{Consumer, ConsumerOptions}, + translation::Translation, +}; use crate::sink::SinkFactory; -use crate::translate::{Registry, SessionCtx}; -use crate::wire::{Envelope, SessionRoute}; +use crate::translate::Registry; use parking_lot::Mutex; +use crate::wire::Envelope; use std::path::PathBuf; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; @@ -45,10 +45,6 @@ enum SessionMsg { /// Where to rebuild a session's translator state from, streamed at startup. pub struct ReplayPlan { pub journal_path: PathBuf, - /// Replay stops here — the journal's length when this session was - /// created, so the event creating it is not replayed and then delivered - /// a second time from the queue. - pub through: u64, /// Events through this offset reached the backend before the prior daemon /// stopped. Replay still rebuilds translator state from them, but must not /// deliver them again. @@ -57,17 +53,12 @@ pub struct ReplayPlan { pub(crate) struct SessionOptions { pub session_id: String, - pub translator_session_id: String, pub source: String, pub bt_version: String, - pub replay: Option, pub config: crate::wire::SessionConfig, - pub correlation_key: String, - pub route: SessionRoute, - pub correlation: Arc, pub data_dir: PathBuf, - pub journal: Arc>, - pub correlation_changed: Arc, + pub translation: Arc, + pub cursor: crate::pipeline::journal::Cursor, } /// Handle to one live session: its queue plus observable counters/state. @@ -76,6 +67,8 @@ pub struct Session { tx: mpsc::Sender, pub counters: Arc, pub last_error: Arc>>, + source_progress: Arc, + pub source_error: Arc>>, pub permalink: Arc>>, last_activity: Mutex, } @@ -84,63 +77,131 @@ impl Session { /// Spawn a session's actor task and return its handle. pub fn spawn( options: SessionOptions, - translators: Arc, + _translators: Arc, sink_factory: Arc, ) -> Arc { - let SessionOptions { - session_id, - translator_session_id, - source, - bt_version, - replay, - config, - correlation_key, - route, - correlation, - data_dir, - journal, - correlation_changed, - } = options; - let (tx, rx) = mpsc::channel(QUEUE_CAPACITY); + let (tx, mut rx) = mpsc::channel(QUEUE_CAPACITY); let counters = Arc::new(Counters::default()); let last_error = Arc::new(Mutex::new(None)); let permalink = Arc::new(Mutex::new(None)); - - let actor = SessionActor { - session_id: session_id.clone(), - translator_session_id, - source: source.clone(), - bt_version, - translators, - sink_factory, - counters: counters.clone(), - last_error: last_error.clone(), - permalink: permalink.clone(), - replay, - config, - correlation_key, - route, - correlation, - data_dir, - journal, - correlation_changed, - }; - tokio::spawn(actor.run(rx)); + let source = options.source.clone(); + let route = Consumer::spawn( + options.translation.journal.clone(), + options.cursor, + ConsumerOptions { + source: options.source.clone(), + session_id: options.session_id.clone(), + version: options.bt_version, + data_dir: options.data_dir, + config: options.config, + counters: counters.clone(), + error: last_error.clone(), + permalink: permalink.clone(), + factory: sink_factory, + }, + ); + let producer = options.translation; + let source_error = producer.error.clone(); + let source_progress = producer.progress.clone(); + let errors = last_error.clone(); + let queued = counters.clone(); + tokio::spawn(async move { + while let Some(message) = rx.recv().await { + let finalize = matches!(&message, SessionMsg::Finalize(_)); + match message { + SessionMsg::Event(env, through) => { + if let Some(config) = env.config { + route.configure(config); + } + if let Err(error) = producer.advance(through).await { + *errors.lock().unwrap() = Some(error.to_string()); + } + } + SessionMsg::Configure(config, reply) => { + route.configure(*config); + let _ = reply.send(()); + } + SessionMsg::Barrier(reply) => { + let _ = producer.barrier().await; + let _ = reply.send(()); + } + SessionMsg::Flush(reply) | SessionMsg::Finalize(reply) => { + let kind = if finalize { + crate::journal::ControlKind::Finalize + } else { + crate::journal::ControlKind::Checkpoint + }; + let target = producer.control(kind).await; + let route = route.clone(); + let queued = queued.clone(); + let errors = errors.clone(); + tokio::spawn(async move { + let mut reply = reply; + let waiting = async { + match target { + Ok(tail) => match route.through(tail).await { + Ok(()) => 0, + Err(error) => { + *errors.lock().unwrap() = Some(error); + queued.queued.load(Ordering::Relaxed).max(1) + } + }, + Err(error) => { + *errors.lock().unwrap() = Some(error); + queued.queued.load(Ordering::Relaxed).max(1) + } + } + }; + tokio::select! { _=reply.closed()=>{}, pending=waiting=> { let _=reply.send(pending); } } + }); + } + SessionMsg::Shutdown(reply) => { + if producer.routes.fetch_sub(1, Ordering::Relaxed) == 1 { + // Process retirement is not a native session end. + if let Ok(tail) = + producer.control(crate::journal::ControlKind::Retire).await + { + let _ = tokio::time::timeout( + std::time::Duration::from_secs(2), + route.through(tail), + ) + .await; + } + } + route.shutdown().await; + if producer.routes.load(Ordering::Relaxed) == 0 { + producer.shutdown().await; + } + let _ = reply.send(()); + break; + } + } + } + }); Arc::new(Session { source, tx, counters, last_error, + source_error, + source_progress, permalink, last_activity: Mutex::new(Instant::now()), }) } + pub(crate) fn pending(&self) -> u64 { + self.counters + .queued + .load(Ordering::Relaxed) + .saturating_add(self.source_progress.pending()) + } + /// Queue a journaled event without waiting for translation or delivery. pub async fn enqueue(&self, env: Envelope, journal_through: u64) -> anyhow::Result<()> { self.touch(); - self.counters.queued.fetch_add(1, Ordering::Relaxed); + self.tx .send(SessionMsg::Event(Box::new(env), journal_through)) .await @@ -451,522 +512,6 @@ async fn capture_grok_transcript_pass( } } -struct SessionActor { - session_id: String, - translator_session_id: String, - source: String, - bt_version: String, - translators: Arc, - sink_factory: Arc, - counters: Arc, - last_error: Arc>>, - permalink: Arc>>, - replay: Option, - config: crate::wire::SessionConfig, - correlation_key: String, - route: SessionRoute, - correlation: Arc, - data_dir: PathBuf, - journal: Arc>, - correlation_changed: Arc, -} - -#[derive(Clone, Copy)] -enum BatchMode { - Live, - Replay, - Checkpoint, - TerminalFinalize, - ShutdownFinalize, -} - -impl BatchMode { - fn errors(self) -> (&'static str, &'static str) { - match self { - Self::Live => ("translate failed", "sink emit failed"), - Self::Replay => ("journal replay failed", "sink replay emit failed"), - Self::Checkpoint => ( - "translate checkpoint failed", - "sink emit (checkpoint) failed", - ), - Self::TerminalFinalize | Self::ShutdownFinalize => ( - "translate finalization failed", - "sink emit (finalization) failed", - ), - } - } - - fn observes_correlation(self) -> bool { - !matches!(self, Self::ShutdownFinalize) - } -} - -impl SessionActor { - async fn run(self, mut rx: mpsc::Receiver) { - let mut translator = match self - .translators - .create_checked_with_session_namespace(&self.source, &self.translator_session_id) - { - Ok(translator) => translator, - Err(error) => { - self.set_error(format!("translator init failed: {error}")); - return; - } - }; - let sink = match self.sink_factory.create(&self.session_id, &self.source) { - Ok(s) => s, - Err(e) => { - self.set_error(format!("sink init failed: {e}")); - // Still drain the queue so the daemon's counters settle and - // callers waiting on flush don't hang. - while let Some(msg) = rx.recv().await { - match msg { - SessionMsg::Event(_, _) => { - self.counters.queued.fetch_sub(1, Ordering::Relaxed); - } - SessionMsg::Configure(_, r) => { - let _ = r.send(()); - } - SessionMsg::Barrier(r) => { - let _ = r.send(()); - } - SessionMsg::Flush(r) => { - let _ = r.send(0); - } - SessionMsg::Finalize(r) => { - let _ = r.send(0); - } - SessionMsg::Shutdown(r) => { - let _ = r.send(()); - break; - } - } - } - return; - } - }; - let sink: Box = Box::new( - LedgerSink::new( - sink, - Some(&self.data_dir), - &self.source, - &self.session_id, - Some(&self.config), - &self.bt_version, - ) - .await, - ); - let mut sink: Box = Box::new(RoutingSink::spawn( - sink, - self.source.clone(), - self.session_id.clone(), - self.data_dir.clone(), - self.last_error.clone(), - )); - let mut ctx = SessionCtx { - session_id: self.session_id.clone(), - config: Some(self.config.clone()), - }; - sink.configure(&self.config); - self.refresh_permalink(sink.as_ref()); - // Rebuild translator state before accepting the first new event. - // Stable span ids make this both crash recovery and a complete copy - // when an existing source session is sent to another destination. - let mut acknowledged_through = self - .replay - .as_ref() - .map_or(0, |plan| plan.acknowledged_through); - let mut pending_through = acknowledged_through; - // A failed sink write makes the contiguous delivery boundary unknown. - // Keep the journal conservative for this actor generation rather than - // checkpointing past an event that may need recovery delivery. - let mut checkpointable = self.replay_into(&mut translator, &mut sink, &ctx).await; - - while let Some(msg) = rx.recv().await { - match msg { - SessionMsg::Event(env, journal_through) => { - let correlation_barrier = is_tool_lifecycle_event(&env.event); - if let Some(cfg) = &env.config { - sink.configure(cfg); - ctx.config = Some(cfg.clone()); - self.refresh_permalink(sink.as_ref()); - } - sink.set_capture_versions( - env.plugin_version.as_deref(), - env.source_version.as_deref(), - ); - let translated = translator.handle(&env, &ctx); - let (correlation_changed, delivered) = self - .emit_translator_batches( - &mut translator, - &mut sink, - &ctx, - translated, - BatchMode::Live, - ) - .await; - if correlation_changed || correlation_barrier { - if let Err(error) = crate::server::persist_active_parent_snapshot( - &self.data_dir, - &self.correlation_key, - &self.correlation, - ) - .await - { - self.set_error(error); - } - self.correlation_changed.notify_one(); - } - self.counters.queued.fetch_sub(1, Ordering::Relaxed); - if delivered && checkpointable { - pending_through = pending_through.max(journal_through); - } else { - checkpointable = false; - } - } - SessionMsg::Configure(config, reply) => { - sink.configure(&config); - ctx.config = Some(*config); - self.refresh_permalink(sink.as_ref()); - let _ = reply.send(()); - } - SessionMsg::Barrier(reply) => { - let _ = reply.send(()); - } - SessionMsg::Flush(reply) => { - checkpointable &= self - .checkpoint_and_flush(&mut translator, &mut sink, &ctx) - .await; - if checkpointable { - checkpointable &= self - .record_delivery_checkpoint( - &mut sink, - pending_through, - &mut acknowledged_through, - ) - .await; - } - let _ = reply.send(self.counters.queued.load(Ordering::Relaxed)); - } - SessionMsg::Finalize(reply) => { - checkpointable &= self - .finalize_and_flush( - &mut translator, - &mut sink, - &ctx, - BatchMode::TerminalFinalize, - ) - .await; - if checkpointable { - checkpointable &= self - .record_delivery_checkpoint( - &mut sink, - pending_through, - &mut acknowledged_through, - ) - .await; - } - let _ = reply.send(self.counters.queued.load(Ordering::Relaxed)); - } - SessionMsg::Shutdown(reply) => { - checkpointable &= self - .finalize_and_flush( - &mut translator, - &mut sink, - &ctx, - BatchMode::ShutdownFinalize, - ) - .await; - if checkpointable { - self.record_delivery_checkpoint( - &mut sink, - pending_through, - &mut acknowledged_through, - ) - .await; - } - let _ = reply.send(()); - break; - } - } - } - } - - /// Emit a translator result and every bounded continuation it schedules. - /// A continuation may be empty (for example, irrelevant rollout rows), so - /// only `None` signals that the translator is fully caught up. - async fn emit_translator_batches( - &self, - translator: &mut Box, - sink: &mut Box, - ctx: &SessionCtx, - first: anyhow::Result>, - mode: BatchMode, - ) -> (bool, bool) { - let (translate_error, emit_error) = mode.errors(); - let Some(config) = ctx.config.as_ref() else { - self.set_error("session configuration is missing".into()); - return (false, false); - }; - let mut next = match first { - Ok(ops) => Some(ops), - Err(e) => { - self.set_error(format!("{translate_error}: {e}")); - return (false, false); - } - }; - let mut correlation_changed = false; - let mut delivered = true; - while let Some(ops) = next { - if !ops.is_empty() { - if mode.observes_correlation() { - let changed = self.observe_ops(&ops, config); - correlation_changed |= changed; - if changed { - self.persist_correlation_if_changed(true).await; - } - } - if !ops.is_empty() { - match sink.emit(&ops).await { - Ok(n) => { - self.counters.spans_emitted.fetch_add(n, Ordering::Relaxed); - } - Err(e) => { - delivered = false; - if let Some(failure) = - e.downcast_ref::() - { - self.counters - .spans_emitted - .fetch_add(failure.emitted, Ordering::Relaxed); - if failure.report { - self.set_error(format!("{emit_error}: {e}")); - } - } else { - self.set_error(format!("{emit_error}: {e}")); - } - } - } - } - } - next = match translator.drain_pending(ctx) { - Ok(next) => next, - Err(e) => { - self.set_error(format!("{translate_error}: {e}")); - delivered = false; - None - } - }; - } - (correlation_changed, delivered) - } - - /// Stream the journal through the translator, emitting each entry's spans - /// as they are produced. Nothing is accumulated across entries: peak - /// memory is one journal entry and the ops it yields, so recovering a - /// long session costs the same as running it. - async fn replay_into( - &self, - translator: &mut Box, - sink: &mut Box, - ctx: &SessionCtx, - ) -> bool { - let Some(plan) = &self.replay else { - return true; - }; - let mut reader = match crate::journal::JournalReader::open(&plan.journal_path, plan.through) - .await - { - Ok(Some(reader)) => reader, - Ok(None) => return true, - Err(error) => { - tracing::warn!(session_id = %self.session_id, "journal replay skipped: {error}"); - return false; - } - }; - let mut delivered = true; - loop { - let entry = match reader.next_record().await { - Ok(Some(entry)) => entry, - Ok(None) => break, - Err(error) => { - tracing::warn!(session_id = %self.session_id, "journal replay stopped: {error}"); - delivered = false; - break; - } - }; - let entry_through = entry.through; - let crate::journal::JournalRecord::Event(entry) = entry.record else { - continue; - }; - let mut env = crate::journal::envelope_from_redacted(entry); - let Some(canonical_source) = self.translators.canonical_source(&env.source) else { - continue; - }; - if canonical_source != self.source { - continue; - } - env.source = self.source.clone(); - if env.source == "claude-code" { - if let Some(payload) = env.payload.as_object_mut() { - payload.insert("_bt_transcript_replay".to_string(), serde_json::json!(true)); - } - } - sink.set_capture_versions(env.plugin_version.as_deref(), env.source_version.as_deref()); - let translated = translator.handle(&env, ctx); - if entry_through <= plan.acknowledged_through { - self.replay_without_delivery(translator, ctx, translated) - .await; - } else { - let (_, replayed) = self - .emit_translator_batches(translator, sink, ctx, translated, BatchMode::Replay) - .await; - delivered &= replayed; - } - } - delivered - } - - /// Continue a replayed translator exactly as usual but deliberately omit - /// sink delivery for entries recorded as durably accepted. This lets every - /// agent recover correlation/open-span state without retriggering backend - /// side effects such as online scoring. - async fn replay_without_delivery( - &self, - translator: &mut Box, - ctx: &SessionCtx, - first: anyhow::Result>, - ) { - let Some(config) = ctx.config.as_ref() else { - self.set_error("session configuration is missing".into()); - return; - }; - let mut next = match first { - Ok(ops) => Some(ops), - Err(error) => { - self.set_error(format!("journal replay failed: {error}")); - return; - } - }; - while let Some(ops) = next { - if !ops.is_empty() { - let _ = self.observe_ops(&ops, config); - } - next = match translator.drain_pending(ctx) { - Ok(next) => next, - Err(error) => { - self.set_error(format!("journal replay failed: {error}")); - None - } - }; - } - } - - async fn checkpoint_and_flush( - &self, - translator: &mut Box, - sink: &mut Box, - ctx: &SessionCtx, - ) -> bool { - let translated = translator.checkpoint(ctx); - let (correlation_changed, delivered) = self - .emit_translator_batches(translator, sink, ctx, translated, BatchMode::Checkpoint) - .await; - self.persist_correlation_if_changed(correlation_changed) - .await; - delivered - } - - async fn finalize_and_flush( - &self, - translator: &mut Box, - sink: &mut Box, - ctx: &SessionCtx, - mode: BatchMode, - ) -> bool { - let translated = translator.finalize(ctx); - let (correlation_changed, delivered) = self - .emit_translator_batches(translator, sink, ctx, translated, mode) - .await; - self.persist_correlation_if_changed(correlation_changed) - .await; - delivered - } - - fn observe_ops( - &self, - ops: &[crate::translate::SpanOp], - config: &crate::wire::SessionConfig, - ) -> bool { - let mut routed = ops.to_vec(); - crate::pipeline::prepare::Prepare::new(&self.source).apply(&mut routed, config); - self.correlation - .observe_ops(&self.correlation_key, &self.route, config, &routed) - } - - async fn persist_correlation_if_changed(&self, changed: bool) { - if !changed { - return; - } - if let Err(error) = crate::server::persist_active_parent_snapshot( - &self.data_dir, - &self.correlation_key, - &self.correlation, - ) - .await - { - self.set_error(error); - } - } - - /// False means delivery was rejected and this actor must never advance its - /// contiguous event boundary. Deferred delivery and failed checkpoint writes - /// leave accepted work eligible for a later checkpoint. - async fn record_delivery_checkpoint( - &self, - sink: &mut Box, - through: u64, - acknowledged_through: &mut u64, - ) -> bool { - if let Err(error) = sink.flush().await { - self.set_error(format!("sink flush failed: {error}")); - return false; - } - self.refresh_permalink(sink.as_ref()); - if sink.has_pending_delivery() || through <= *acknowledged_through { - return true; - } - let pending = Delivery::new( - through, - EventCompletion { - journal: self.journal.clone(), - route: self.route.clone(), - through, - }, - ); - match pending.into_completed().await { - Ok(through) => *acknowledged_through = through, - Err(failed) => self.set_error(format!( - "delivery checkpoint at {} failed: {}", - failed.delivery.value(), - failed.error - )), - } - true - } - - fn refresh_permalink(&self, sink: &dyn crate::sink::Sink) { - if let Some(link) = sink.permalink() { - *self.permalink.lock() = Some(link); - } - } - - fn set_error(&self, msg: String) { - tracing::warn!(session_id = %self.session_id, "{msg}"); - *self.last_error.lock() = Some(msg); - } -} - pub(crate) fn is_tool_lifecycle_event(event: &str) -> bool { matches!( event, @@ -983,23 +528,6 @@ pub(crate) fn is_tool_lifecycle_event(event: &str) -> bool { ) } -struct EventCompletion { - journal: Arc>, - route: SessionRoute, - through: u64, -} - -#[async_trait::async_trait] -impl Completion for EventCompletion { - async fn complete(&mut self) -> anyhow::Result<()> { - self.journal - .lock() - .await - .append_delivery_checkpoint(&self.route, self.through) - .await - } -} - #[cfg(test)] mod tests { use super::*; @@ -1044,6 +572,8 @@ mod tests { tx, counters: Arc::new(Counters::default()), last_error: Arc::new(Mutex::new(None)), + source_error: Arc::new(Mutex::new(None)), + source_progress: Arc::new(crate::pipeline::translation::Progress::default()), permalink: Arc::new(Mutex::new(None)), last_activity: Mutex::new(Instant::now()), }; diff --git a/bt-daemon/src/journal.rs b/bt-daemon/src/journal.rs index ced128d2..7af6c6a9 100644 --- a/bt-daemon/src/journal.rs +++ b/bt-daemon/src/journal.rs @@ -246,9 +246,7 @@ impl JournalWriter { self.position } - /// Append one event in redacted form and flush to the OS. Not fsync'd per - /// event (that would dominate hook latency); an OS crash can lose the last - /// few lines, which replay tolerates. + /// Append one redacted event and confirm its durable write before capture acknowledges it. /// /// The journal is a durability record and is never capped or truncated: /// dropping entries would silently cost recovery fidelity. Its size is @@ -262,28 +260,24 @@ impl JournalWriter { line.push(b'\n'); self.file.write_all(&line).await?; self.file.flush().await?; + self.file.sync_data().await?; self.position += line.len() as u64; Ok(self.position) } - /// Record that the sink durably accepted every event through this byte - /// offset for one route. It lives in the event journal so a cold worker - /// restores delivery knowledge together with translator state. - pub async fn append_delivery_checkpoint( + pub(crate) async fn append_control( &mut self, - route: &SessionRoute, - through: u64, - ) -> anyhow::Result<()> { - let mut line = serde_json::to_vec(&DeliveryCheckpointRecord { - record_type: "delivery_checkpoint".into(), - route: route.clone(), - through, - })?; + control: &PipelineControl, + ) -> anyhow::Result { + let mut value = serde_json::to_value(control)?; + value["_bt_record_type"] = "pipeline_control".into(); + let mut line = serde_json::to_vec(&value)?; line.push(b'\n'); self.file.write_all(&line).await?; self.file.flush().await?; + self.file.sync_data().await?; self.position += line.len() as u64; - Ok(()) + Ok(self.position) } } @@ -301,9 +295,24 @@ struct DeliveryCheckpointRecord { #[allow(clippy::large_enum_variant)] pub enum JournalRecord { Event(RedactedEnvelope), + Control(PipelineControl), DeliveryCheckpoint { route: SessionRoute, through: u64 }, } +#[derive(Debug, Clone, Copy, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ControlKind { + Checkpoint, + Finalize, + Retire, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PipelineControl { + pub kind: ControlKind, + pub bounds: std::collections::HashMap, +} + #[derive(Debug)] pub struct JournalRecordEntry { pub record: JournalRecord, @@ -400,6 +409,12 @@ impl JournalReader { route: checkpoint.route, through: checkpoint.through, } + } else if value + .get("_bt_record_type") + .and_then(serde_json::Value::as_str) + == Some("pipeline_control") + { + JournalRecord::Control(serde_json::from_value(value)?) } else { let mut event = serde_json::from_value(value).map_err(|error| { anyhow::anyhow!("journal {}:{}: {error}", self.path.display(), self.line_no) @@ -416,6 +431,13 @@ impl JournalReader { } } + pub(crate) fn extend(&mut self, through: u64) { + let buffered = self.reader.buffer().len() as u64; + self.reader + .get_mut() + .set_limit(through.saturating_sub(self.position + buffered)); + } + /// The latest acknowledged event offset for `route` in the portion of the /// journal being recovered. Older journals contain no checkpoints and /// therefore retain their existing replay behavior. @@ -565,7 +587,19 @@ pub async fn gc_old_journals(data_dir: &Path, max_age: std::time::Duration) { .and_then(|m| m.modified().ok()) .and_then(|modified| now.duration_since(modified).ok()) .is_some_and(|age| age > max_age); - if old { + let partition = path + .file_stem() + .and_then(|name| name.to_str()) + .and_then(|name| name.rsplit("--").next()); + let protected = match partition { + Some(id) => { + tokio::fs::try_exists(data_dir.join("pipeline").join(format!("{id}.sqlite"))) + .await + .unwrap_or(true) + } + None => false, + }; + if old && !protected { if let Err(e) = tokio::fs::remove_file(&path).await { tracing::warn!(path = %path.display(), "failed to remove stale journal: {e}"); } diff --git a/bt-daemon/src/pipeline/consumer.rs b/bt-daemon/src/pipeline/consumer.rs new file mode 100644 index 00000000..e09b816c --- /dev/null +++ b/bt-daemon/src/pipeline/consumer.rs @@ -0,0 +1,214 @@ +//! A routing subscription owns pending work independently of the source producer. +use super::{ + journal::{Cursor, SpanJournal}, + routing::RoutingSink, + Delivery, +}; +use crate::{ + delivery_ledger::LedgerSink, + dispatch::Counters, + sink::{Sink, SinkFactory}, + wire::SessionConfig, +}; +use std::{ + path::PathBuf, + sync::{atomic::Ordering, Arc, Mutex}, +}; +use tokio::sync::{watch, Notify}; + +pub(crate) struct Consumer { + task: Mutex>>, + config: watch::Sender, + progress: watch::Receiver, + stop: Arc, + error: Arc>>, +} + +pub(crate) struct ConsumerOptions { + pub source: String, + pub session_id: String, + pub version: String, + pub data_dir: PathBuf, + pub config: SessionConfig, + pub counters: Arc, + pub error: Arc>>, + pub permalink: Arc>>, + pub factory: Arc, +} + +impl Consumer { + pub fn spawn(journal: SpanJournal, mut cursor: Cursor, options: ConsumerOptions) -> Arc { + let (config, mut settings) = watch::channel(options.config.clone()); + let (progress, receive) = watch::channel(cursor.position()); + let stop = Arc::new(Notify::new()); + let handle = Arc::new(Self { + task: Mutex::new(None), + config, + progress: receive, + stop: stop.clone(), + error: options.error.clone(), + }); + let task = tokio::spawn(async move { + let task = async { + let sink = options + .factory + .create(&options.session_id, &options.source)?; + let sink = Box::new( + LedgerSink::new( + sink, + Some(&options.data_dir), + &options.source, + &options.session_id, + Some(&options.config), + &options.version, + ) + .await, + ); + let mut sink = RoutingSink::spawn( + sink, + options.source.clone(), + options.session_id.clone(), + options.data_dir.clone(), + options.error.clone(), + ); + let mut held = false; + if let Some(failure) = journal.failure(&cursor.route).await? { + *options.error.lock().unwrap() = Some(failure); + } + let mut completion: Option> = None; + loop { + if held { + tokio::select! { + _=stop.notified()=>return Ok::<(),anyhow::Error>(()), + result=settings.changed()=> { + if result.is_err() { return Ok(()); } + if let Some(pending)=completion.take() { + match pending.into_completed().await { + Ok(())=> { cursor.rewind().await?; progress.send_replace(cursor.completed_position()); }, + Err(failure)=> { completion=Some(failure.delivery); continue; }, + } + } + held=false; + cursor.rewind().await?; + } + } + continue; + } + let tail = journal.tail().await?; + options.counters.queued.store( + tail.saturating_sub(cursor.completed_position()), + Ordering::Relaxed, + ); + let received = tokio::select! { + _=stop.notified()=>return Ok(()), + result=super::delivery::InputPort::receive(&mut cursor)=>result?, + result=settings.changed()=> { if result.is_err() { return Ok(()); } continue; }, + }; + match received { + Some(delivery) => { + sink.configure(&settings.borrow_and_update()); + let ops = delivery.value(); + let attempt = async { + let count = sink.emit(ops).await?; + sink.flush().await?; + Ok::<_, anyhow::Error>(count) + }; + let result = tokio::select! { _=stop.notified()=>return Ok(()),result=attempt=>result }; + match result { + Ok(count) => { + options + .counters + .spans_emitted + .fetch_add(count, Ordering::Relaxed); + *options.permalink.lock().unwrap() = sink.permalink(); + // The journal retains a deferred prefix. No payloads or token list + // need accumulate in memory while waiting for its releasing child. + drop(delivery); + if !sink.has_pending_delivery() { + match cursor.pending_prefix().into_completed().await { + Ok(()) => { + cursor.rewind().await?; + progress.send_replace(cursor.completed_position()); + *options.error.lock().unwrap() = None; + } + Err(failure) => { + *options.error.lock().unwrap() = Some(format!( + "route completion held: {}", + failure.error + )); + completion = Some(failure.delivery); + held = true; + progress.send_replace(cursor.completed_position()); + } + } + } + } + Err(error) => { + if let Some(failure) = + error.downcast_ref::() + { + if failure.report { + tracing::warn!(session_id=%options.session_id, %error, "route processing failed"); + } + options + .counters + .spans_emitted + .fetch_add(failure.emitted, Ordering::Relaxed); + } + *options.error.lock().unwrap() = + Some(format!("routing held: {error:#}")); + let failure = options.error.lock().unwrap().clone(); + journal.set_failure(&cursor.route, failure).await?; + held = true; + progress.send_replace(cursor.completed_position()); + } + } + } + None => tokio::select! { + _=stop.notified()=>return Ok(()), + _=cursor.changed()=>{}, + result=settings.changed()=>if result.is_err() { return Ok(()); }, + }, + } + } + }; + if let Err(error) = task.await { + *options.error.lock().unwrap() = Some(format!("routing journal held: {error:#}")); + progress.send_replace(cursor.completed_position()); + } + }); + *handle.task.lock().unwrap() = Some(task); + handle + } + pub fn configure(&self, config: SessionConfig) { + self.config.send_if_modified(|current| { + if serde_json::to_value(&*current).ok() == serde_json::to_value(&config).ok() { + return false; + } + *current = config; + true + }); + } + pub async fn shutdown(&self) { + self.stop.notify_one(); + let task = self.task.lock().unwrap().take(); + if let Some(task) = task { + let _ = task.await; + } + } + pub async fn through(&self, tail: u64) -> Result<(), String> { + let mut progress = self.progress.clone(); + loop { + if *progress.borrow_and_update() >= tail { + return Ok(()); + } + if let Some(error) = self.error.lock().unwrap().clone() { + return Err(error); + } + progress + .changed() + .await + .map_err(|_| "routing actor is gone".to_owned())?; + } + } +} diff --git a/bt-daemon/src/pipeline/delivery.rs b/bt-daemon/src/pipeline/delivery.rs index da9371cf..41a48e69 100644 --- a/bt-daemon/src/pipeline/delivery.rs +++ b/bt-daemon/src/pipeline/delivery.rs @@ -14,7 +14,6 @@ pub(crate) trait Completion: Send { } pub(crate) struct Unfinished { - #[cfg_attr(not(test), allow(dead_code))] pub delivery: Delivery, pub error: anyhow::Error, } @@ -55,7 +54,7 @@ impl Delivery { #[async_trait::async_trait] pub(crate) trait InputPort: Send { type Item: Send; - async fn receive(&mut self) -> Option>; + async fn receive(&mut self) -> anyhow::Result>>; } // async_trait adds a redundant must_use annotation to its boxed futures. @@ -69,8 +68,8 @@ pub(crate) trait OutputPort: Send { #[async_trait::async_trait] impl InputPort for mpsc::Receiver> { type Item = T; - async fn receive(&mut self) -> Option> { - self.recv().await + async fn receive(&mut self) -> anyhow::Result>> { + Ok(self.recv().await) } } diff --git a/bt-daemon/src/pipeline/journal.rs b/bt-daemon/src/pipeline/journal.rs index 7b48ab39..eadf704e 100644 --- a/bt-daemon/src/pipeline/journal.rs +++ b/bt-daemon/src/pipeline/journal.rs @@ -1,9 +1,6 @@ //! Concrete durable channel: one ordered SpanOp log and a head per route. use super::{Completion, Delivery}; -use crate::{ - translate::{OriginSnapshot, SpanOp}, - wire::SessionRoute, -}; +use crate::{translate::SpanOp, wire::SessionRoute}; use rusqlite::{params, Connection, OptionalExtension}; use serde::{Deserialize, Serialize}; use std::{ @@ -65,11 +62,12 @@ impl SpanJournal { let db = Connection::open(path)?; db.busy_timeout(std::time::Duration::from_secs(5))?; let version: u32 = db.query_row("PRAGMA user_version", [], |row| row.get(0))?; - anyhow::ensure!(version <= 1, "unsupported SpanOp journal schema {version}"); + anyhow::ensure!(version <= 2, "unsupported SpanOp journal schema {version}"); db.execute_batch( - "PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL; + "PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL; BEGIN IMMEDIATE; CREATE TABLE IF NOT EXISTS batches ( - input INTEGER PRIMARY KEY, committed INTEGER NOT NULL DEFAULT 0); + input INTEGER PRIMARY KEY, committed INTEGER NOT NULL DEFAULT 0, + observation INTEGER NOT NULL DEFAULT 1); CREATE TABLE IF NOT EXISTS ops ( position INTEGER PRIMARY KEY AUTOINCREMENT, input INTEGER NOT NULL, value BLOB NOT NULL); @@ -77,8 +75,14 @@ impl SpanJournal { CREATE TABLE IF NOT EXISTS heads ( route TEXT PRIMARY KEY, position INTEGER NOT NULL DEFAULT 0, failure TEXT); - PRAGMA user_version=1;", + ", )?; + if version == 1 { + db.execute_batch( + "ALTER TABLE batches ADD COLUMN observation INTEGER NOT NULL DEFAULT 1", + )?; + } + db.execute_batch("PRAGMA user_version=2; COMMIT")?; Ok(db) }) .await??; @@ -115,7 +119,7 @@ impl SpanJournal { } /// Restart abandons only invisible chunks; a committed identity is never rewritten. - pub async fn begin(&self, input: u64) -> anyhow::Result { + pub async fn begin(&self, input: u64, observation: bool) -> anyhow::Result { self.with_db(move |db| { let tx = db.transaction()?; let done: Option = tx @@ -130,8 +134,8 @@ impl SpanJournal { } tx.execute("DELETE FROM ops WHERE input=?", [input])?; tx.execute( - "INSERT OR REPLACE INTO batches(input,committed) VALUES (?,0)", - [input], + "INSERT OR REPLACE INTO batches(input,committed,observation) VALUES (?,0,?)", + params![input, observation], )?; tx.commit()?; Ok(true) @@ -174,6 +178,15 @@ impl SpanJournal { |r| r.get(0), )?; anyhow::ensure!(input >= previous, "translation commits must be ordered"); + let earlier: u64 = tx.query_row( + "SELECT count(*) FROM batches WHERE committed=0 AND input anyhow::Result> { + self.with_db(move |db| { + let mut statement = db.prepare("SELECT position,value FROM ops JOIN batches USING(input) WHERE committed=1 AND observation=1 AND position>? ORDER BY position LIMIT ?")?; + let rows = statement.query_map(params![after,limit as u64], |r| Ok((r.get::<_,u64>(0)?,r.get::<_,Vec>(1)?)))?; + rows.map(|row| { let (position,bytes)=row?; Ok((position,StoredOp::decode(&bytes)?)) }).collect() + }).await + } + pub async fn tail(&self) -> anyhow::Result { self.with_db(|db| { Ok(db.query_row( @@ -242,6 +263,19 @@ impl SpanJournal { .await } + /// A completed subscription can remain dormant until its source has new work. + pub async fn route_complete(&self, route: &SessionRoute, input: u64) -> anyhow::Result { + let route = serde_json::to_string(route)?; + self.with_db(move |db| { + Ok(db.query_row( + "SELECT EXISTS(SELECT 1 FROM heads WHERE route=? AND failure IS NULL + AND position >= (SELECT coalesce(max(position),0) FROM ops JOIN batches USING(input) WHERE committed=1)) + AND (SELECT coalesce(max(input),0) FROM batches WHERE committed=1)>=?", + params![route,input], |row| row.get(0), + )?) + }).await + } + pub async fn failure(&self, route: &str) -> anyhow::Result> { let route = route.to_owned(); self.with_db(move |db| { @@ -314,6 +348,25 @@ impl Cursor { ))) } + pub fn position(&self) -> u64 { + self.read + } + pub fn completed_position(&self) -> u64 { + self.completed + } + /// Terminal handling can aggregate its fully handled contiguous read prefix. + pub fn pending_prefix(&self) -> Delivery<()> { + Delivery::new( + (), + HeadCompletion { + journal: self.journal.clone(), + route: self.route.clone(), + expected: self.completed, + through: self.read, + }, + ) + } + pub async fn rewind(&mut self) -> anyhow::Result<()> { let key = self.route.clone(); self.completed = self @@ -335,6 +388,19 @@ impl Cursor { } } +#[async_trait::async_trait] +impl super::delivery::InputPort for Cursor { + type Item = Vec; + async fn receive(&mut self) -> anyhow::Result>> { + loop { + if let Some(delivery) = self.next_batch(128).await? { + return Ok(Some(delivery)); + } + self.changed().await; + } + } +} + struct HeadCompletion { journal: SpanJournal, route: String, @@ -370,7 +436,7 @@ impl Completion for HeadCompletion { #[cfg(test)] mod tests { use super::*; - use crate::translate::SpanRow; + use crate::translate::{OriginSnapshot, SpanRow}; fn op(name: &str) -> SpanOp { SpanOp::Insert(SpanRow { @@ -394,12 +460,36 @@ mod tests { } } + #[tokio::test] + async fn preparatory_schema_upgrades_without_changing_committed_positions() { + let tmp = tempfile::tempdir().unwrap(); + let path = tmp.path().join("old.sqlite"); + let db = Connection::open(&path).unwrap(); + db.execute_batch("CREATE TABLE batches(input INTEGER PRIMARY KEY,committed INTEGER NOT NULL DEFAULT 0); INSERT INTO batches VALUES(10,1); PRAGMA user_version=1;").unwrap(); + drop(db); + let journal = SpanJournal::open(&path).await.unwrap(); + assert_eq!(journal.translated_through().await.unwrap(), 10); + journal.begin(20, false).await.unwrap(); + journal.append(20, &[op("retirement")]).await.unwrap(); + journal.commit(20).await.unwrap(); + assert!( + journal.read_after(0, 8).await.unwrap().is_empty(), + "retirement output is not a native correlation observation" + ); + let mut route = journal.subscribe(&route("a"), 0).await.unwrap(); + assert_eq!( + route.next_batch(8).await.unwrap().unwrap().value().len(), + 1, + "all published output still reaches durable delivery subscribers" + ); + } + #[tokio::test] async fn invisible_chunks_empty_commit_and_restart_preserve_the_boundary() { let tmp = tempfile::tempdir().unwrap(); let path = tmp.path().join("spans.sqlite"); let journal = SpanJournal::open(&path).await.unwrap(); - journal.begin(10).await.unwrap(); + journal.begin(10, true).await.unwrap(); journal.append(10, &[op("abandoned")]).await.unwrap(); let mut reader = journal.subscribe(&route("a"), 0).await.unwrap(); assert!(reader.next_batch(8).await.unwrap().is_none()); @@ -407,11 +497,11 @@ mod tests { drop(journal); let journal = SpanJournal::open(&path).await.unwrap(); assert_eq!(journal.translated_through().await.unwrap(), 0); - journal.begin(10).await.unwrap(); + journal.begin(10, true).await.unwrap(); journal.append(10, &[op("retained")]).await.unwrap(); journal.commit(10).await.unwrap(); - assert!(!journal.begin(10).await.unwrap()); - journal.begin(20).await.unwrap(); + assert!(!journal.begin(10, true).await.unwrap()); + journal.begin(20, true).await.unwrap(); journal.commit(20).await.unwrap(); assert_eq!(journal.translated_through().await.unwrap(), 20); let mut reader = journal.subscribe(&route("a"), 0).await.unwrap(); @@ -431,7 +521,7 @@ mod tests { let tmp = tempfile::tempdir().unwrap(); let path = tmp.path().join("spans.sqlite"); let journal = SpanJournal::open(&path).await.unwrap(); - journal.begin(10).await.unwrap(); + journal.begin(10, true).await.unwrap(); journal.append(10, &[op("one"), op("two")]).await.unwrap(); journal.commit(10).await.unwrap(); let mut a = journal.subscribe(&route("a"), 0).await.unwrap(); @@ -467,7 +557,7 @@ mod tests { .await .unwrap(); let mut reader = journal.subscribe(&route("a"), 0).await.unwrap(); - journal.begin(10).await.unwrap(); + journal.begin(10, true).await.unwrap(); journal.append(10, &[op("one")]).await.unwrap(); journal.commit(10).await.unwrap(); tokio::time::timeout(std::time::Duration::from_secs(1), reader.changed()) diff --git a/bt-daemon/src/pipeline/mod.rs b/bt-daemon/src/pipeline/mod.rs index e1e8579a..e8148719 100644 --- a/bt-daemon/src/pipeline/mod.rs +++ b/bt-daemon/src/pipeline/mod.rs @@ -17,6 +17,7 @@ pub(crate) trait Stage: Send { async fn process(&mut self, input: &Self::Input) -> Self::Outcome; } -// Storage is activated by the next stack layer after its crash contract is tested. -#[cfg(test)] -mod journal; +pub(crate) mod journal; +pub(crate) mod translation; + +pub(crate) mod consumer; diff --git a/bt-daemon/src/pipeline/routing.rs b/bt-daemon/src/pipeline/routing.rs index cb0a992c..daeb2040 100644 --- a/bt-daemon/src/pipeline/routing.rs +++ b/bt-daemon/src/pipeline/routing.rs @@ -69,7 +69,7 @@ impl RoutingSink { let (tx, mut rx) = mpsc::channel::>(16); tokio::spawn(async move { let mut prepare = super::prepare::Prepare::new(&source); - while let Some(delivery) = InputPort::receive(&mut rx).await { + while let Ok(Some(delivery)) = InputPort::receive(&mut rx).await { let request = delivery.value(); sink.configure(&request.config); sink.set_capture_versions( diff --git a/bt-daemon/src/pipeline/translation.rs b/bt-daemon/src/pipeline/translation.rs new file mode 100644 index 00000000..f3e2e994 --- /dev/null +++ b/bt-daemon/src/pipeline/translation.rs @@ -0,0 +1,501 @@ +//! One ordered source consumer. Output publication is independent of route delivery. +use super::journal::SpanJournal; +use super::prepare::Prepare; +use crate::{ + journal::{ControlKind, JournalReader, JournalRecord, JournalWriter, PipelineControl}, + translate::{AgentTranslator, OriginSnapshot, Registry, SessionCtx, SpanOp, SpanType}, + wire::{SessionConfig, SessionRoute}, +}; +use std::{ + collections::HashMap, + path::PathBuf, + sync::{ + atomic::{AtomicU64, AtomicUsize, Ordering}, + Arc, Mutex, + }, +}; +use tokio::sync::{mpsc, oneshot, Notify}; + +pub(crate) struct Observation { + pub key: String, + pub route: SessionRoute, + pub config: SessionConfig, +} + +enum Command { + Observe(Box), + Advance(u64), + Stop, + Barrier(oneshot::Sender>), + Control(ControlKind, oneshot::Sender>), +} + +#[derive(Default)] +pub(crate) struct Progress { + requested: AtomicU64, + committed: AtomicU64, +} +impl Progress { + pub fn pending(&self) -> u64 { + u64::from(self.requested.load(Ordering::Relaxed) > self.committed.load(Ordering::Relaxed)) + } +} + +pub(crate) struct Translation { + tx: mpsc::Sender, + task: Mutex>>, + pub journal: SpanJournal, + pub error: Arc>>, + pub routes: AtomicUsize, + pub progress: Arc, +} + +pub(crate) struct TranslationOptions { + pub source: String, + pub session_id: String, + pub namespace: String, + pub version: String, + pub data_dir: PathBuf, + pub events: Arc>, + pub event_path: PathBuf, + pub registry: Arc, + pub correlation: Arc, + pub correlation_changed: Arc, +} + +impl Translation { + pub async fn spawn(options: TranslationOptions) -> anyhow::Result> { + let path = options.data_dir.join("pipeline").join(format!( + "{}.sqlite", + crate::ids::session_storage_id(&options.source, &options.session_id) + )); + let journal = SpanJournal::open(&path).await?; + let committed = journal.translated_through().await?; + let progress = Arc::new(Progress { + requested: AtomicU64::new(committed), + committed: AtomicU64::new(committed), + }); + let error = Arc::new(Mutex::new(None)); + let (tx, rx) = mpsc::channel(64); + let actor = Producer { + options, + progress: progress.clone(), + journal: journal.clone(), + error: error.clone(), + translator: None, + reader: None, + observers: Vec::new(), + origins: HashMap::new(), + bounds: HashMap::new(), + position: 0, + committed, + held: false, + }; + let task = tokio::spawn(actor.run(rx)); + Ok(Arc::new(Self { + tx, + task: Mutex::new(Some(task)), + journal, + error, + routes: AtomicUsize::new(0), + progress, + })) + } + pub async fn shutdown(&self) { + let _ = self.tx.send(Command::Stop).await; + let task = self.task.lock().unwrap().take(); + if let Some(task) = task { + let _ = task.await; + } + } + pub async fn observe(&self, observer: Observation) -> anyhow::Result<()> { + self.routes.fetch_add(1, Ordering::Relaxed); + self.tx + .send(Command::Observe(Box::new(observer))) + .await + .map_err(|_| anyhow::anyhow!("translation actor is gone")) + } + pub async fn advance(&self, through: u64) -> anyhow::Result<()> { + self.progress + .requested + .fetch_max(through, Ordering::Relaxed); + self.tx + .send(Command::Advance(through)) + .await + .map_err(|_| anyhow::anyhow!("translation actor is gone")) + } + pub async fn barrier(&self) -> Result { + let (tx, rx) = oneshot::channel(); + self.tx + .send(Command::Barrier(tx)) + .await + .map_err(|_| "translation actor is gone".to_owned())?; + rx.await + .map_err(|_| "translation barrier lost".to_owned())? + } + pub async fn control(&self, kind: ControlKind) -> Result { + let (tx, rx) = oneshot::channel(); + self.tx + .send(Command::Control(kind, tx)) + .await + .map_err(|_| "translation actor is gone".to_owned())?; + rx.await + .map_err(|_| "translation control lost".to_owned())? + } +} + +struct Producer { + options: TranslationOptions, + progress: Arc, + journal: SpanJournal, + error: Arc>>, + translator: Option>, + reader: Option, + observers: Vec, + origins: HashMap, + bounds: HashMap, + position: u64, + committed: u64, + held: bool, +} + +impl Producer { + async fn run(mut self, mut rx: mpsc::Receiver) { + while let Some(command) = rx.recv().await { + let result = match command { + Command::Stop => break, + Command::Observe(observer) => { + let result = self.observe_history(&observer).await; + self.observers.push(*observer); + result + } + Command::Advance(through) => self.advance(through).await, + Command::Barrier(reply) => { + let _ = reply.send(self.boundary().await); + Ok(()) + } + Command::Control(kind, reply) => { + let result = self.control(kind).await; + let _ = reply.send(result.as_ref().copied().map_err(ToString::to_string)); + result.map(|_| ()) + } + }; + if let Err(error) = result { + self.held = true; + *self.error.lock().unwrap() = + Some(format!("translation held at {}: {error:#}", self.position)); + } + } + } + + async fn boundary(&mut self) -> Result { + if self.held { + return Err(self + .error + .lock() + .unwrap() + .clone() + .unwrap_or_else(|| "translation held".into())); + } + self.journal.tail().await.map_err(|error| error.to_string()) + } + + async fn control(&mut self, kind: ControlKind) -> anyhow::Result { + anyhow::ensure!(!self.held, "translation is held"); + let mut bounds = HashMap::new(); + for path in self.bounds.keys() { + let through = tokio::fs::metadata(path) + .await + .map(|m| m.len()) + .unwrap_or(0); + bounds.insert(path.clone(), through); + } + let control = PipelineControl { kind, bounds }; + let through = self + .options + .events + .lock() + .await + .append_control(&control) + .await?; + self.progress + .requested + .fetch_max(through, Ordering::Relaxed); + self.advance(through).await?; + self.journal.tail().await + } + + async fn initialize(&mut self) -> anyhow::Result<()> { + if self.translator.is_some() { + return Ok(()); + } + self.committed = self.journal.translated_through().await?; + // Load creation provenance so an upgrade cannot restamp old spans. + let mut after = 0; + loop { + let rows = self.journal.read_after(after, 128).await?; + if rows.is_empty() { + break; + } + for (position, op) in rows { + after = position; + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = op; + if let Some(origin) = row.origin { + self.origins.entry(row.span_id).or_insert(origin); + } + } + } + self.translator = Some( + self.options + .registry + .create_checked_with_session_namespace( + &self.options.source, + &self.options.namespace, + )?, + ); + self.reader = JournalReader::open(&self.options.event_path, self.committed).await?; + Ok(()) + } + + async fn advance(&mut self, through: u64) -> anyhow::Result<()> { + anyhow::ensure!(!self.held, "translation is held"); + if self.translator.is_none() && through <= self.committed { + return Ok(()); + } + self.initialize().await?; + let through = through.max(self.committed); + if self.position >= through { + return Ok(()); + } + let reader = self + .reader + .as_mut() + .ok_or_else(|| anyhow::anyhow!("event journal is missing"))?; + reader.extend(through); + while let Some(entry) = self + .reader + .as_mut() + .expect("reader initialized") + .next_record() + .await? + { + let input = entry.through; + let observation = !matches!(&entry.record,JournalRecord::Control(control) if matches!(control.kind,ControlKind::Retire)); + let ctx = SessionCtx { + session_id: self.options.session_id.clone(), + config: None, + }; + let (first, capture) = match entry.record { + JournalRecord::Event(event) => { + let mut env = crate::journal::envelope_from_redacted(event); + if crate::translate::canonical_source_name(&env.source) != self.options.source { + self.position = input; + continue; + } + if env.source == "claude-code" { + if let Some(payload) = env.payload.as_object_mut() { + payload.insert("_bt_transcript_replay".into(), true.into()); + } + } + if let Some(path) = env + .payload + .pointer("/_bt_transcript_mirror/mirror") + .and_then(|v| v.as_str()) + .or_else(|| env.payload.get("transcript_path").and_then(|v| v.as_str())) + .or_else(|| { + env.payload + .get("agent_transcript_path") + .and_then(|v| v.as_str()) + }) + { + self.bounds.entry(path.into()).or_insert(0); + } + let capture = OriginSnapshot { + plugin_version: env.plugin_version.clone(), + source_version: env.source_version.clone(), + bt_version: self.options.version.clone(), + }; + ( + self.translator + .as_mut() + .expect("translator initialized") + .handle(&env, &ctx)?, + Some(capture), + ) + } + JournalRecord::Control(control) => { + // Retirement exports defensive terminal updates but does not + // close native state when rebuilding for later captured work. + if input <= self.committed && matches!(control.kind, ControlKind::Retire) { + self.position = input; + continue; + } + let translator = self.translator.as_mut().expect("translator initialized"); + let ops = if !matches!(control.kind, ControlKind::Checkpoint) { + translator.finalize_at(&ctx, &control.bounds)? + } else { + translator.checkpoint_at(&ctx, &control.bounds)? + }; + (ops, None) + } + JournalRecord::DeliveryCheckpoint { .. } => { + if input > self.committed { + self.journal.begin(input, observation).await?; + self.journal.commit(input).await?; + self.committed = input; + self.progress.committed.store(input, Ordering::Relaxed); + } + self.position = input; + continue; + } + }; + let publish = input > self.committed; + if publish { + self.journal.begin(input, observation).await?; + } + let mut next = Some(first); + while let Some(mut ops) = next { + for op in &mut ops { + let insert = matches!(op, SpanOp::Insert(_)); + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = op; + if insert + && (row.is_turn + || (row.span_type == SpanType::Task && row.parent_span_ids.is_empty())) + { + if let Some(capture) = &capture { + self.origins + .entry(row.span_id.clone()) + .or_insert_with(|| capture.clone()); + } + } + row.origin = self.origins.get(&row.span_id).cloned(); + } + if observation { + for observer in &self.observers { + Self::observe(&self.options, &ops, observer).await?; + } + } + if publish { + self.journal.append(input, &ops).await?; + } + next = self + .translator + .as_mut() + .expect("translator initialized") + .drain_pending(&ctx)?; + } + if publish { + self.journal.commit(input).await?; + self.committed = input; + self.progress.committed.store(input, Ordering::Relaxed); + } + self.position = input; + } + Ok(()) + } + + async fn observe_history(&mut self, observer: &Observation) -> anyhow::Result<()> { + let mut after = 0; + loop { + let rows = self.journal.read_after(after, 128).await?; + if rows.is_empty() { + break; + } + after = rows.last().expect("nonempty").0; + let ops = rows.into_iter().map(|(_, op)| op).collect::>(); + Self::observe(&self.options, &ops, observer).await?; + } + Ok(()) + } + async fn observe( + options: &TranslationOptions, + ops: &[SpanOp], + observer: &Observation, + ) -> anyhow::Result<()> { + let mut ops = ops.to_vec(); + Prepare::new(&options.source).apply(&mut ops, &observer.config); + if options + .correlation + .observe_ops(&observer.key, &observer.route, &observer.config, &ops) + { + crate::server::persist_active_parent_snapshot( + &options.data_dir, + &observer.key, + &options.correlation, + ) + .await + .map_err(anyhow::Error::msg)?; + options.correlation_changed.notify_one(); + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{pipeline::delivery::InputPort, translate::SpanRow}; + + #[tokio::test] + async fn routing_resume_does_not_read_or_translate_the_committed_event_prefix() { + let tmp = tempfile::tempdir().unwrap(); + let source = "debug"; + let session = "stored"; + let event_path = crate::journal::source_journal_path(tmp.path(), source, session); + std::fs::create_dir_all(event_path.parent().unwrap()).unwrap(); + // Any translation replay of this file would fail immediately. + std::fs::write(&event_path, b"not JSON\n").unwrap(); + let input = std::fs::metadata(&event_path).unwrap().len(); + let journal = SpanJournal::open(&tmp.path().join("pipeline").join(format!( + "{}.sqlite", + crate::ids::session_storage_id(source, session) + ))) + .await + .unwrap(); + journal.begin(input, true).await.unwrap(); + journal + .append( + input, + &[SpanOp::Insert(SpanRow { + span_id: "durable".into(), + root_span_id: "durable".into(), + name: "already translated".into(), + ..SpanRow::default() + })], + ) + .await + .unwrap(); + let tail = journal.commit(input).await.unwrap(); + let translation = Translation::spawn(TranslationOptions { + source: source.into(), + session_id: session.into(), + namespace: crate::ids::session_namespace(source, session), + version: "test".into(), + data_dir: tmp.path().into(), + events: Arc::new(tokio::sync::Mutex::new( + JournalWriter::open_path(&event_path).await.unwrap(), + )), + event_path, + registry: Arc::new(Registry::default_agents()), + correlation: Arc::new(crate::correlation::CorrelationRegistry::default()), + correlation_changed: Arc::new(Notify::new()), + }) + .await + .unwrap(); + translation.advance(input).await.unwrap(); + assert_eq!(translation.barrier().await.unwrap(), tail); + assert_eq!(translation.progress.pending(), 0); + assert!(translation.error.lock().unwrap().is_none()); + let mut route = translation + .journal + .subscribe(&SessionRoute::default(), 0) + .await + .unwrap(); + let delivery = InputPort::receive(&mut route).await.unwrap().unwrap(); + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = &delivery.value()[0]; + assert_eq!(row.name, "already translated"); + delivery.into_completed().await.ok().unwrap(); + route.rewind().await.unwrap(); + assert_eq!(route.completed_position(), tail); + } +} diff --git a/bt-daemon/src/server/ingress.rs b/bt-daemon/src/server/ingress.rs index f6727a49..7f16cbd5 100644 --- a/bt-daemon/src/server/ingress.rs +++ b/bt-daemon/src/server/ingress.rs @@ -7,6 +7,7 @@ use super::*; pub(super) enum IngressMsg { Event(Box), Barrier(oneshot::Sender<()>), + Stop(oneshot::Sender<()>), } /// The hook path never waits for this queue. Once it fills, the append-only @@ -24,6 +25,10 @@ pub(super) fn spawn_ingress_worker(daemon: Arc, mut rx: mpsc::Receiver { + let _ = reply.send(()); + break; + } IngressMsg::Event(event) => { dispatch_ingress_event(&daemon, *event).await; } diff --git a/bt-daemon/src/server/maintenance.rs b/bt-daemon/src/server/maintenance.rs index 4f222501..837cc5e3 100644 --- a/bt-daemon/src/server/maintenance.rs +++ b/bt-daemon/src/server/maintenance.rs @@ -12,7 +12,17 @@ pub(super) const GC_INTERVAL: Duration = Duration::from_secs(60 * 60); pub(super) async fn collect_garbage(data_dir: &std::path::Path) { journal::gc_old_journals(data_dir, RETENTION).await; journal::gc_old_managed_runs(data_dir, RETENTION).await; - crate::transcript_mirror::gc_old_mirrors(data_dir, RETENTION).await; + // Durable partitions may still need any recorded transcript prefix. + let mut partitions = tokio::fs::read_dir(data_dir.join("pipeline")).await.ok(); + let mut retained = false; + if let Some(entries) = partitions.as_mut() { + while let Ok(Some(entry)) = entries.next_entry().await { + retained |= entry.path().extension().and_then(|ext| ext.to_str()) == Some("sqlite"); + } + } + if !retained { + crate::transcript_mirror::gc_old_mirrors(data_dir, RETENTION).await; + } gc_old_correlation_states(data_dir, RETENTION).await; } @@ -113,10 +123,43 @@ pub(super) async fn drain_all(daemon: &Arc) { if let Err(error) = resolve_pending_sessions(daemon, |_| true).await { tracing::warn!(%error, "pending correlation events retained for recovery during shutdown"); } - let sessions: Vec> = daemon.sessions.lock().values().cloned().collect(); + // Publish shared retirement output before stopping any route subscriber. + let sources: Vec<_> = daemon + .translations + .lock() + .unwrap() + .values() + .cloned() + .collect(); + for source in sources { + let _ = source.control(crate::journal::ControlKind::Retire).await; + } + let sessions: Vec> = daemon.sessions.lock().unwrap().values().cloned().collect(); for s in sessions { s.shutdown().await; } + let sources: Vec<_> = daemon + .translations + .lock() + .unwrap() + .values() + .cloned() + .collect(); + for source in sources { + source.shutdown().await; + } + daemon.sessions.lock().unwrap().clear(); + daemon.translations.lock().unwrap().clear(); + daemon.journals.lock().unwrap().clear(); + let (reply, done) = oneshot::channel(); + if daemon + .ingress_tx + .send(super::ingress::IngressMsg::Stop(reply)) + .await + .is_ok() + { + let _ = done.await; + } *drained = true; } diff --git a/bt-daemon/src/server/mod.rs b/bt-daemon/src/server/mod.rs index 5e2b7652..f29dbf59 100644 --- a/bt-daemon/src/server/mod.rs +++ b/bt-daemon/src/server/mod.rs @@ -120,6 +120,7 @@ pub struct Daemon { managed_run_sessions: Mutex>>, auth_errors: Mutex>, sessions: Mutex>>, + translations: Mutex>>, correlation: Arc, automatic_links: Mutex>, standalone_links: Mutex>, @@ -159,6 +160,7 @@ impl Daemon { managed_run_sessions: Mutex::new(HashMap::new()), auth_errors: Mutex::new(HashMap::new()), sessions: Mutex::new(HashMap::new()), + translations: Mutex::new(HashMap::new()), correlation: Arc::new(crate::correlation::CorrelationRegistry::default()), automatic_links: Mutex::new(HashMap::new()), standalone_links: Mutex::new(HashSet::new()), @@ -193,7 +195,7 @@ impl Daemon { .sessions .lock() .values() - .map(|s| s.counters.queued.load(Ordering::Relaxed)) + .map(|s| s.pending()) .sum(); dispatched.saturating_add( self.pending_sessions diff --git a/bt-daemon/src/server/recovery.rs b/bt-daemon/src/server/recovery.rs index e32a594f..2eb6d4c2 100644 --- a/bt-daemon/src/server/recovery.rs +++ b/bt-daemon/src/server/recovery.rs @@ -64,6 +64,7 @@ pub(super) async fn recover_unprocessed_journals(daemon: &Arc) { ); } } + journal::JournalRecord::Control(_) => {} journal::JournalRecord::DeliveryCheckpoint { route, through } => { let key = serde_json::to_string(&route).unwrap_or_default(); let acknowledged = acknowledged_by_route.entry(key).or_default(); @@ -72,7 +73,7 @@ pub(super) async fn recover_unprocessed_journals(daemon: &Arc) { } before = through; } - for event in latest_by_route.into_values() { + for mut event in latest_by_route.into_values() { if pending.contains(&( event.env.source.clone(), event.env.session_id.clone(), @@ -80,12 +81,33 @@ pub(super) async fn recover_unprocessed_journals(daemon: &Arc) { )) { continue; } - let route = recovered_delivery_route(daemon, &event.env) - .await + let recovered_route = recovered_delivery_route(daemon, &event.env).await; + let route = recovered_route .as_ref() .and_then(|route| serde_json::to_string(route).ok()) .unwrap_or_default(); - if acknowledged_by_route.get(&route).copied().unwrap_or(0) < event.journal_through { + let span_path = daemon.data_dir.join("pipeline").join(format!( + "{}.sqlite", + crate::ids::session_storage_id(&event.env.source, &event.env.session_id) + )); + let completed = if tokio::fs::try_exists(&span_path).await.unwrap_or(false) { + match ( + crate::pipeline::journal::SpanJournal::open(&span_path).await, + recovered_route.as_ref(), + ) { + (Ok(journal), Some(route)) => journal + .route_complete(route, recorded_len) + .await + .unwrap_or(false), + _ => false, + } + } else { + acknowledged_by_route.get(&route).copied().unwrap_or(0) >= event.journal_through + }; + if !completed { + // Ordered controls after the final native event also belong to recovery. + event.replay_through = recorded_len; + event.journal_through = recorded_len; candidates.push(event); } else if event.env.source == "claude-code" { // A fully delivered historical session has no actor to retire diff --git a/bt-daemon/src/server/rpc.rs b/bt-daemon/src/server/rpc.rs index 78378c10..2bf7cb0c 100644 --- a/bt-daemon/src/server/rpc.rs +++ b/bt-daemon/src/server/rpc.rs @@ -334,10 +334,15 @@ impl Daemon { session_id: key.session_id.clone(), source: s.source.clone(), route: serde_json::from_str(&key.route).ok(), - queued: s.counters.queued.load(Ordering::Relaxed), + queued: s.pending(), spans_emitted: s.counters.spans_emitted.load(Ordering::Relaxed), - permalink: s.permalink.lock().clone(), - last_error: s.last_error.lock().clone(), + permalink: s.permalink.lock().unwrap().clone(), + last_error: s + .last_error + .lock() + .unwrap() + .clone() + .or_else(|| s.source_error.lock().unwrap().clone()), }) .collect(); for (key, (source, error)) in self.auth_errors.lock().iter() { diff --git a/bt-daemon/src/server/sessions.rs b/bt-daemon/src/server/sessions.rs index c965f159..4b52003a 100644 --- a/bt-daemon/src/server/sessions.rs +++ b/bt-daemon/src/server/sessions.rs @@ -75,30 +75,65 @@ impl Daemon { route, ) .await, - through: replay_through, journal_path, }; let journal = self .journal_writer_for(&env.source, &env.session_id) .await?; - let mut map = self.sessions.lock(); + let storage_key = crate::ids::session_namespace(&env.source, &env.session_id); + let existing = self.translations.lock().unwrap().get(&storage_key).cloned(); + let translation = match existing { + Some(translation) => translation, + None => { + let translation = crate::pipeline::translation::Translation::spawn( + crate::pipeline::translation::TranslationOptions { + source: env.source.clone(), + session_id: env.session_id.clone(), + namespace: translator_session_id.clone(), + version: self.version.clone(), + data_dir: self.data_dir.clone(), + events: journal.clone(), + event_path: replay.journal_path.clone(), + registry: self.translators.clone(), + correlation: self.correlation.clone(), + correlation_changed: self.correlation_changed.clone(), + }, + ) + .await?; + self.translations + .lock() + .unwrap() + .insert(storage_key, translation.clone()); + translation + } + }; + translation + .observe(crate::pipeline::translation::Observation { + key: key.correlation_key(), + route: route.clone(), + config: config.clone(), + }) + .await?; + // Materialize historical output before seeding a released completion boundary. + translation.advance(replay_through).await?; + let _ = translation.barrier().await; + let cursor = translation + .journal + .subscribe(route, replay.acknowledged_through) + .await?; + let mut map = self.sessions.lock().unwrap(); if let Some(s) = map.get(key) { return Ok(s.clone()); } let session = Session::spawn( SessionOptions { + translation, + cursor, session_id: env.session_id.clone(), - translator_session_id, source: env.source.clone(), bt_version: self.version.clone(), - replay: Some(replay), config, - correlation_key: key.correlation_key(), - route: route.clone(), - correlation: self.correlation.clone(), data_dir: self.data_dir.clone(), - journal, - correlation_changed: self.correlation_changed.clone(), }, self.translators.clone(), self.sink_factory.clone(), @@ -147,10 +182,14 @@ impl Daemon { // lock for the brief writer-map cleanup after daemon work ends. let capture_lock = self.session_lock(&key.source, &key.session_id); let _capture_guard = capture_lock.lock().await; - self.claude_transcripts.lock().retain(|candidate, _| { - candidate.source != key.source || candidate.session_id != key.session_id - }); - self.journals.lock().remove(&storage_key); + self.claude_transcripts + .lock() + .unwrap() + .retain(|candidate, _| { + candidate.source != key.source || candidate.session_id != key.session_id + }); + self.journals.lock().unwrap().remove(&storage_key); + self.translations.lock().unwrap().remove(&storage_key); { let mut locks = self.session_locks.lock(); // A waiting capture or observer must keep using this same @@ -173,10 +212,7 @@ impl Daemon { self.sessions .lock() .iter() - .filter(|(_, session)| { - session.idle_for() >= idle_timeout - && session.counters.queued.load(Ordering::Relaxed) == 0 - }) + .filter(|(_, session)| session.idle_for() >= idle_timeout && session.pending() == 0) .filter(|(key, _)| !self.correlation.has_active_tools(&key.correlation_key())) .map(|(key, _)| key.clone()) .collect() diff --git a/bt-daemon/src/span_processor.rs b/bt-daemon/src/span_processor.rs index 226e3276..55c33db2 100644 --- a/bt-daemon/src/span_processor.rs +++ b/bt-daemon/src/span_processor.rs @@ -281,6 +281,9 @@ pub fn process( }; // Turn identity belongs to the translator, not the JSON transform. candidate.is_turn = row.is_turn; + candidate.origin = row.origin.clone(); + candidate.turn_span_id = row.turn_span_id.clone(); + candidate.late_merge_key = row.late_merge_key.clone(); row = candidate; } Ok(()) diff --git a/bt-daemon/src/translate/codex.rs b/bt-daemon/src/translate/codex.rs index 8b654770..bc3d6b80 100644 --- a/bt-daemon/src/translate/codex.rs +++ b/bt-daemon/src/translate/codex.rs @@ -228,6 +228,7 @@ enum PendingWork { }, CatchUp { paths: Vec, + bounds: Option>, next_path: usize, finalize: bool, }, @@ -351,12 +352,19 @@ impl AgentTranslator for CodexTranslator { } PendingWork::CatchUp { paths, + bounds, mut next_path, finalize, } => { while next_path < paths.len() { let path = &paths[next_path]; - if self.catch_up_chunk(path, 0, None, None, &mut ops) { + if self.catch_up_chunk( + path, + 0, + None, + bounds.as_ref().and_then(|b| b.get(path).copied()), + &mut ops, + ) { if finalize { if let Some(mut scope) = self.scopes.remove(path) { self.close_dangling(&mut scope, None, &mut ops); @@ -370,6 +378,7 @@ impl AgentTranslator for CodexTranslator { if !ops.is_empty() || next_path < paths.len() { self.pending = (next_path < paths.len()).then_some(PendingWork::CatchUp { paths, + bounds, next_path, finalize, }); @@ -382,11 +391,25 @@ impl AgentTranslator for CodexTranslator { } fn checkpoint(&mut self, ctx: &SessionCtx) -> anyhow::Result> { - self.start_catch_up(ctx, false) + self.start_catch_up(ctx, false, None) } fn finalize(&mut self, ctx: &SessionCtx) -> anyhow::Result> { - self.start_catch_up(ctx, true) + self.start_catch_up(ctx, true, None) + } + fn checkpoint_at( + &mut self, + ctx: &SessionCtx, + bounds: &std::collections::HashMap, + ) -> anyhow::Result> { + self.start_catch_up(ctx, false, Some(bounds)) + } + fn finalize_at( + &mut self, + ctx: &SessionCtx, + bounds: &std::collections::HashMap, + ) -> anyhow::Result> { + self.start_catch_up(ctx, true, Some(bounds)) } } @@ -405,7 +428,12 @@ impl CodexTranslator { } } - fn start_catch_up(&mut self, ctx: &SessionCtx, finalize: bool) -> anyhow::Result> { + fn start_catch_up( + &mut self, + ctx: &SessionCtx, + finalize: bool, + bounds: Option<&std::collections::HashMap>, + ) -> anyhow::Result> { anyhow::ensure!( self.pending.is_none(), "Codex translator has pending catch-up work; drain it before checkpointing" @@ -418,6 +446,7 @@ impl CodexTranslator { } self.pending = Some(PendingWork::CatchUp { paths, + bounds: bounds.cloned(), next_path: 0, finalize, }); diff --git a/bt-daemon/src/translate/mod.rs b/bt-daemon/src/translate/mod.rs index 7f5bfd3e..5adbc61b 100644 --- a/bt-daemon/src/translate/mod.rs +++ b/bt-daemon/src/translate/mod.rs @@ -153,6 +153,22 @@ pub trait AgentTranslator: Send { Ok(Vec::new()) } + /// Ordered daemon controls bound external observations during replay. + fn checkpoint_at( + &mut self, + ctx: &SessionCtx, + _bounds: &HashMap, + ) -> anyhow::Result> { + self.checkpoint(ctx) + } + fn finalize_at( + &mut self, + ctx: &SessionCtx, + _bounds: &HashMap, + ) -> anyhow::Result> { + self.finalize(ctx) + } + /// Backward-compatible terminal flush used by transcript import callers. /// Live delivery barriers use [`Self::checkpoint`] instead. fn flush(&mut self, ctx: &SessionCtx) -> anyhow::Result> { diff --git a/bt-daemon/src/translate/turn_lineage.rs b/bt-daemon/src/translate/turn_lineage.rs index 41e35df0..50d466b3 100644 --- a/bt-daemon/src/translate/turn_lineage.rs +++ b/bt-daemon/src/translate/turn_lineage.rs @@ -65,6 +65,21 @@ impl AgentTranslator for TurnLineage { Ok(stamp(self.inner.finalize(ctx)?)) } + fn checkpoint_at( + &mut self, + ctx: &SessionCtx, + bounds: &std::collections::HashMap, + ) -> anyhow::Result> { + Ok(stamp(self.inner.checkpoint_at(ctx, bounds)?)) + } + fn finalize_at( + &mut self, + ctx: &SessionCtx, + bounds: &std::collections::HashMap, + ) -> anyhow::Result> { + Ok(stamp(self.inner.finalize_at(ctx, bounds)?)) + } + fn flush(&mut self, ctx: &SessionCtx) -> anyhow::Result> { Ok(stamp(self.inner.flush(ctx)?)) } diff --git a/bt-daemon/tests/cursor_pipeline.rs b/bt-daemon/tests/cursor_pipeline.rs index d0785fda..8e45b9ec 100644 --- a/bt-daemon/tests/cursor_pipeline.rs +++ b/bt-daemon/tests/cursor_pipeline.rs @@ -347,6 +347,15 @@ fn transcript(prompt: &str, response: &str) -> String { ) } +fn span_journal(data: &Path) -> rusqlite::Connection { + let path = std::fs::read_dir(data.join("pipeline")) + .unwrap() + .map(|entry| entry.unwrap().path()) + .find(|path| path.extension().and_then(|ext| ext.to_str()) == Some("sqlite")) + .unwrap(); + rusqlite::Connection::open(path).unwrap() +} + fn rows(path: &Path) -> Vec { std::fs::read_to_string(path) .unwrap() @@ -437,7 +446,7 @@ async fn late_response_is_delivered_and_checkpointed_without_another_hook() { let text = std::fs::read_to_string(&journal).unwrap_or_default(); let mut position = 0; let mut response_through = None; - let mut checkpoint_through = 0; + for line in text.split_inclusive('\n') { position += line.len() as u64; let Ok(row) = serde_json::from_str::(line) else { @@ -446,12 +455,11 @@ async fn late_response_is_delivered_and_checkpointed_without_another_hook() { if row["event"] == "afterAgentResponse" { response_through = Some(position); } - if row.get("_bt_record_type").is_some() { - checkpoint_through = - checkpoint_through.max(row["through"].as_u64().unwrap_or(0)); - } } - if delivered && response_through.is_some_and(|offset| checkpoint_through >= offset) { + let completed = delivered && response_through.is_some_and(|offset| { + span_journal(&data).query_row("SELECT min(position)>=(SELECT coalesce(max(position),0) FROM ops JOIN batches USING(input) WHERE committed=1) AND (SELECT coalesce(max(input),0) FROM batches WHERE committed=1)>=? FROM heads", [offset], |row|row.get::<_,bool>(0)).unwrap_or(false) + }); + if delivered && completed { break; } tokio::time::sleep(Duration::from_millis(10)).await; @@ -676,13 +684,10 @@ async fn cursor_mirror_generations_and_capture_bounds_survive_deleted_native_rec .unwrap() .contains("cursor-test-secret")); - // Simulate a delivery loss: retain accepted ingress, discard sink receipts. - let pending = persisted - .iter() - .filter(|row| row.get("_bt_record_type").is_none()) - .map(|row| format!("{row}\n")) - .collect::(); - std::fs::write(&journal, pending).unwrap(); + // Simulate unresolved sink delivery while retaining both durable source stores. + span_journal(&data) + .execute("UPDATE heads SET position=0,failure=NULL", []) + .unwrap(); std::fs::remove_dir_all(data.join("delivery-ledger")).unwrap(); std::fs::remove_file(&native).unwrap(); std::fs::remove_file(&output).unwrap(); diff --git a/bt-daemon/tests/distributed_tracing.rs b/bt-daemon/tests/distributed_tracing.rs index dc429672..fb7c9c6b 100644 --- a/bt-daemon/tests/distributed_tracing.rs +++ b/bt-daemon/tests/distributed_tracing.rs @@ -2915,6 +2915,21 @@ async fn pending_restart_preserves_current_candidates_and_decided_legacy_links() // Restore the accepted-but-undelivered checkpoint, as after an abrupt // process exit. Shutdown itself is covered separately above. tokio::fs::write(&journal_path, journal).await.unwrap(); + // This crash fixture rolls back source persistence as well as delivery. + let id = journal_path + .file_stem() + .unwrap() + .to_str() + .unwrap() + .rsplit("--") + .next() + .unwrap(); + for suffix in [".sqlite", ".sqlite-wal", ".sqlite-shm"] { + let path = data_dir.join("pipeline").join(format!("{id}{suffix}")); + if path.exists() { + tokio::fs::remove_file(path).await.unwrap(); + } + } // Graceful shutdown delivered the buffered spans. Roll back its local // delivery acknowledgements as well as the journal to model a crash. tokio::fs::remove_dir_all(data_dir.join("delivery-ledger")) diff --git a/bt-daemon/tests/pipeline.rs b/bt-daemon/tests/pipeline.rs index c284d3d8..f811eefd 100644 --- a/bt-daemon/tests/pipeline.rs +++ b/bt-daemon/tests/pipeline.rs @@ -375,6 +375,21 @@ async fn start_daemon_at_with( handle } +/// These fixtures contain one source partition. Read its durable progress directly. +fn pipeline_database(data_dir: &Path) -> rusqlite::Connection { + let path = std::fs::read_dir(data_dir.join("pipeline")) + .unwrap() + .map(|entry| entry.unwrap().path()) + .find(|path| path.extension().and_then(|ext| ext.to_str()) == Some("sqlite")) + .unwrap(); + rusqlite::Connection::open(path).unwrap() +} +fn head_and_tail(data_dir: &Path) -> (u64, u64) { + pipeline_database(data_dir).query_row( + "SELECT position,(SELECT coalesce(max(position),0) FROM ops JOIN batches USING(input) WHERE committed=1) FROM heads", + [],|row| Ok((row.get(0)?,row.get(1)?))).unwrap() +} + async fn wait_for(endpoint: &Path) { for _ in 0..200 { if let Ok(Some(_)) = run_status(StatusArgs { @@ -654,9 +669,12 @@ async fn rejected_flush_blocks_checkpoints_until_journal_replay() { wait_for(&socket).await; flush_session(session, &socket, 5_000).await.unwrap(); assert!(sink.emitted.load(Ordering::SeqCst) > emitted_before_restart); - assert!(std::fs::read_to_string(journal_path) - .unwrap() - .contains("delivery_checkpoint")); + let (head, tail) = head_and_tail(&data_dir); + assert!(tail > 0); + assert_eq!( + head, tail, + "restart completes the durable SpanOp subscription" + ); shutdown(&socket).await; handle.await.unwrap(); } @@ -845,7 +863,7 @@ async fn one_session_reports_to_multiple_routes_and_orgs_concurrently() { sink.emitted.load(std::sync::atomic::Ordering::Relaxed) > 0, "every destination must receive spans" ); - assert_eq!(sink.flushes.load(std::sync::atomic::Ordering::Relaxed), 1); + assert!(sink.flushes.load(std::sync::atomic::Ordering::Relaxed) >= 1); } let status = run_status(StatusArgs { @@ -1004,8 +1022,8 @@ async fn a_route_backfills_observations_captured_by_another_route_once() { ); assert_eq!( sinks[0].emitted.load(std::sync::atomic::Ordering::Relaxed), - 2, - "work receives the personal-route observation once, then its new Stop" + 1, + "the shared producer already delivered the personal observation to work; only the new Stop remains" ); shutdown(&socket).await; @@ -1104,11 +1122,17 @@ async fn a_new_destination_receives_history_once_then_uses_its_own_checkpoint() .await .unwrap(); let delivered = third_sinks.sinks.lock().unwrap().clone(); - assert_eq!(delivered.len(), 1); assert_eq!( - delivered[0] - .emitted - .load(std::sync::atomic::Ordering::Relaxed), + delivered.len(), + 2, + "pending subscriptions resume without a new native event for that route" + ); + let personal = delivered + .iter() + .find(|sink| sink.org.lock().unwrap().as_deref() == Some("personal-org")) + .unwrap(); + assert_eq!( + personal.emitted.load(std::sync::atomic::Ordering::Relaxed), 1, "personal's acknowledged history is not re-delivered" ); @@ -1188,7 +1212,7 @@ async fn events_are_ordered_journaled_and_emitted() { let events: Vec = jlines .iter() - .filter(|line| !line.contains("\"_bt_record_type\":\"delivery_checkpoint\"")) + .filter(|line| !line.contains("\"_bt_record_type\"")) .map(|l| { serde_json::from_str::(l).unwrap()["event"] .as_str() @@ -1240,7 +1264,7 @@ async fn session_ids_longer_than_a_file_name_are_journaled_and_emitted() { assert_eq!( jtext .lines() - .filter(|line| !line.contains("\"_bt_record_type\":\"delivery_checkpoint\"")) + .filter(|line| !line.contains("\"_bt_record_type\"")) .count(), 2 ); @@ -1800,14 +1824,29 @@ async fn cold_worker_rebuilds_acknowledged_journal_without_redelivery() { shutdown(&socket).await; first.await.unwrap(); - // A cold worker receives only the saved journal, not the first worker's - // daemon state. Its delivery checkpoint must travel with that journal. + // A cold worker receives durable files only: event history plus SpanOps and heads. let recovered_data_dir = tmp.path().join("recovered-data"); let source = source_journal_path(&data_dir, "debug", "resume"); let recovered_journal = source_journal_path(&recovered_data_dir, "debug", "resume"); std::fs::create_dir_all(recovered_journal.parent().unwrap()).unwrap(); std::fs::copy(source, &recovered_journal).unwrap(); + let db = pipeline_database(&data_dir); + db.execute_batch("PRAGMA wal_checkpoint(TRUNCATE)").unwrap(); + std::fs::create_dir_all(recovered_data_dir.join("pipeline")).unwrap(); + for entry in std::fs::read_dir(data_dir.join("pipeline")).unwrap() { + let path = entry.unwrap().path(); + if path.extension().and_then(|ext| ext.to_str()) == Some("sqlite") { + std::fs::copy( + &path, + recovered_data_dir + .join("pipeline") + .join(path.file_name().unwrap()), + ) + .unwrap(); + } + } + drop(db); let second = start_daemon_at(recovered_data_dir.clone(), socket.clone()).await; forward_envelope(&envelope("resume", "Stop", 2), &socket, &host, false) .await @@ -1816,9 +1855,12 @@ async fn cold_worker_rebuilds_acknowledged_journal_without_redelivery() { let journal = std::fs::read_to_string(recovered_journal).unwrap(); assert_eq!( - journal.lines().count(), - 4, - "two events plus two delivery checkpoints" + journal + .lines() + .filter(|line| !line.contains("\"_bt_record_type\"")) + .count(), + 2, + "the event history remains separate from ordered pipeline controls" ); let spans = std::fs::read_to_string(recovered_data_dir.join("spans/resume.ndjson")).unwrap(); let rows: Vec = spans @@ -2447,7 +2489,7 @@ async fn idle_sessions_are_retired_and_can_resume_from_their_journal() { #[tokio::test] async fn managed_run_flush_is_scoped_to_its_accepted_sessions() { - let (socket, handle, flushes, _tmp) = start_tracking_daemon("test").await; + let (socket, handle, _flushes, tmp) = start_tracking_daemon("test").await; let host = dummy_host(); for (session_id, managed_run_id) in [ ("run-a-main", "run-a"), @@ -2462,17 +2504,24 @@ async fn managed_run_flush_is_scoped_to_its_accepted_sessions() { let result = flush_managed_run("run-a", &socket, 5_000).await.unwrap(); assert!(result.flushed, "managed run did not flush: {result:?}"); assert_eq!(result.pending, 0); - assert_eq!( - *flushes.lock().unwrap(), - HashMap::from([ - ("run-a-main".to_string(), 1), - ("run-a-child".to_string(), 1) - ]) - ); + let controls = |session| { + std::fs::read_to_string(source_journal_path( + &tmp.path().join("data"), + "debug", + session, + )) + .unwrap() + .lines() + .filter(|line| line.contains("pipeline_control")) + .count() + }; + assert_eq!(controls("run-a-main"), 1); + assert_eq!(controls("run-a-child"), 1); + assert_eq!(controls("run-b-main"), 0); let result = flush_managed_run("run-b", &socket, 5_000).await.unwrap(); assert!(result.flushed, "managed run did not flush: {result:?}"); - assert_eq!(flushes.lock().unwrap().get("run-b-main"), Some(&1)); + assert_eq!(controls("run-b-main"), 1); shutdown(&socket).await; handle.await.unwrap(); @@ -2929,19 +2978,17 @@ async fn packaged_grok_hook_replays_bounded_transcripts_to_isolated_debug_routes ) .await; wait_for_grok_translation(&socket, "grok-primary").await; - assert_eq!( - flushes - .lock() - .unwrap() - .get("grok-primary") - .copied() - .unwrap_or_default(), - 0, - "stop alone must not masquerade as the terminal flush" - ); - let primary_spans_path = data_dir.join("spans/grok-primary.ndjson"); let first_ops = read_span_ops(&primary_spans_path); + assert!( + !first_ops.iter().any(|op| match op { + SpanOp::Merge(row) => + row.span_type == SpanType::Task && row.name == "Grok" && row.end_ms.is_some(), + _ => false, + }), + "stop alone must not finalize the session root" + ); + let first_llm = first_ops .iter() .find_map(|op| match op { @@ -3059,7 +3106,7 @@ async fn packaged_grok_hook_replays_bounded_transcripts_to_isolated_debug_routes let journal: Vec = journal_text .lines() .map(|line| serde_json::from_str(line).unwrap()) - .filter(|record: &serde_json::Value| record["_bt_record_type"] != "delivery_checkpoint") + .filter(|record: &serde_json::Value| record.get("_bt_record_type").is_none()) .collect(); assert_eq!(journal.len(), 2); assert_eq!(journal[0]["source"], "grok"); diff --git a/bt-daemon/tests/root_provenance.rs b/bt-daemon/tests/root_provenance.rs index 5a022f64..e0f09230 100644 --- a/bt-daemon/tests/root_provenance.rs +++ b/bt-daemon/tests/root_provenance.rs @@ -398,12 +398,22 @@ async fn span_birth_snapshots_survive_version_changes_and_daemon_restart() { let turn3 = turn_id(&rows[before..], 3_000); assert_ne!(turn3, turn1); assert_ne!(turn3, turn2); - // Graceful shutdown completed these spans; recovery must not redeliver them. - assert!(rows[before..].iter().all(|row| { - ![&root, &turn1, &turn2, &tool] - .iter() - .any(|span| row["span_id"] == **span) - })); + // Journal resume does not repeat old inserts. Newly captured native + // closure may still refine defensive retirement timestamps. + for row in &rows[before..] { + if row["span_id"] == root { + assert_eq!(row["_is_merge"], true); + assert_eq!(row["metrics"]["end"], 4.0); + } + if row["span_id"] == turn2 { + assert_eq!(row["_is_merge"], true); + assert_eq!(row["metrics"]["end"], 3.0); + } + assert_ne!( + row["span_id"], turn1, + "fully resolved history must not be repeated" + ); + } assert_provenance( &rows[before..], &turn3, @@ -478,15 +488,19 @@ async fn legacy_acknowledged_spans_receive_late_updates_without_inventing_origin let data_dir = tmp.path().join("data"); let journal = source_journal_path(&data_dir, "claude-code", session); - let acknowledged_history = std::fs::read(&journal).unwrap(); - assert!(std::str::from_utf8(&acknowledged_history) - .unwrap() + // Model a released pre-SpanOp-journal daemon: accepted native history and + // a legacy delivery checkpoint, with no newer durable partition. + let raw = std::fs::read_to_string(&journal).unwrap(); + let mut history = raw .lines() - .map(|line| serde_json::from_str::(line).unwrap()) - .any(|record| { - record["_bt_record_type"] == "delivery_checkpoint" - && record["through"].as_u64().is_some_and(|offset| offset > 0) - })); + .filter(|line| !line.contains("_bt_record_type")) + .map(|line| format!("{line}\n")) + .collect::(); + let through = history.len() as u64; + history.push_str(&format!("{}\n",json!({"_bt_record_type":"delivery_checkpoint", "route":capture("UserPromptSubmit","v1","native1",1_000).route, "through":through}))); + std::fs::write(&journal, &history).unwrap(); + std::fs::remove_dir_all(data_dir.join("pipeline")).unwrap(); + let acknowledged_history = history.into_bytes(); let ledger_paths: Vec<_> = std::fs::read_dir(data_dir.join("delivery-ledger")) .unwrap() .map(|entry| entry.unwrap().path()) @@ -687,7 +701,11 @@ async fn attached_provisional_root_keeps_birth_snapshot_until_work_after_restart let first = Daemon::start(tmp.path(), &server, "bt1").await; first.send(&birth).await; - first.flush(session).await; + let pending = flush_session(session, &first.socket, 250).await.unwrap(); + assert!( + !pending.flushed && pending.pending > 0, + "a deferred root has not been delivered" + ); assert!( logs3_rows(&server).await.is_empty(), "provisional root exported before work" From 241923b21fd1cfb1a9606aec0b4df911adee2c46 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Wed, 14 Oct 2026 09:09:58 +0800 Subject: [PATCH 2/3] Drain every retirement head and recognize migrated routes --- bt-daemon/src/dispatch.rs | 41 +++++++---- bt-daemon/src/pipeline/journal.rs | 46 ++++++++++-- bt-daemon/src/server/maintenance.rs | 29 ++++++-- bt-daemon/src/wire/envelope.rs | 9 +++ bt-daemon/tests/pipeline.rs | 109 ++++++++++++++++++++++++++++ 5 files changed, 205 insertions(+), 29 deletions(-) diff --git a/bt-daemon/src/dispatch.rs b/bt-daemon/src/dispatch.rs index 42cf3796..3f51fd8b 100644 --- a/bt-daemon/src/dispatch.rs +++ b/bt-daemon/src/dispatch.rs @@ -39,7 +39,7 @@ enum SessionMsg { Barrier(oneshot::Sender<()>), Flush(oneshot::Sender), Finalize(oneshot::Sender), - Shutdown(oneshot::Sender<()>), + Shutdown(Option, oneshot::Sender<()>), } /// Where to rebuild a session's translator state from, streamed at startup. @@ -155,18 +155,22 @@ impl Session { tokio::select! { _=reply.closed()=>{}, pending=waiting=> { let _=reply.send(pending); } } }); } - SessionMsg::Shutdown(reply) => { - if producer.routes.fetch_sub(1, Ordering::Relaxed) == 1 { - // Process retirement is not a native session end. - if let Ok(tail) = - producer.control(crate::journal::ControlKind::Retire).await - { - let _ = tokio::time::timeout( - std::time::Duration::from_secs(2), - route.through(tail), - ) - .await; - } + SessionMsg::Shutdown(target, reply) => { + let last = producer.routes.fetch_sub(1, Ordering::Relaxed) == 1; + let target = match target { + Some(tail) => Some(tail), + None if last => producer + .control(crate::journal::ControlKind::Retire) + .await + .ok(), + None => None, + }; + if let Some(tail) = target { + let _ = tokio::time::timeout( + std::time::Duration::from_secs(2), + route.through(tail), + ) + .await; } route.shutdown().await; if producer.routes.load(Ordering::Relaxed) == 0 { @@ -279,8 +283,17 @@ impl Session { /// Drain, flush, and stop the actor (used on daemon shutdown and when an /// idle session is retired). pub async fn shutdown(&self) { + self.shutdown_through(None).await; + } + + pub(crate) async fn shutdown_through(&self, tail: Option) { let (reply_tx, reply_rx) = oneshot::channel(); - if self.tx.send(SessionMsg::Shutdown(reply_tx)).await.is_ok() { + if self + .tx + .send(SessionMsg::Shutdown(tail, reply_tx)) + .await + .is_ok() + { let _ = reply_rx.await; } } diff --git a/bt-daemon/src/pipeline/journal.rs b/bt-daemon/src/pipeline/journal.rs index eadf704e..96b3f5ab 100644 --- a/bt-daemon/src/pipeline/journal.rs +++ b/bt-daemon/src/pipeline/journal.rs @@ -265,14 +265,17 @@ impl SpanJournal { /// A completed subscription can remain dormant until its source has new work. pub async fn route_complete(&self, route: &SessionRoute, input: u64) -> anyhow::Result { - let route = serde_json::to_string(route)?; + let route = route.clone(); self.with_db(move |db| { - Ok(db.query_row( - "SELECT EXISTS(SELECT 1 FROM heads WHERE route=? AND failure IS NULL - AND position >= (SELECT coalesce(max(position),0) FROM ops JOIN batches USING(input) WHERE committed=1)) - AND (SELECT coalesce(max(input),0) FROM batches WHERE committed=1)>=?", - params![route,input], |row| row.get(0), - )?) + let committed: u64 = db.query_row("SELECT coalesce(max(input),0) FROM batches WHERE committed=1",[],|r|r.get(0))?; + if committed < input {return Ok(false);} + let tail: u64 = db.query_row("SELECT coalesce(max(position),0) FROM ops JOIN batches USING(input) WHERE committed=1",[],|r|r.get(0))?; + let mut query = db.prepare("SELECT route FROM heads WHERE failure IS NULL AND position>=?")?; + for stored in query.query_map([tail],|r|r.get::<_,String>(0))? { + let stored: SessionRoute = serde_json::from_str(&stored?)?; + if stored.same_route(&route) {return Ok(true);} + } + Ok(false) }).await } @@ -593,4 +596,33 @@ mod tests { reader.rewind().await.unwrap(); assert!(reader.next_batch(8).await.unwrap().is_none()); } + #[tokio::test] + async fn a_completed_canonical_head_also_covers_its_legacy_named_route() { + let tmp = tempfile::tempdir().unwrap(); + let journal = SpanJournal::open(&tmp.path().join("spans.sqlite")) + .await + .unwrap(); + journal.begin(10, true).await.unwrap(); + journal.append(10, &[op("one")]).await.unwrap(); + journal.commit(10).await.unwrap(); + let mut legacy = route("work"); + legacy.auth.profile = Some("work".into()); + let mut canonical = legacy.clone(); + canonical.auth.profile_id = Some("profile-id".into()); + let mut head = journal.subscribe(&canonical, 0).await.unwrap(); + head.next_batch(8) + .await + .unwrap() + .unwrap() + .into_completed() + .await + .ok() + .unwrap(); + assert!(journal.route_complete(&legacy, 10).await.unwrap()); + let mut other_profile = canonical.clone(); + other_profile.auth.profile_id = Some("different-id".into()); + assert!(!journal.route_complete(&other_profile, 10).await.unwrap()); + legacy.tags.push("different route".into()); + assert!(!journal.route_complete(&legacy, 10).await.unwrap()); + } } diff --git a/bt-daemon/src/server/maintenance.rs b/bt-daemon/src/server/maintenance.rs index 837cc5e3..5654ed0a 100644 --- a/bt-daemon/src/server/maintenance.rs +++ b/bt-daemon/src/server/maintenance.rs @@ -123,20 +123,33 @@ pub(super) async fn drain_all(daemon: &Arc) { if let Err(error) = resolve_pending_sessions(daemon, |_| true).await { tracing::warn!(%error, "pending correlation events retained for recovery during shutdown"); } - // Publish shared retirement output before stopping any route subscriber. + // Publish each source's retirement boundary once, then give every route + // a chance to consume it before its actor stops. let sources: Vec<_> = daemon .translations .lock() .unwrap() - .values() - .cloned() + .iter() + .map(|(key, source)| (key.clone(), source.clone())) .collect(); - for source in sources { - let _ = source.control(crate::journal::ControlKind::Retire).await; + let mut targets = HashMap::new(); + for (key, source) in sources { + if let Ok(tail) = source.control(crate::journal::ControlKind::Retire).await { + targets.insert(key, tail); + } } - let sessions: Vec> = daemon.sessions.lock().unwrap().values().cloned().collect(); - for s in sessions { - s.shutdown().await; + let sessions: Vec<_> = daemon + .sessions + .lock() + .unwrap() + .iter() + .map(|(key, session)| (key.clone(), session.clone())) + .collect(); + for (key, session) in sessions { + let target = targets + .get(&crate::ids::session_namespace(&key.source, &key.session_id)) + .copied(); + session.shutdown_through(target).await; } let sources: Vec<_> = daemon .translations diff --git a/bt-daemon/src/wire/envelope.rs b/bt-daemon/src/wire/envelope.rs index 599dec00..ba9d5e13 100644 --- a/bt-daemon/src/wire/envelope.rs +++ b/bt-daemon/src/wire/envelope.rs @@ -174,6 +174,15 @@ impl SessionRoute { pub fn same_route(&self, other: &Self) -> bool { let mut left = self.clone(); let mut right = other.clone(); + if left + .auth + .profile_id + .as_ref() + .zip(right.auth.profile_id.as_ref()) + .is_some_and(|(left, right)| left != right) + { + return false; + } // A stable-ID route is the canonical successor to an old named route. // Journals can legitimately contain both forms for the same session, // so replay compares their unchanged routing intent, not the newly diff --git a/bt-daemon/tests/pipeline.rs b/bt-daemon/tests/pipeline.rs index f811eefd..78d0de47 100644 --- a/bt-daemon/tests/pipeline.rs +++ b/bt-daemon/tests/pipeline.rs @@ -3375,3 +3375,112 @@ async fn status_counts_rows_emitted_before_a_plugin_failure() { shutdown(&socket).await; handle.await.unwrap(); } + +#[tokio::test] +async fn clean_shutdown_delivers_the_shared_retirement_tail_to_every_route() { + use std::sync::atomic::{AtomicUsize, Ordering}; + struct Shared { + started: AtomicUsize, + release: tokio::sync::Semaphore, + delivered: Mutex>, + } + struct Factory(Arc); + struct RouteSink { + shared: Arc, + tag: String, + } + impl SinkFactory for Factory { + fn create(&self, _: &str, _: &str) -> anyhow::Result> { + Ok(Box::new(RouteSink { + shared: self.0.clone(), + tag: String::new(), + })) + } + } + #[async_trait] + impl Sink for RouteSink { + fn configure(&mut self, config: &SessionConfig) { + self.tag = config.tags[0].clone(); + } + async fn emit(&mut self, ops: &[SpanOp]) -> anyhow::Result { + if ops.iter().any(|op| match op { + SpanOp::Insert(row) | SpanOp::Merge(row) => { + row.parent_span_ids.is_empty() && row.end_ms.is_some() + } + }) { + self.shared.started.fetch_add(1, Ordering::SeqCst); + self.shared.release.acquire().await.unwrap().forget(); + self.shared + .delivered + .lock() + .unwrap() + .insert(self.tag.clone()); + } + Ok(ops.len() as u64) + } + async fn flush(&mut self) -> anyhow::Result<()> { + Ok(()) + } + } + let tmp = tempfile::tempdir().unwrap(); + let data_dir = tmp.path().join("data"); + let socket = test_endpoint(tmp.path()); + let shared = Arc::new(Shared { + started: AtomicUsize::new(0), + release: tokio::sync::Semaphore::new(0), + delivered: Mutex::new(HashSet::new()), + }); + let mut opts = debug_serve_options("test", &data_dir); + opts.auth_provider = Some(Arc::new(TestAuthProvider { + calls: Mutex::new(Vec::new()), + fail: false, + first_lease_expired: false, + })); + opts.sink_factory = Arc::new(Factory(shared.clone())); + let args = ServeArgs { + socket: Some(socket.clone()), + data_dir: Some(data_dir.clone()), + idle_timeout_secs: 0, + session_idle_timeout_secs: 0, + }; + let handle = tokio::spawn(async move { run_serve(args, opts).await.unwrap() }); + wait_for(&socket).await; + for tag in ["left", "right"] { + let mut env = envelope("retirement-routes", "UserPromptSubmit", 1); + env.source = "claude-code".into(); + env.payload = serde_json::json!({"session_id":env.session_id,"hook_event_name":"UserPromptSubmit","prompt":"work"}); + env.route.as_mut().unwrap().tags = vec![tag.into()]; + forward_envelope(&env, &socket, &dummy_host(), false) + .await + .unwrap(); + } + assert!( + flush_session("retirement-routes", &socket, 5000) + .await + .unwrap() + .flushed + ); + let gate = shared.clone(); + let release = tokio::spawn(async move { + let all_started = tokio::time::timeout(Duration::from_secs(1), async { + while gate.started.load(Ordering::SeqCst) < 2 { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await; + gate.release.add_permits(2); + all_started.unwrap(); + }); + shutdown(&socket).await; + handle.await.unwrap(); + release.await.unwrap(); + assert_eq!( + *shared.delivered.lock().unwrap(), + HashSet::from(["left".into(), "right".into()]) + ); + let (head, tail) = head_and_tail(&data_dir); + assert_eq!( + head, tail, + "every route durably completed the retirement output" + ); +} From efa21193ca763aee8a0a647ccb749e7b45d72fe8 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Fri, 16 Oct 2026 22:46:00 +0800 Subject: [PATCH 3/3] Align journal activation with all-span provenance and panic bans --- bt-daemon/src/dispatch.rs | 8 ++--- bt-daemon/src/pipeline/consumer.rs | 23 ++++++------ bt-daemon/src/pipeline/translation.rs | 51 +++++++++++++++------------ bt-daemon/src/server/maintenance.rs | 16 +++------ bt-daemon/src/server/mod.rs | 7 +--- bt-daemon/src/server/rpc.rs | 5 ++- bt-daemon/src/server/sessions.rs | 18 ++++------ 7 files changed, 58 insertions(+), 70 deletions(-) diff --git a/bt-daemon/src/dispatch.rs b/bt-daemon/src/dispatch.rs index 3f51fd8b..21d50103 100644 --- a/bt-daemon/src/dispatch.rs +++ b/bt-daemon/src/dispatch.rs @@ -13,8 +13,8 @@ use crate::pipeline::{ }; use crate::sink::SinkFactory; use crate::translate::Registry; -use parking_lot::Mutex; use crate::wire::Envelope; +use parking_lot::Mutex; use std::path::PathBuf; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; @@ -114,7 +114,7 @@ impl Session { route.configure(config); } if let Err(error) = producer.advance(through).await { - *errors.lock().unwrap() = Some(error.to_string()); + *errors.lock() = Some(error.to_string()); } } SessionMsg::Configure(config, reply) => { @@ -142,12 +142,12 @@ impl Session { Ok(tail) => match route.through(tail).await { Ok(()) => 0, Err(error) => { - *errors.lock().unwrap() = Some(error); + *errors.lock() = Some(error); queued.queued.load(Ordering::Relaxed).max(1) } }, Err(error) => { - *errors.lock().unwrap() = Some(error); + *errors.lock() = Some(error); queued.queued.load(Ordering::Relaxed).max(1) } } diff --git a/bt-daemon/src/pipeline/consumer.rs b/bt-daemon/src/pipeline/consumer.rs index e09b816c..b9f3fef7 100644 --- a/bt-daemon/src/pipeline/consumer.rs +++ b/bt-daemon/src/pipeline/consumer.rs @@ -10,9 +10,10 @@ use crate::{ sink::{Sink, SinkFactory}, wire::SessionConfig, }; +use parking_lot::Mutex; use std::{ path::PathBuf, - sync::{atomic::Ordering, Arc, Mutex}, + sync::{atomic::Ordering, Arc}, }; use tokio::sync::{watch, Notify}; @@ -73,7 +74,7 @@ impl Consumer { ); let mut held = false; if let Some(failure) = journal.failure(&cursor.route).await? { - *options.error.lock().unwrap() = Some(failure); + *options.error.lock() = Some(failure); } let mut completion: Option> = None; loop { @@ -120,7 +121,7 @@ impl Consumer { .counters .spans_emitted .fetch_add(count, Ordering::Relaxed); - *options.permalink.lock().unwrap() = sink.permalink(); + *options.permalink.lock() = sink.permalink(); // The journal retains a deferred prefix. No payloads or token list // need accumulate in memory while waiting for its releasing child. drop(delivery); @@ -129,10 +130,10 @@ impl Consumer { Ok(()) => { cursor.rewind().await?; progress.send_replace(cursor.completed_position()); - *options.error.lock().unwrap() = None; + *options.error.lock() = None; } Err(failure) => { - *options.error.lock().unwrap() = Some(format!( + *options.error.lock() = Some(format!( "route completion held: {}", failure.error )); @@ -155,9 +156,9 @@ impl Consumer { .spans_emitted .fetch_add(failure.emitted, Ordering::Relaxed); } - *options.error.lock().unwrap() = + *options.error.lock() = Some(format!("routing held: {error:#}")); - let failure = options.error.lock().unwrap().clone(); + let failure = options.error.lock().clone(); journal.set_failure(&cursor.route, failure).await?; held = true; progress.send_replace(cursor.completed_position()); @@ -173,11 +174,11 @@ impl Consumer { } }; if let Err(error) = task.await { - *options.error.lock().unwrap() = Some(format!("routing journal held: {error:#}")); + *options.error.lock() = Some(format!("routing journal held: {error:#}")); progress.send_replace(cursor.completed_position()); } }); - *handle.task.lock().unwrap() = Some(task); + *handle.task.lock() = Some(task); handle } pub fn configure(&self, config: SessionConfig) { @@ -191,7 +192,7 @@ impl Consumer { } pub async fn shutdown(&self) { self.stop.notify_one(); - let task = self.task.lock().unwrap().take(); + let task = self.task.lock().take(); if let Some(task) = task { let _ = task.await; } @@ -202,7 +203,7 @@ impl Consumer { if *progress.borrow_and_update() >= tail { return Ok(()); } - if let Some(error) = self.error.lock().unwrap().clone() { + if let Some(error) = self.error.lock().clone() { return Err(error); } progress diff --git a/bt-daemon/src/pipeline/translation.rs b/bt-daemon/src/pipeline/translation.rs index f3e2e994..bf032fd9 100644 --- a/bt-daemon/src/pipeline/translation.rs +++ b/bt-daemon/src/pipeline/translation.rs @@ -3,15 +3,16 @@ use super::journal::SpanJournal; use super::prepare::Prepare; use crate::{ journal::{ControlKind, JournalReader, JournalRecord, JournalWriter, PipelineControl}, - translate::{AgentTranslator, OriginSnapshot, Registry, SessionCtx, SpanOp, SpanType}, + translate::{AgentTranslator, OriginSnapshot, Registry, SessionCtx, SpanOp}, wire::{SessionConfig, SessionRoute}, }; +use parking_lot::Mutex; use std::{ collections::HashMap, path::PathBuf, sync::{ atomic::{AtomicU64, AtomicUsize, Ordering}, - Arc, Mutex, + Arc, }, }; use tokio::sync::{mpsc, oneshot, Notify}; @@ -86,6 +87,7 @@ impl Translation { reader: None, observers: Vec::new(), origins: HashMap::new(), + capture: None, bounds: HashMap::new(), position: 0, committed, @@ -103,7 +105,7 @@ impl Translation { } pub async fn shutdown(&self) { let _ = self.tx.send(Command::Stop).await; - let task = self.task.lock().unwrap().take(); + let task = self.task.lock().take(); if let Some(task) = task { let _ = task.await; } @@ -153,6 +155,7 @@ struct Producer { reader: Option, observers: Vec, origins: HashMap, + capture: Option, bounds: HashMap, position: u64, committed: u64, @@ -182,7 +185,7 @@ impl Producer { }; if let Err(error) = result { self.held = true; - *self.error.lock().unwrap() = + *self.error.lock() = Some(format!("translation held at {}: {error:#}", self.position)); } } @@ -193,7 +196,6 @@ impl Producer { return Err(self .error .lock() - .unwrap() .clone() .unwrap_or_else(|| "translation held".into())); } @@ -275,7 +277,7 @@ impl Producer { while let Some(entry) = self .reader .as_mut() - .expect("reader initialized") + .ok_or_else(|| anyhow::anyhow!("event journal is missing"))? .next_record() .await? { @@ -315,10 +317,11 @@ impl Producer { source_version: env.source_version.clone(), bt_version: self.options.version.clone(), }; + self.capture = Some(capture.clone()); ( self.translator .as_mut() - .expect("translator initialized") + .ok_or_else(|| anyhow::anyhow!("translator is missing"))? .handle(&env, &ctx)?, Some(capture), ) @@ -330,13 +333,16 @@ impl Producer { self.position = input; continue; } - let translator = self.translator.as_mut().expect("translator initialized"); + let translator = self + .translator + .as_mut() + .ok_or_else(|| anyhow::anyhow!("translator is missing"))?; let ops = if !matches!(control.kind, ControlKind::Checkpoint) { translator.finalize_at(&ctx, &control.bounds)? } else { translator.checkpoint_at(&ctx, &control.bounds)? }; - (ops, None) + (ops, self.capture.clone()) } JournalRecord::DeliveryCheckpoint { .. } => { if input > self.committed { @@ -358,15 +364,14 @@ impl Producer { for op in &mut ops { let insert = matches!(op, SpanOp::Insert(_)); let (SpanOp::Insert(row) | SpanOp::Merge(row)) = op; - if insert - && (row.is_turn - || (row.span_type == SpanType::Task && row.parent_span_ids.is_empty())) - { - if let Some(capture) = &capture { - self.origins - .entry(row.span_id.clone()) - .or_insert_with(|| capture.clone()); - } + if insert { + self.origins.entry(row.span_id.clone()).or_insert_with(|| { + capture.clone().unwrap_or_else(|| OriginSnapshot { + plugin_version: None, + source_version: None, + bt_version: self.options.version.clone(), + }) + }); } row.origin = self.origins.get(&row.span_id).cloned(); } @@ -381,7 +386,7 @@ impl Producer { next = self .translator .as_mut() - .expect("translator initialized") + .ok_or_else(|| anyhow::anyhow!("translator is missing"))? .drain_pending(&ctx)?; } if publish { @@ -398,10 +403,10 @@ impl Producer { let mut after = 0; loop { let rows = self.journal.read_after(after, 128).await?; - if rows.is_empty() { + let Some(&(through, _)) = rows.last() else { break; - } - after = rows.last().expect("nonempty").0; + }; + after = through; let ops = rows.into_iter().map(|(_, op)| op).collect::>(); Self::observe(&self.options, &ops, observer).await?; } @@ -485,7 +490,7 @@ mod tests { translation.advance(input).await.unwrap(); assert_eq!(translation.barrier().await.unwrap(), tail); assert_eq!(translation.progress.pending(), 0); - assert!(translation.error.lock().unwrap().is_none()); + assert!(translation.error.lock().is_none()); let mut route = translation .journal .subscribe(&SessionRoute::default(), 0) diff --git a/bt-daemon/src/server/maintenance.rs b/bt-daemon/src/server/maintenance.rs index 5654ed0a..ea5ded27 100644 --- a/bt-daemon/src/server/maintenance.rs +++ b/bt-daemon/src/server/maintenance.rs @@ -128,7 +128,6 @@ pub(super) async fn drain_all(daemon: &Arc) { let sources: Vec<_> = daemon .translations .lock() - .unwrap() .iter() .map(|(key, source)| (key.clone(), source.clone())) .collect(); @@ -141,7 +140,6 @@ pub(super) async fn drain_all(daemon: &Arc) { let sessions: Vec<_> = daemon .sessions .lock() - .unwrap() .iter() .map(|(key, session)| (key.clone(), session.clone())) .collect(); @@ -151,19 +149,13 @@ pub(super) async fn drain_all(daemon: &Arc) { .copied(); session.shutdown_through(target).await; } - let sources: Vec<_> = daemon - .translations - .lock() - .unwrap() - .values() - .cloned() - .collect(); + let sources: Vec<_> = daemon.translations.lock().values().cloned().collect(); for source in sources { source.shutdown().await; } - daemon.sessions.lock().unwrap().clear(); - daemon.translations.lock().unwrap().clear(); - daemon.journals.lock().unwrap().clear(); + daemon.sessions.lock().clear(); + daemon.translations.lock().clear(); + daemon.journals.lock().clear(); let (reply, done) = oneshot::channel(); if daemon .ingress_tx diff --git a/bt-daemon/src/server/mod.rs b/bt-daemon/src/server/mod.rs index f29dbf59..8bd27b84 100644 --- a/bt-daemon/src/server/mod.rs +++ b/bt-daemon/src/server/mod.rs @@ -191,12 +191,7 @@ impl Daemon { /// durable on disk and resume after a restart, so they must not keep an /// otherwise idle daemon alive forever. fn total_queued(&self) -> u64 { - let dispatched: u64 = self - .sessions - .lock() - .values() - .map(|s| s.pending()) - .sum(); + let dispatched: u64 = self.sessions.lock().values().map(|s| s.pending()).sum(); dispatched.saturating_add( self.pending_sessions .lock() diff --git a/bt-daemon/src/server/rpc.rs b/bt-daemon/src/server/rpc.rs index 2bf7cb0c..84c7b531 100644 --- a/bt-daemon/src/server/rpc.rs +++ b/bt-daemon/src/server/rpc.rs @@ -336,13 +336,12 @@ impl Daemon { route: serde_json::from_str(&key.route).ok(), queued: s.pending(), spans_emitted: s.counters.spans_emitted.load(Ordering::Relaxed), - permalink: s.permalink.lock().unwrap().clone(), + permalink: s.permalink.lock().clone(), last_error: s .last_error .lock() - .unwrap() .clone() - .or_else(|| s.source_error.lock().unwrap().clone()), + .or_else(|| s.source_error.lock().clone()), }) .collect(); for (key, (source, error)) in self.auth_errors.lock().iter() { diff --git a/bt-daemon/src/server/sessions.rs b/bt-daemon/src/server/sessions.rs index 4b52003a..22f2d2e5 100644 --- a/bt-daemon/src/server/sessions.rs +++ b/bt-daemon/src/server/sessions.rs @@ -81,7 +81,7 @@ impl Daemon { .journal_writer_for(&env.source, &env.session_id) .await?; let storage_key = crate::ids::session_namespace(&env.source, &env.session_id); - let existing = self.translations.lock().unwrap().get(&storage_key).cloned(); + let existing = self.translations.lock().get(&storage_key).cloned(); let translation = match existing { Some(translation) => translation, None => { @@ -102,7 +102,6 @@ impl Daemon { .await?; self.translations .lock() - .unwrap() .insert(storage_key, translation.clone()); translation } @@ -121,7 +120,7 @@ impl Daemon { .journal .subscribe(route, replay.acknowledged_through) .await?; - let mut map = self.sessions.lock().unwrap(); + let mut map = self.sessions.lock(); if let Some(s) = map.get(key) { return Ok(s.clone()); } @@ -182,14 +181,11 @@ impl Daemon { // lock for the brief writer-map cleanup after daemon work ends. let capture_lock = self.session_lock(&key.source, &key.session_id); let _capture_guard = capture_lock.lock().await; - self.claude_transcripts - .lock() - .unwrap() - .retain(|candidate, _| { - candidate.source != key.source || candidate.session_id != key.session_id - }); - self.journals.lock().unwrap().remove(&storage_key); - self.translations.lock().unwrap().remove(&storage_key); + self.claude_transcripts.lock().retain(|candidate, _| { + candidate.source != key.source || candidate.session_id != key.session_id + }); + self.journals.lock().remove(&storage_key); + self.translations.lock().remove(&storage_key); { let mut locks = self.session_locks.lock(); // A waiting capture or observer must keep using this same