From d9373bc42f35d3f70207302e967282cd1dea62bd Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Wed, 14 Oct 2026 02:40:08 +0800 Subject: [PATCH 1/4] Recover pipeline failures at durable boundaries --- bt-daemon/src/delivery_ledger.rs | 14 + bt-daemon/src/dispatch.rs | 19 +- bt-daemon/src/pipeline/consumer.rs | 767 ++++++++++++++++++---- bt-daemon/src/pipeline/journal.rs | 25 +- bt-daemon/src/pipeline/mod.rs | 2 + bt-daemon/src/pipeline/plugins.rs | 106 ++- bt-daemon/src/pipeline/recovery.rs | 277 ++++++++ bt-daemon/src/pipeline/routing.rs | 19 +- bt-daemon/src/pipeline/translation.rs | 351 ++++++++-- bt-daemon/src/server/correlation_store.rs | 51 +- bt-daemon/src/server/ingress.rs | 12 + bt-daemon/src/server/mod.rs | 7 +- bt-daemon/src/server/rpc.rs | 23 +- bt-daemon/src/server/sessions.rs | 90 +-- bt-daemon/src/span_processor.rs | 92 ++- bt-daemon/src/trace_runtime.rs | 7 +- bt-daemon/src/wire/methods.rs | 11 + bt-daemon/src/wire/mod.rs | 2 +- bt-daemon/tests/pipeline.rs | 160 ++++- 19 files changed, 1756 insertions(+), 279 deletions(-) create mode 100644 bt-daemon/src/pipeline/recovery.rs diff --git a/bt-daemon/src/delivery_ledger.rs b/bt-daemon/src/delivery_ledger.rs index b389cdf8..814687a2 100644 --- a/bt-daemon/src/delivery_ledger.rs +++ b/bt-daemon/src/delivery_ledger.rs @@ -248,6 +248,20 @@ fn late_merge_id(row: &SpanRow, key: &str) -> String { format!("{}\u{1f}{key}", row.span_id) } +/// The backend accepted delivery; only the local ledger write needs retrying. +#[derive(Debug)] +pub(crate) struct CommitFailure(pub anyhow::Error); +impl std::fmt::Display for CommitFailure { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + self.0.fmt(f) + } +} +impl std::error::Error for CommitFailure { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + Some(self.0.as_ref()) + } +} + /// Wrap a sink with destination-scoped terminal-span suppression. If durable /// ledger state cannot be loaded, callers retain normal delivery rather than /// risking lost trace data. diff --git a/bt-daemon/src/dispatch.rs b/bt-daemon/src/dispatch.rs index 21d50103..132f3efe 100644 --- a/bt-daemon/src/dispatch.rs +++ b/bt-daemon/src/dispatch.rs @@ -43,14 +43,6 @@ enum SessionMsg { } /// Where to rebuild a session's translator state from, streamed at startup. -pub struct ReplayPlan { - pub journal_path: PathBuf, - /// 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. - pub acknowledged_through: u64, -} - pub(crate) struct SessionOptions { pub session_id: String, pub source: String, @@ -59,6 +51,7 @@ pub(crate) struct SessionOptions { pub data_dir: PathBuf, pub translation: Arc, pub cursor: crate::pipeline::journal::Cursor, + pub auth_provider: Option>, } /// Handle to one live session: its queue plus observable counters/state. @@ -68,6 +61,8 @@ pub struct Session { pub counters: Arc, pub last_error: Arc>>, source_progress: Arc, + pub recovery: Arc>>, + pub source_recovery: Arc>>, pub source_error: Arc>>, pub permalink: Arc>>, last_activity: Mutex, @@ -82,6 +77,7 @@ impl Session { ) -> Arc { let (tx, mut rx) = mpsc::channel(QUEUE_CAPACITY); let counters = Arc::new(Counters::default()); + let recovery = Arc::new(Mutex::new(None)); let last_error = Arc::new(Mutex::new(None)); let permalink = Arc::new(Mutex::new(None)); let source = options.source.clone(); @@ -98,10 +94,13 @@ impl Session { error: last_error.clone(), permalink: permalink.clone(), factory: sink_factory, + auth_provider: options.auth_provider, + recovery: recovery.clone(), }, ); let producer = options.translation; let source_error = producer.error.clone(); + let source_recovery = producer.recovery.clone(); let source_progress = producer.progress.clone(); let errors = last_error.clone(); let queued = counters.clone(); @@ -190,6 +189,8 @@ impl Session { last_error, source_error, source_progress, + recovery, + source_recovery, permalink, last_activity: Mutex::new(Instant::now()), }) @@ -586,6 +587,8 @@ mod tests { counters: Arc::new(Counters::default()), last_error: Arc::new(Mutex::new(None)), source_error: Arc::new(Mutex::new(None)), + recovery: Arc::new(Mutex::new(None)), + source_recovery: 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/pipeline/consumer.rs b/bt-daemon/src/pipeline/consumer.rs index b9f3fef7..ed7f93e3 100644 --- a/bt-daemon/src/pipeline/consumer.rs +++ b/bt-daemon/src/pipeline/consumer.rs @@ -1,14 +1,17 @@ -//! A routing subscription owns pending work independently of the source producer. +//! One route owns processing and recovery at its durable SpanOp head. use super::{ + delivery::InputPort, journal::{Cursor, SpanJournal}, + recovery::{Boundary, Failure, Recovery, RecoveryOutcome}, routing::RoutingSink, - Delivery, + Delivery, Stage, }; use crate::{ - delivery_ledger::LedgerSink, + delivery_ledger::{CommitFailure, LedgerSink}, dispatch::Counters, sink::{Sink, SinkFactory}, - wire::SessionConfig, + wire::{RecoveryInfo, SessionConfig, SessionRoute}, + AuthProvider, AuthResolveReason, }; use parking_lot::Mutex; use std::{ @@ -17,6 +20,16 @@ use std::{ }; use tokio::sync::{watch, Notify}; +enum Work { + Resubmit, + Flush, + Complete(Delivery<()>), +} +enum State { + Processing, + Recovering(Failure, Work), +} + pub(crate) struct Consumer { task: Mutex>>, config: watch::Sender, @@ -24,7 +37,6 @@ pub(crate) struct Consumer { stop: Arc, error: Arc>>, } - pub(crate) struct ConsumerOptions { pub source: String, pub session_id: String, @@ -33,8 +45,78 @@ pub(crate) struct ConsumerOptions { pub config: SessionConfig, pub counters: Arc, pub error: Arc>>, + pub recovery: Arc>>, pub permalink: Arc>>, pub factory: Arc, + pub auth_provider: Option>, +} + +impl ConsumerOptions { + async fn sink(&self, config: &SessionConfig) -> anyhow::Result { + let inner = self.factory.create(&self.session_id, &self.source)?; + let ledger = LedgerSink::new( + inner, + Some(&self.data_dir), + &self.source, + &self.session_id, + Some(config), + &self.version, + ) + .await; + let mut sink = RoutingSink::spawn( + Box::new(ledger), + self.source.clone(), + self.session_id.clone(), + self.data_dir.clone(), + self.error.clone(), + ); + sink.configure(config); + Ok(sink) + } + async fn hold(&self, journal: &SpanJournal, cursor: &Cursor, failure: &mut Failure) -> bool { + failure.through.get_or_insert(cursor.position()); + *self.error.lock().unwrap() = Some(failure.message()); + *self.recovery.lock().unwrap() = Some(failure.info(false)); + if let Err(error) = journal + .set_failure( + &cursor.route, + Some(serde_json::to_string(failure).expect("failure evidence")), + ) + .await + { + tracing::warn!(%error,"route recovery evidence could not be persisted; original failure retained"); + return false; + } + true + } + async fn credentials(&self, route: &str, config: &mut SessionConfig) -> anyhow::Result<()> { + let provider = self + .auth_provider + .as_ref() + .ok_or_else(|| anyhow::anyhow!("daemon host has no auth provider"))?; + let route: SessionRoute = serde_json::from_str(route)?; + let lease = crate::server::resolve_route_auth( + provider.as_ref(), + &route.auth, + AuthResolveReason::Unauthorized, + route.auth.org_name.as_deref(), + ) + .await?; + anyhow::ensure!( + route.auth.profile_id.is_none() || route.auth.profile_id == lease.selection.profile_id, + "credential refresh changed the pinned profile" + ); + anyhow::ensure!( + config.auth.org_id.is_none() || config.auth.org_id == lease.auth.org_id, + "credential refresh changed the pinned organization" + ); + anyhow::ensure!( + config.auth.api_url.is_none() || config.auth.api_url == lease.auth.api_url, + "credential refresh changed the pinned backend" + ); + config.auth = lease.auth; + Ok(()) + } } impl Consumer { @@ -44,138 +126,349 @@ impl Consumer { let stop = Arc::new(Notify::new()); let handle = Arc::new(Self { task: Mutex::new(None), - config, + config: config.clone(), 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() = Some(failure); - } - let mut completion: Option> = None; - loop { - if held { + let mut sink: Option = None; + let restored = journal.failure(&cursor.route).await; + let mut state = match restored { + Ok(Some(raw)) => State::Recovering( + serde_json::from_str(&raw).unwrap_or_else(|_| { + Failure::legacy(raw, cursor.completed_position() + 1, &options.config) + }), + Work::Resubmit, + ), + Ok(None) => State::Processing, + Err(error) => State::Recovering( + Failure::storage(&error, cursor.completed_position() + 1, &options.version, 0), + Work::Resubmit, + ), + }; + let mut attempts = 0; + let mut recorded = None; + let mut record_retry = 0; + loop { + match state { + State::Recovering(mut failure, work) => { + if matches!(failure.boundary, Boundary::Completion) + && cursor.completed_position() + >= failure.through.unwrap_or(failure.position) + && journal.set_failure(&cursor.route, None).await.is_ok() + { + *options.error.lock().unwrap() = None; + *options.recovery.lock().unwrap() = None; + state = State::Processing; + continue; + } + failure.through.get_or_insert(cursor.position()); + let encoded = serde_json::to_string(&failure).expect("failure evidence"); + if recorded.as_ref() != Some(&encoded) + && super::recovery::now_ms() >= record_retry + { + if options.hold(&journal, &cursor, &mut failure).await { + recorded = Some(encoded.clone()); + } else { + record_retry = super::recovery::now_ms().saturating_add(1_000); + } + } + progress.send_replace(cursor.completed_position()); 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; }, + _=stop.notified()=>break, + _=tokio::time::sleep(std::time::Duration::from_millis(250))=>{}, + result=settings.changed()=>if result.is_err() { break; }, + } + let mut current = settings.borrow_and_update().clone(); + if recorded.as_ref() != Some(&encoded) + || matches!( + (Recovery { + config: Some(¤t), + version: &options.version + }) + .process(&failure) + .await, + RecoveryOutcome::Waiting + ) + { + state = State::Recovering(failure, work); + continue; + } + *options.recovery.lock().unwrap() = Some(failure.info(true)); + attempts = failure.attempt.saturating_add(1); + if matches!(failure.boundary, Boundary::Credentials) { + if let Err(error) = + options.credentials(&cursor.route, &mut current).await + { + state = State::Recovering( + Failure::credentials( + cursor.completed_position() + 1, + error.to_string(), + attempts, + ), + work, + ); + continue; + } + config.send_replace(current.clone()); + } + let retry = match work { + Work::Complete(delivery) => match delivery.into_completed().await { + Ok(()) => Ok(true), + Err(unfinished) => { + let mut failure = Failure::storage( + &unfinished.error, + cursor.completed_position() + 1, + &options.version, + attempts, + ); + failure.boundary = Boundary::Completion; + state = State::Recovering( + failure, + Work::Complete(unfinished.delivery), + ); + continue; + } + }, + Work::Flush => sink + .as_mut() + .expect("accepted delivery retains its sink") + .flush() + .await + .map(|()| true), + Work::Resubmit => { + sink = None; + Ok(false) + } + }; + match retry { + Ok(completed) => { + if completed + && !sink + .as_ref() + .is_some_and(|sink| sink.has_pending_delivery()) + { + if let Err(unfinished) = + cursor.pending_prefix().into_completed().await + { + let mut failure = Failure::storage( + &unfinished.error, + cursor.completed_position() + 1, + &options.version, + attempts, + ); + failure.boundary = Boundary::Completion; + state = State::Recovering( + failure, + Work::Complete(unfinished.delivery), + ); + continue; } } - held=false; - cursor.rewind().await?; + if let Err(error) = cursor.rewind().await { + state = State::Recovering( + Failure::storage( + &error, + cursor.completed_position() + 1, + &options.version, + attempts, + ), + Work::Resubmit, + ); + continue; + } + progress.send_replace(cursor.completed_position()); + if completed { + *options.error.lock().unwrap() = None; + *options.recovery.lock().unwrap() = None; + recorded = None; + } + state = State::Processing; + } + Err(error) if error.downcast_ref::().is_some() => { + let mut failure = Failure::storage( + &error, + cursor.completed_position() + 1, + &options.version, + attempts, + ); + failure.boundary = Boundary::Ledger; + state = State::Recovering(failure, Work::Flush); + } + Err(error) => { + state = State::Recovering( + Failure::routing( + &error, + cursor.completed_position() + 1, + ¤t, + attempts, + ), + Work::Resubmit, + ); } } - 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) => { + State::Processing => { + let tail = match journal.tail().await { + Ok(tail) => tail, + Err(error) => { + state = State::Recovering( + Failure::storage( + &error, + cursor.completed_position() + 1, + &options.version, + attempts, + ), + Work::Resubmit, + ); + continue; + } + }; + options.counters.queued.store( + tail.saturating_sub(cursor.completed_position()), + Ordering::Relaxed, + ); + let received = tokio::select! { + _=stop.notified()=>break, + result=InputPort::receive(&mut cursor)=>result, + result=settings.changed()=> { if result.is_err() {break;} state=State::Processing; continue; }, + }; + let delivery = match received { + Ok(Some(delivery)) => delivery, + Ok(None) => break, + Err(error) => { + state = State::Recovering( + Failure::storage( + &error, + cursor.completed_position() + 1, + &options.version, + attempts, + ), + Work::Resubmit, + ); + continue; + } + }; + let current = settings.borrow_and_update().clone(); + if sink.is_none() { + match options.sink(¤t).await { + Ok(created) => sink = Some(created), + Err(error) => { + state = State::Recovering( + Failure::routing( + &error, + cursor.completed_position() + 1, + ¤t, + attempts, + ), + Work::Resubmit, + ); + continue; + } + } + } + let sink = sink.as_mut().expect("configured sink"); + sink.configure(¤t); + let emitting = tokio::select! { _=stop.notified()=>break,result=sink.emit(delivery.value())=>result }; + match emitting { + Ok(count) => { + options + .counters + .spans_emitted + .fetch_add(count, Ordering::Relaxed); + } + Err(error) => { + if let Some(partial) = + error.downcast_ref::() + { options .counters .spans_emitted - .fetch_add(count, Ordering::Relaxed); - *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); - 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() = None; - } - Err(failure) => { - *options.error.lock() = Some(format!( - "route completion held: {}", - failure.error - )); - completion = Some(failure.delivery); - held = true; - progress.send_replace(cursor.completed_position()); - } - } + .fetch_add(partial.emitted, Ordering::Relaxed); + if partial.report { + tracing::warn!(%error,session_id=%options.session_id,"route processing failed"); } } - 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); + state = State::Recovering( + Failure::routing( + &error, + cursor.completed_position() + 1, + ¤t, + attempts, + ), + Work::Resubmit, + ); + continue; + } + } + let flushing = + tokio::select! { _=stop.notified()=>break,result=sink.flush()=>result }; + if let Err(error) = flushing { + state = if error.downcast_ref::().is_some() { + let mut failure = Failure::storage( + &error, + cursor.completed_position() + 1, + &options.version, + attempts, + ); + failure.boundary = Boundary::Ledger; + State::Recovering(failure, Work::Flush) + } else { + State::Recovering( + Failure::routing( + &error, + cursor.completed_position() + 1, + ¤t, + attempts, + ), + Work::Resubmit, + ) + }; + continue; + } + *options.permalink.lock().unwrap() = sink.permalink(); + drop(delivery); + if !sink.has_pending_delivery() { + match cursor.pending_prefix().into_completed().await { + Ok(()) => { + if let Err(error) = cursor.rewind().await { + state = State::Recovering( + Failure::storage( + &error, + cursor.completed_position() + 1, + &options.version, + attempts, + ), + Work::Resubmit, + ); + continue; } - *options.error.lock() = - Some(format!("routing held: {error:#}")); - let failure = options.error.lock().clone(); - journal.set_failure(&cursor.route, failure).await?; - held = true; progress.send_replace(cursor.completed_position()); + *options.error.lock().unwrap() = None; + *options.recovery.lock().unwrap() = None; + attempts = 0; + recorded = None; + } + Err(unfinished) => { + let mut failure = Failure::storage( + &unfinished.error, + cursor.completed_position() + 1, + &options.version, + attempts, + ); + failure.boundary = Boundary::Completion; + state = State::Recovering( + failure, + Work::Complete(unfinished.delivery), + ); + continue; } } } - None => tokio::select! { - _=stop.notified()=>return Ok(()), - _=cursor.changed()=>{}, - result=settings.changed()=>if result.is_err() { return Ok(()); }, - }, + state = State::Processing; } } - }; - if let Err(error) = task.await { - *options.error.lock() = Some(format!("routing journal held: {error:#}")); - progress.send_replace(cursor.completed_position()); } }); *handle.task.lock() = Some(task); @@ -213,3 +506,249 @@ impl Consumer { } } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ + translate::{SpanOp, SpanRow}, + wire::BackendAuth, + }; + use std::sync::atomic::{AtomicBool, AtomicU64}; + + #[derive(Default)] + struct Recording { + submitted: AtomicU64, + accepted: AtomicU64, + reject: AtomicBool, + unauthorized: AtomicBool, + } + struct Factory(Arc); + struct CheckedSink { + record: Arc, + pending: u64, + } + impl SinkFactory for Factory { + fn create(&self, _: &str, _: &str) -> anyhow::Result> { + Ok(Box::new(CheckedSink { + record: self.0.clone(), + pending: 0, + })) + } + } + #[async_trait::async_trait] + impl Sink for CheckedSink { + async fn emit(&mut self, ops: &[SpanOp]) -> anyhow::Result { + self.pending += ops.len() as u64; + self.record + .submitted + .fetch_add(ops.len() as u64, Ordering::SeqCst); + Ok(ops.len() as u64) + } + async fn flush(&mut self) -> anyhow::Result<()> { + // The SDK consumes its pending batch even when delivery is rejected. + let pending = std::mem::take(&mut self.pending); + if self.record.reject.swap(false, Ordering::SeqCst) { + return Err(braintrust_sdk_rust::BraintrustError::Api { + status: if self.record.unauthorized.load(Ordering::SeqCst) { + 401 + } else { + 503 + }, + message: "unavailable".into(), + } + .into()); + } + self.record.accepted.fetch_add(pending, Ordering::SeqCst); + Ok(()) + } + } + fn options( + dir: &std::path::Path, + route: &SessionRoute, + record: Arc, + ) -> ConsumerOptions { + ConsumerOptions { + source: "debug".into(), + session_id: "test".into(), + version: "test".into(), + data_dir: dir.into(), + config: route.with_auth(BackendAuth { + token: "never-persist-this".into(), + api_url: None, + app_url: None, + org_id: None, + org_name: None, + }), + counters: Arc::new(Counters::default()), + error: Arc::new(Mutex::new(None)), + recovery: Arc::new(Mutex::new(None)), + permalink: Arc::new(Mutex::new(None)), + factory: Arc::new(Factory(record)), + auth_provider: None, + } + } + async fn wait_for(mut predicate: impl FnMut() -> bool) { + tokio::time::timeout(std::time::Duration::from_secs(5), async { + while !predicate() { + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }) + .await + .unwrap(); + } + async fn stored(journal: &SpanJournal) { + journal.begin(10, true).await.unwrap(); + journal + .append( + 10, + &[SpanOp::Insert(SpanRow { + span_id: "one".into(), + name: "one".into(), + root_span_id: "one".into(), + end_ms: Some(1), + ..Default::default() + })], + ) + .await + .unwrap(); + journal.commit(10).await.unwrap(); + } + #[tokio::test] + async fn transient_rejection_resubmits_consumed_batch_before_completing() { + let tmp = tempfile::tempdir().unwrap(); + let journal = SpanJournal::open(&tmp.path().join("spans.sqlite")) + .await + .unwrap(); + stored(&journal).await; + let route = SessionRoute::default(); + let record = Arc::new(Recording::default()); + record.reject.store(true, Ordering::SeqCst); + let cursor = journal.subscribe(&route, 0).await.unwrap(); + let consumer = Consumer::spawn( + journal.clone(), + cursor, + options(tmp.path(), &route, record.clone()), + ); + wait_for(|| record.accepted.load(Ordering::SeqCst) == 1).await; + wait_for(|| *consumer.progress.borrow() == 1).await; + assert_eq!(record.submitted.load(Ordering::SeqCst), 2); + assert!(consumer.error.lock().unwrap().is_none()); + assert!(journal + .failure(&serde_json::to_string(&route).unwrap()) + .await + .unwrap() + .is_none()); + consumer.shutdown().await; + } + #[tokio::test] + async fn head_write_recovery_retries_only_completion() { + let tmp = tempfile::tempdir().unwrap(); + let path = tmp.path().join("spans.sqlite"); + let journal = SpanJournal::open(&path).await.unwrap(); + stored(&journal).await; + let route = SessionRoute::default(); + let cursor = journal.subscribe(&route, 0).await.unwrap(); + let db = rusqlite::Connection::open(&path).unwrap(); + db.execute_batch("CREATE TRIGGER fail_head BEFORE UPDATE OF position ON heads BEGIN SELECT RAISE(FAIL,'injected head failure'); END;").unwrap(); + let record = Arc::new(Recording::default()); + let consumer = Consumer::spawn( + journal.clone(), + cursor, + options(tmp.path(), &route, record.clone()), + ); + wait_for(|| consumer.error.lock().unwrap().is_some()).await; + assert_eq!(*consumer.progress.borrow(), 0); + assert_eq!(record.accepted.load(Ordering::SeqCst), 1); + db.execute_batch("DROP TRIGGER fail_head").unwrap(); + wait_for(|| *consumer.progress.borrow() == 1).await; + assert_eq!(record.submitted.load(Ordering::SeqCst), 1); + assert_eq!(record.accepted.load(Ordering::SeqCst), 1); + consumer.shutdown().await; + } + #[tokio::test] + async fn plugin_hold_survives_restart_and_file_edit_recovers_without_native_input() { + let tmp = tempfile::tempdir().unwrap(); + let plugin = tmp.path().join("plugin.mjs"); + std::fs::write(&plugin, "export default () => {throw new Error('bad')}").unwrap(); + let journal = SpanJournal::open(&tmp.path().join("spans.sqlite")) + .await + .unwrap(); + stored(&journal).await; + let route = SessionRoute { + span_plugins: vec![plugin.clone()], + ..Default::default() + }; + let key = serde_json::to_string(&route).unwrap(); + let record = Arc::new(Recording::default()); + let consumer = Consumer::spawn( + journal.clone(), + journal.subscribe(&route, 0).await.unwrap(), + options(tmp.path(), &route, record.clone()), + ); + wait_for(|| consumer.error.lock().unwrap().is_some()).await; + consumer.shutdown().await; + let evidence = journal.failure(&key).await.unwrap().unwrap(); + assert!(!evidence.contains("never-persist-this")); + let consumer = Consumer::spawn( + journal.clone(), + journal.subscribe(&route, 0).await.unwrap(), + options(tmp.path(), &route, record.clone()), + ); + tokio::time::sleep(std::time::Duration::from_millis(600)).await; + assert_eq!(record.submitted.load(Ordering::SeqCst), 0); + std::fs::write(&plugin, "export default span => span").unwrap(); + wait_for(|| *consumer.progress.borrow() == 1).await; + assert_eq!(record.accepted.load(Ordering::SeqCst), 1); + assert!(journal.failure(&key).await.unwrap().is_none()); + consumer.shutdown().await; + } + struct RefreshAuth(Arc); + #[async_trait::async_trait] + impl AuthProvider for RefreshAuth { + async fn resolve( + &self, + selection: &crate::wire::AuthSelection, + reason: AuthResolveReason, + ) -> anyhow::Result { + assert_eq!(reason, AuthResolveReason::Unauthorized); + self.0.fetch_add(1, Ordering::SeqCst); + Ok(crate::AuthLease { + selection: selection.clone(), + auth: BackendAuth { + token: "fresh".into(), + api_url: None, + app_url: None, + org_id: None, + org_name: None, + }, + expires_at_ms: None, + }) + } + } + #[tokio::test] + async fn unauthorized_delivery_refreshes_credentials_at_the_same_head() { + let tmp = tempfile::tempdir().unwrap(); + let journal = SpanJournal::open(&tmp.path().join("spans.sqlite")) + .await + .unwrap(); + stored(&journal).await; + let route = SessionRoute::default(); + let record = Arc::new(Recording::default()); + record.reject.store(true, Ordering::SeqCst); + record.unauthorized.store(true, Ordering::SeqCst); + let refreshed = Arc::new(AtomicU64::new(0)); + let mut opts = options(tmp.path(), &route, record.clone()); + opts.auth_provider = Some(Arc::new(RefreshAuth(refreshed.clone()))); + let consumer = Consumer::spawn( + journal.clone(), + journal.subscribe(&route, 0).await.unwrap(), + opts, + ); + wait_for(|| *consumer.progress.borrow() == 1).await; + assert_eq!(refreshed.load(Ordering::SeqCst), 1); + assert_eq!(consumer.config.borrow().auth.token, "fresh"); + assert_eq!(record.submitted.load(Ordering::SeqCst), 2); + consumer.shutdown().await; + } +} diff --git a/bt-daemon/src/pipeline/journal.rs b/bt-daemon/src/pipeline/journal.rs index 96b3f5ab..a9a12940 100644 --- a/bt-daemon/src/pipeline/journal.rs +++ b/bt-daemon/src/pipeline/journal.rs @@ -75,6 +75,7 @@ impl SpanJournal { CREATE TABLE IF NOT EXISTS heads ( route TEXT PRIMARY KEY, position INTEGER NOT NULL DEFAULT 0, failure TEXT); + CREATE TABLE IF NOT EXISTS recovery (id INTEGER PRIMARY KEY CHECK(id=1),failure TEXT); ", )?; if version == 1 { @@ -279,6 +280,28 @@ impl SpanJournal { }).await } + pub async fn source_failure(&self) -> anyhow::Result> { + self.with_db(|db| { + Ok(db + .query_row("SELECT failure FROM recovery WHERE id=1", [], |row| { + row.get(0) + }) + .optional()? + .flatten()) + }) + .await + } + pub async fn set_source_failure(&self, failure: Option) -> anyhow::Result<()> { + self.with_db(move |db| { + db.execute( + "INSERT OR REPLACE INTO recovery(id,failure) VALUES(1,?)", + [failure], + )?; + Ok(()) + }) + .await + } + pub async fn failure(&self, route: &str) -> anyhow::Result> { let route = route.to_owned(); self.with_db(move |db| { @@ -424,7 +447,7 @@ impl Completion for HeadCompletion { if head >= through { return Ok(()); } - anyhow::ensure!(head == expected, "completion cannot skip unresolved input"); + anyhow::ensure!(head >= expected, "completion cannot skip unresolved input"); tx.execute( "UPDATE heads SET position=?, failure=NULL WHERE route=?", params![through, route], diff --git a/bt-daemon/src/pipeline/mod.rs b/bt-daemon/src/pipeline/mod.rs index e8148719..19ee0f93 100644 --- a/bt-daemon/src/pipeline/mod.rs +++ b/bt-daemon/src/pipeline/mod.rs @@ -21,3 +21,5 @@ pub(crate) mod journal; pub(crate) mod translation; pub(crate) mod consumer; + +pub(crate) mod recovery; diff --git a/bt-daemon/src/pipeline/plugins.rs b/bt-daemon/src/pipeline/plugins.rs index ddce96b9..6f8143ce 100644 --- a/bt-daemon/src/pipeline/plugins.rs +++ b/bt-daemon/src/pipeline/plugins.rs @@ -6,7 +6,7 @@ use std::path::PathBuf; pub(crate) enum PluginOutcome { Prepared(Box), Filtered, - Failed(span_processor::PluginFailure), + Failed(PluginRejected), Unavailable(anyhow::Error), } @@ -14,6 +14,35 @@ pub(crate) struct Plugins { pub paths: Vec, pub source: String, pub session_id: String, + blocked: Option<(Vec, PluginRejected)>, +} + +#[derive(Debug, Clone)] +pub(crate) struct PluginRejected { + pub failure: crate::span_processor::PluginFailure, + pub digest: Option<[u8; 32]>, +} +impl std::fmt::Display for PluginRejected { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + f, + "span plugin {} failed; delivery is held: {}", + self.failure.path.display(), + self.failure.message + ) + } +} +impl std::error::Error for PluginRejected {} + +impl Plugins { + pub fn new(source: &str, session_id: &str) -> Self { + Self { + paths: Vec::new(), + source: source.into(), + session_id: session_id.into(), + blocked: None, + } + } } #[async_trait::async_trait] @@ -22,9 +51,24 @@ impl Stage for Plugins { type Outcome = PluginOutcome; async fn process(&mut self, input: &SpanOp) -> PluginOutcome { - match span_processor::process(&self.paths, input, &self.source, &self.session_id) { + if let Some((paths, failure)) = &self.blocked { + if paths == &self.paths + && span_processor::file_digest(&failure.failure.path) == failure.digest + { + let mut failure = failure.clone(); + failure.failure.newly_seen = false; + return PluginOutcome::Failed(failure); + } + } + self.blocked = None; + match span_processor::process_isolated(&self.paths, input, &self.source, &self.session_id) { Ok(result) => match (result.op, result.failure) { - (_, Some(failure)) => PluginOutcome::Failed(failure), + (_, Some(failure)) => { + let digest = span_processor::failed_digest(&failure.path); + let failure = PluginRejected { failure, digest }; + self.blocked = Some((self.paths.clone(), failure.clone())); + PluginOutcome::Failed(failure) + } (Some(op), None) => PluginOutcome::Prepared(Box::new(op)), (None, None) => PluginOutcome::Filtered, }, @@ -32,3 +76,59 @@ impl Stage for Plugins { } } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::translate::SpanRow; + #[tokio::test] + async fn failure_is_local_to_one_route_and_equal_stat_edit_reloads() { + let tmp = tempfile::tempdir().unwrap(); + let path = tmp.path().join("conditional.mjs"); + let failing = + "export default span => {if(span.name==='bad') throw new Error('bad'); return span}"; + std::fs::write(&path, failing).unwrap(); + let modified = std::fs::metadata(&path).unwrap().modified().unwrap(); + let mut bad = Plugins::new("debug", "bad"); + bad.paths = vec![path.clone()]; + let bad_op = SpanOp::Insert(SpanRow { + span_id: "one".into(), + name: "bad".into(), + ..Default::default() + }); + assert!(matches!( + bad.process(&bad_op).await, + PluginOutcome::Failed(_) + )); + let mut healthy = Plugins::new("debug", "healthy"); + healthy.paths = vec![path.clone()]; + assert!(matches!( + healthy + .process(&SpanOp::Insert(SpanRow { + name: "good".into(), + ..Default::default() + })) + .await, + PluginOutcome::Prepared(_) + )); + std::fs::write(&path, failing.replace("==='bad'", "==='foo'")).unwrap(); + std::fs::OpenOptions::new() + .write(true) + .open(&path) + .unwrap() + .set_times(std::fs::FileTimes::new().set_modified(modified)) + .unwrap(); + assert_eq!( + std::fs::metadata(&path).unwrap().modified().unwrap(), + modified + ); + assert_eq!( + std::fs::metadata(&path).unwrap().len(), + failing.len() as u64 + ); + assert!(matches!( + bad.process(&bad_op).await, + PluginOutcome::Prepared(_) + )); + } +} diff --git a/bt-daemon/src/pipeline/recovery.rs b/bt-daemon/src/pipeline/recovery.rs new file mode 100644 index 00000000..f0b43149 --- /dev/null +++ b/bt-daemon/src/pipeline/recovery.rs @@ -0,0 +1,277 @@ +//! Failure outcomes wait for explicit evidence before re-entering processing. +use super::{plugins::PluginRejected, Stage}; +use crate::wire::{RecoveryInfo, SessionConfig}; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use std::path::PathBuf; + +#[derive(Debug, Clone, Copy, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub(crate) enum Boundary { + Plugin, + Sink, + Credentials, + Translation, + Journal, + Ledger, + Completion, +} +impl std::fmt::Display for Boundary { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(match self { + Self::Plugin => "plugin", + Self::Sink => "sink", + Self::Credentials => "credentials", + Self::Translation => "translation", + Self::Journal => "journal", + Self::Ledger => "delivery_ledger", + Self::Completion => "completion", + }) + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "when", rename_all = "snake_case")] +pub(crate) enum Condition { + Plugin { + path: PathBuf, + digest: Option<[u8; 32]>, + chain: String, + }, + Configuration { + signature: String, + recheck_ms: Option, + }, + Credentials { + recheck_ms: i64, + }, + Delay { + recheck_ms: i64, + }, + Revision { + version: String, + }, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(crate) struct Failure { + pub boundary: Boundary, + pub position: u64, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub through: Option, + pub reason: String, + pub attempt: u32, + pub condition: Condition, +} + +pub(crate) fn now_ms() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis() + .min(i64::MAX as u128) as i64 +} +fn digest(value: &impl Serialize) -> String { + format!( + "{:x}", + Sha256::digest(serde_json::to_vec(value).expect("serializable condition evidence")) + ) +} +fn chain(config: &SessionConfig) -> String { + digest(&config.span_plugins) +} +fn configuration(config: &SessionConfig) -> String { + let processors: Vec<_> = config + .span_plugins + .iter() + .map(|path| (path, crate::span_processor::file_digest(path))) + .collect(); + digest( + &serde_json::json!({"destination":config.destination,"metadata":config.additional_metadata, + "tags":config.tags,"processors":processors,"api_url":config.auth.api_url,"app_url":config.auth.app_url, + "org_name":config.auth.org_name,"org_id":config.auth.org_id}), + ) +} + +impl Failure { + pub fn delay(boundary: Boundary, position: u64, reason: String, attempt: u32) -> Self { + let delay = 250i64.saturating_mul(1i64 << attempt.min(8)).min(30_000); + Self { + boundary, + position, + through: None, + reason, + attempt, + condition: Condition::Delay { + recheck_ms: now_ms().saturating_add(delay), + }, + } + } + pub fn revision(boundary: Boundary, position: u64, reason: String, version: &str) -> Self { + Self { + boundary, + position, + through: None, + reason, + attempt: 0, + condition: Condition::Revision { + version: version.into(), + }, + } + } + pub fn routing( + error: &anyhow::Error, + position: u64, + config: &SessionConfig, + attempt: u32, + ) -> Self { + let error = error + .downcast_ref::() + .map_or(error, |failure| &failure.cause); + let mut failure = Self::delay(Boundary::Sink, position, format!("{error:#}"), attempt); + if let Some(plugin) = error.downcast_ref::() { + failure.boundary = Boundary::Plugin; + failure.condition = Condition::Plugin { + path: plugin.failure.path.clone(), + digest: plugin.digest, + chain: chain(config), + }; + } else if let Some(error) = error.downcast_ref::() { + use braintrust_sdk_rust::BraintrustError; + match error { + BraintrustError::Api { status: 401, .. } => { + failure.boundary = Boundary::Credentials; + failure.condition = Condition::Credentials { + recheck_ms: now_ms().saturating_add(1_000), + }; + } + BraintrustError::Api { + status: 403 | 404, .. + } => { + failure.condition = Condition::Configuration { + signature: configuration(config), + recheck_ms: Some(now_ms().saturating_add(60_000)), + } + } + BraintrustError::Api { + status: 408 | 429 | 500..=599, + .. + } + | BraintrustError::Http(_) + | BraintrustError::Network(_) + | BraintrustError::ChannelClosed => {} + _ => { + failure.condition = Condition::Configuration { + signature: configuration(config), + recheck_ms: None, + } + } + } + } + if !config.auth.token.is_empty() { + failure.reason = failure.reason.replace(&config.auth.token, ""); + } + failure + } + pub fn credentials(position: u64, reason: String, attempt: u32) -> Self { + let mut failure = Self::delay(Boundary::Credentials, position, reason, attempt); + if let Condition::Delay { recheck_ms } = failure.condition { + failure.condition = Condition::Credentials { recheck_ms }; + } + failure + } + pub fn legacy(reason: String, position: u64, config: &SessionConfig) -> Self { + Self { + boundary: Boundary::Sink, + position, + through: None, + reason, + attempt: 0, + condition: Condition::Configuration { + signature: configuration(config), + recheck_ms: None, + }, + } + } + pub fn storage(error: &anyhow::Error, position: u64, version: &str, attempt: u32) -> Self { + if error.downcast_ref::().is_some() || error.downcast_ref::().is_some_and(|error| { + matches!(error,rusqlite::Error::SqliteFailure(code,_) if matches!(code.code,rusqlite::ErrorCode::DatabaseCorrupt|rusqlite::ErrorCode::NotADatabase)) + }) { + Self::revision(Boundary::Journal,position,error.to_string(),version) + } else { Self::delay(Boundary::Journal,position,error.to_string(),attempt) } + } + pub fn info(&self, eligible: bool) -> RecoveryInfo { + let retry = match &self.condition { + Condition::Plugin { path, .. } => format!( + "bytes/existence of {} or configured plugin chain changes", + path.display() + ), + Condition::Configuration { recheck_ms, .. } => match recheck_ms { + Some(when) => format!("relevant settings/processors change or recheck at {when}"), + None => "relevant settings or processors change".into(), + }, + Condition::Credentials { recheck_ms } => { + format!("fresh credential resolution at {recheck_ms}") + } + Condition::Delay { recheck_ms } => format!("bounded retry at {recheck_ms}"), + Condition::Revision { version } => format!("processor revision changes from {version}"), + }; + RecoveryInfo { + stage: self.boundary.to_string(), + position: self.position, + reason: self.reason.clone(), + retry, + eligible, + } + } + pub fn message(&self) -> String { + let info = self.info(false); + format!( + "recovering {} at {}: {}; waiting for {}", + info.stage, info.position, info.reason, info.retry + ) + } +} + +pub(crate) enum RecoveryOutcome { + Waiting, + Retry, +} +pub(crate) struct Recovery<'a> { + pub config: Option<&'a SessionConfig>, + pub version: &'a str, +} +#[async_trait::async_trait] +impl Stage for Recovery<'_> { + type Input = Failure; + type Outcome = RecoveryOutcome; + async fn process(&mut self, failure: &Failure) -> RecoveryOutcome { + let ready = match &failure.condition { + Condition::Plugin { + path, + digest, + chain: previous, + } => { + self.config.is_some_and(|config| chain(config) != *previous) + || crate::span_processor::file_digest(path) != *digest + } + Condition::Configuration { + signature, + recheck_ms, + } => { + self.config + .is_some_and(|config| configuration(config) != *signature) + || recheck_ms.is_some_and(|when| now_ms() >= when) + } + Condition::Credentials { recheck_ms } | Condition::Delay { recheck_ms } => { + now_ms() >= *recheck_ms + } + Condition::Revision { version } => self.version != version, + }; + if ready { + RecoveryOutcome::Retry + } else { + RecoveryOutcome::Waiting + } + } +} diff --git a/bt-daemon/src/pipeline/routing.rs b/bt-daemon/src/pipeline/routing.rs index daeb2040..6ddb1ee0 100644 --- a/bt-daemon/src/pipeline/routing.rs +++ b/bt-daemon/src/pipeline/routing.rs @@ -68,6 +68,7 @@ impl RoutingSink { ) -> Self { let (tx, mut rx) = mpsc::channel::>(16); tokio::spawn(async move { + let mut stage = Plugins::new(&source, &session_id); let mut prepare = super::prepare::Prepare::new(&source); while let Ok(Some(delivery)) = InputPort::receive(&mut rx).await { let request = delivery.value(); @@ -80,11 +81,7 @@ impl RoutingSink { Operation::Emit(ops) => { let mut ops = ops.clone(); prepare.apply(&mut ops, &request.config); - let mut stage = Plugins { - paths: request.config.span_plugins.clone(), - source: source.clone(), - session_id: session_id.clone(), - }; + stage.paths = request.config.span_plugins.clone(); let mut prepared = Vec::with_capacity(ops.len()); let mut failure = None; let mut report = false; @@ -93,20 +90,18 @@ impl RoutingSink { PluginOutcome::Prepared(op) => prepared.push(*op), PluginOutcome::Filtered => {} PluginOutcome::Failed(error) => { - report |= error.newly_seen; - if error.newly_seen { + report |= error.failure.newly_seen; + if error.failure.newly_seen { if let Err(diagnostic) = crate::plugin_diagnostics::record( &data_dir, &source, - &error.path, - &error.message, + &error.failure.path, + &error.failure.message, ) { tracing::warn!(%diagnostic, "plugin diagnostic could not be saved"); } } - failure = Some(anyhow::anyhow!( - "span plugin {} failed; span operations are being discarded: {}", - error.path.display(), error.message)); + failure = Some(anyhow::Error::new(error)); } PluginOutcome::Unavailable(error) => { report = true; diff --git a/bt-daemon/src/pipeline/translation.rs b/bt-daemon/src/pipeline/translation.rs index bf032fd9..89c37a56 100644 --- a/bt-daemon/src/pipeline/translation.rs +++ b/bt-daemon/src/pipeline/translation.rs @@ -1,6 +1,10 @@ //! One ordered source consumer. Output publication is independent of route delivery. use super::journal::SpanJournal; use super::prepare::Prepare; +use super::{ + recovery::{Boundary, Failure, Recovery, RecoveryOutcome}, + Stage, +}; use crate::{ journal::{ControlKind, JournalReader, JournalRecord, JournalWriter, PipelineControl}, translate::{AgentTranslator, OriginSnapshot, Registry, SessionCtx, SpanOp}, @@ -47,6 +51,7 @@ pub(crate) struct Translation { task: Mutex>>, pub journal: SpanJournal, pub error: Arc>>, + pub recovery: Arc>>, pub routes: AtomicUsize, pub progress: Arc, } @@ -76,13 +81,21 @@ impl Translation { requested: AtomicU64::new(committed), committed: AtomicU64::new(committed), }); + let recovery = Arc::new(Mutex::new(None)); let error = Arc::new(Mutex::new(None)); let (tx, rx) = mpsc::channel(64); + let state = match journal.source_failure().await? { + Some(raw) => State::Recovering(serde_json::from_str(&raw).unwrap_or_else(|_| { + Failure::revision(Boundary::Journal, committed + 1, raw, &options.version) + })), + None => State::Processing, + }; let actor = Producer { options, progress: progress.clone(), journal: journal.clone(), error: error.clone(), + recovery: recovery.clone(), translator: None, reader: None, observers: Vec::new(), @@ -91,7 +104,9 @@ impl Translation { bounds: HashMap::new(), position: 0, committed, - held: false, + state, + publication: None, + phase: Boundary::Journal, }; let task = tokio::spawn(actor.run(rx)); Ok(Arc::new(Self { @@ -99,6 +114,7 @@ impl Translation { task: Mutex::new(Some(task)), journal, error, + recovery, routes: AtomicUsize::new(0), progress, })) @@ -146,11 +162,25 @@ impl Translation { } } +enum State { + Processing, + Recovering(Failure), +} +struct Publication { + input: u64, + observation: bool, + capture: Option, + ops: Option>, + begun: bool, + complete: bool, +} + struct Producer { options: TranslationOptions, progress: Arc, journal: SpanJournal, error: Arc>>, + recovery: Arc>>, translator: Option>, reader: Option, observers: Vec, @@ -159,15 +189,97 @@ struct Producer { bounds: HashMap, position: u64, committed: u64, - held: bool, + state: State, + publication: Option, + phase: Boundary, } impl Producer { async fn run(mut self, mut rx: mpsc::Receiver) { - while let Some(command) = rx.recv().await { + let mut recorded = None; + let mut record_retry = 0; + let mut attempts = 0; + loop { + if let State::Recovering(mut failure) = self.state { + *self.error.lock().unwrap() = Some(failure.message()); + *self.recovery.lock().unwrap() = Some(failure.info(false)); + let encoded = serde_json::to_string(&failure).expect("source failure evidence"); + if recorded.as_ref() != Some(&encoded) && super::recovery::now_ms() >= record_retry + { + match self.journal.set_source_failure(Some(encoded.clone())).await { + Ok(()) => recorded = Some(encoded.clone()), + Err(error) => { + record_retry = super::recovery::now_ms().saturating_add(1_000); + tracing::warn!(%error,"source recovery evidence could not be persisted; original failure retained"); + } + } + } + self.state = State::Recovering(failure.clone()); + tokio::select! { + command=rx.recv()=>match command { + Some(Command::Stop)|None=>break, + Some(Command::Barrier(reply))=> { let _=reply.send(self.boundary().await); }, + Some(Command::Observe(observer))=> { let _=self.observe_history(&observer).await; self.observers.push(*observer); }, + Some(Command::Advance(_))=>{}, + Some(Command::Control(kind,reply))=> { + let result=self.append_control(kind).await; + if let Err(error)=result { tracing::warn!(%error,"ordered control could not be persisted while source is recovering"); } + let _=reply.send(self.boundary().await); + }, + }, + _=tokio::time::sleep(std::time::Duration::from_millis(250))=>{}, + } + if recorded.as_ref() != Some(&encoded) + || matches!( + (Recovery { + config: None, + version: &self.options.version + }) + .process(&failure) + .await, + RecoveryOutcome::Waiting + ) + { + continue; + } + *self.recovery.lock().unwrap() = Some(failure.info(true)); + attempts = failure.attempt.saturating_add(1); + self.state = State::Processing; + let result = async { + if self.publication.is_some() { + self.publish().await?; + } else if self.progress.requested.load(Ordering::Relaxed) > self.committed { + self.reader = None; + self.translator = None; + self.position = 0; + } + self.advance(self.progress.requested.load(Ordering::Relaxed)) + .await + } + .await; + if let Err(error) = result { + failure = self.failure(error, attempts); + self.state = State::Recovering(failure); + continue; + } + if let Err(error) = self.journal.set_source_failure(None).await { + self.phase = Boundary::Journal; + self.state = State::Recovering(self.failure(error, attempts)); + continue; + } + *self.error.lock().unwrap() = None; + *self.recovery.lock().unwrap() = None; + recorded = None; + attempts = 0; + continue; + } + let Some(command) = rx.recv().await else { + break; + }; let result = match command { Command::Stop => break, Command::Observe(observer) => { + self.phase = Boundary::Journal; let result = self.observe_history(&observer).await; self.observers.push(*observer); result @@ -184,26 +296,48 @@ impl Producer { } }; if let Err(error) = result { - self.held = true; - *self.error.lock() = - Some(format!("translation held at {}: {error:#}", self.position)); + self.state = State::Recovering(self.failure(error, attempts)); } } } + fn failure(&self, error: anyhow::Error, attempt: u32) -> Failure { + let position = self.committed.saturating_add(1); + let mut failure = if matches!(self.phase, Boundary::Translation) { + Failure::revision( + self.phase, + position, + error.to_string(), + &self.options.version, + ) + } else { + Failure::storage(&error, position, &self.options.version, attempt) + }; + failure.through = self + .publication + .as_ref() + .map(|publication| publication.input); + failure + } async fn boundary(&mut self) -> Result { - if self.held { - return Err(self - .error - .lock() - .clone() - .unwrap_or_else(|| "translation held".into())); + if let State::Recovering(failure) = &self.state { + return Err(failure.message()); } 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"); + anyhow::ensure!( + matches!(self.state, State::Processing), + "translation is recovering" + ); + self.append_control(kind).await?; + self.advance(self.progress.requested.load(Ordering::Relaxed)) + .await?; + self.journal.tail().await + } + async fn append_control(&mut self, kind: ControlKind) -> anyhow::Result<()> { + self.phase = Boundary::Journal; let mut bounds = HashMap::new(); for path in self.bounds.keys() { let through = tokio::fs::metadata(path) @@ -223,8 +357,7 @@ impl Producer { self.progress .requested .fetch_max(through, Ordering::Relaxed); - self.advance(through).await?; - self.journal.tail().await + Ok(()) } async fn initialize(&mut self) -> anyhow::Result<()> { @@ -260,10 +393,14 @@ impl Producer { } async fn advance(&mut self, through: u64) -> anyhow::Result<()> { - anyhow::ensure!(!self.held, "translation is held"); + anyhow::ensure!( + matches!(self.state, State::Processing), + "translation is recovering" + ); if self.translator.is_none() && through <= self.committed { return Ok(()); } + self.phase = Boundary::Journal; self.initialize().await?; let through = through.max(self.committed); if self.position >= through { @@ -274,6 +411,7 @@ impl Producer { .as_mut() .ok_or_else(|| anyhow::anyhow!("event journal is missing"))?; reader.extend(through); + self.phase = Boundary::Journal; while let Some(entry) = self .reader .as_mut() @@ -318,6 +456,7 @@ impl Producer { bt_version: self.options.version.clone(), }; self.capture = Some(capture.clone()); + self.phase = Boundary::Translation; ( self.translator .as_mut() @@ -333,6 +472,7 @@ impl Producer { self.position = input; continue; } + self.phase = Boundary::Translation; let translator = self .translator .as_mut() @@ -355,48 +495,91 @@ impl Producer { continue; } }; - let publish = input > self.committed; - if publish { - self.journal.begin(input, observation).await?; + self.publication = Some(Publication { + input, + observation, + capture, + ops: Some(first), + begun: false, + complete: false, + }); + self.publish().await?; + self.phase = Boundary::Journal; + self.position = input; + } + Ok(()) + } + + /// Each chunk stays owned until its local write succeeds. A commit retry + /// cannot call translation or advance to the following event. + async fn publish(&mut self) -> anyhow::Result<()> { + let mut pending = self.publication.take().expect("pending publication"); + let result = async { + let publish = pending.input > self.committed; + let ctx = SessionCtx { + session_id: self.options.session_id.clone(), + config: None, + }; + if publish && !pending.begun { + self.phase = Boundary::Journal; + self.journal + .begin(pending.input, pending.observation) + .await?; + pending.begun = true; } - 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 { - 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(), - }) - }); + while !pending.complete { + if let Some(ops) = pending.ops.as_mut() { + for op in ops.iter_mut() { + 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) = &pending.capture { + self.origins + .entry(row.span_id.clone()) + .or_insert_with(|| capture.clone()); + } + } + row.origin = self.origins.get(&row.span_id).cloned(); } - row.origin = self.origins.get(&row.span_id).cloned(); - } - if observation { - for observer in &self.observers { - Self::observe(&self.options, &ops, observer).await?; + self.phase = Boundary::Journal; + if pending.observation { + for observer in &self.observers { + Self::observe(&self.options, ops, observer).await?; + } } + if publish { + self.journal.append(pending.input, ops).await?; + } + pending.ops = None; } - if publish { - self.journal.append(input, &ops).await?; - } - next = self + self.phase = Boundary::Translation; + pending.ops = self .translator .as_mut() .ok_or_else(|| anyhow::anyhow!("translator is missing"))? .drain_pending(&ctx)?; + pending.complete = pending.ops.is_none(); } if publish { - self.journal.commit(input).await?; - self.committed = input; - self.progress.committed.store(input, Ordering::Relaxed); + self.phase = Boundary::Journal; + self.journal.commit(pending.input).await?; + self.committed = pending.input; + self.progress + .committed + .store(pending.input, Ordering::Relaxed); } - self.position = input; + self.position = pending.input; + Ok(()) } - Ok(()) + .await; + if result.is_err() { + self.publication = Some(pending); + } + result } async fn observe_history(&mut self, observer: &Observation) -> anyhow::Result<()> { @@ -503,4 +686,80 @@ mod tests { route.rewind().await.unwrap(); assert_eq!(route.completed_position(), tail); } + struct CountedTranslation(Arc); + impl AgentTranslator for CountedTranslation { + fn handle( + &mut self, + _: &crate::wire::Envelope, + _: &SessionCtx, + ) -> anyhow::Result> { + panic!("publication retry must not translate the event again") + } + fn drain_pending(&mut self, _: &SessionCtx) -> anyhow::Result>> { + self.0.fetch_add(1, Ordering::Relaxed); + Ok(None) + } + } + #[tokio::test] + async fn failed_append_and_commit_retain_the_built_chunk() { + let tmp = tempfile::tempdir().unwrap(); + let path = tmp.path().join("spans.sqlite"); + let journal = SpanJournal::open(&path).await.unwrap(); + let db = rusqlite::Connection::open(&path).unwrap(); + let event_path = tmp.path().join("events.ndjson"); + let drains = Arc::new(AtomicU64::new(0)); + let mut producer = Producer { + options: TranslationOptions { + source: "debug".into(), + session_id: "test".into(), + namespace: "test".into(), + 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()), + }, + progress: Arc::new(Progress::default()), + journal: journal.clone(), + error: Arc::new(Mutex::new(None)), + recovery: Arc::new(Mutex::new(None)), + translator: Some(Box::new(CountedTranslation(drains.clone()))), + reader: None, + observers: Vec::new(), + origins: HashMap::new(), + bounds: HashMap::new(), + position: 0, + committed: 0, + state: State::Processing, + phase: Boundary::Journal, + publication: Some(Publication { + input: 10, + observation: true, + capture: None, + ops: Some(vec![SpanOp::Insert(SpanRow { + span_id: "built".into(), + ..Default::default() + })]), + begun: false, + complete: false, + }), + }; + db.execute_batch("CREATE TRIGGER fail_append BEFORE INSERT ON ops BEGIN SELECT RAISE(FAIL,'append'); END;").unwrap(); + assert!(producer.publish().await.is_err()); + assert_eq!(drains.load(Ordering::Relaxed), 0); + assert_eq!(journal.translated_through().await.unwrap(), 0); + db.execute_batch("DROP TRIGGER fail_append; CREATE TRIGGER fail_commit BEFORE UPDATE OF committed ON batches BEGIN SELECT RAISE(FAIL,'commit'); END;").unwrap(); + assert!(producer.publish().await.is_err()); + assert_eq!(drains.load(Ordering::Relaxed), 1); + assert_eq!(journal.tail().await.unwrap(), 0); + db.execute_batch("DROP TRIGGER fail_commit").unwrap(); + producer.publish().await.unwrap(); + assert_eq!(drains.load(Ordering::Relaxed), 1); + assert_eq!(journal.translated_through().await.unwrap(), 10); + assert_eq!(journal.tail().await.unwrap(), 1); + } } diff --git a/bt-daemon/src/server/correlation_store.rs b/bt-daemon/src/server/correlation_store.rs index ad193e68..c6520eef 100644 --- a/bt-daemon/src/server/correlation_store.rs +++ b/bt-daemon/src/server/correlation_store.rs @@ -28,10 +28,8 @@ pub(super) struct PendingSession { pub(super) first_seen_ms: i64, #[serde(skip)] pub(super) deadline: Option, - // Discovery expiry is not an authentication/delivery retry loop. A failed - // handoff stays visible until a new event, explicit flush, or restart. - #[serde(skip)] - pub(super) delivery_failed: bool, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub(super) recovery: Option, #[serde(default)] pub(super) events: Vec, } @@ -45,7 +43,7 @@ impl Default for PendingSession { parent: None, first_seen_ms: 0, deadline: None, - delivery_failed: false, + recovery: None, events: Vec::new(), } } @@ -150,6 +148,9 @@ pub(super) async fn accept_event(daemon: &Arc, event: PendingEvent) -> R // Queue behind any earlier events whose handoff failed so a // second failure keeps this event visible and retryable too. state.events.push(event); + if state.recovery.is_some() { + return store_correlation(daemon, &key, &state).await; + } return finish_correlation(daemon, &key, state).await; } @@ -328,8 +329,11 @@ pub(super) async fn finish_correlation( mut state: PendingSession, ) -> Result<(), String> { state.parent = None; - state.delivery_failed = false; - let result = async { + let attempt = state + .recovery + .take() + .map_or(0, |failure| failure.attempt.saturating_add(1)); + let result: Result<(), String> = async { // Persist the final route before delivering any events. A crash may // replay deterministic rows, but must never select a different parent. store_correlation(daemon, key, &state).await?; @@ -345,8 +349,19 @@ pub(super) async fn finish_correlation( Ok(()) } .await; - if result.is_err() { - state.delivery_failed = true; + if let Err(error) = &result { + let position = state + .events + .first() + .map_or(0, |event| event.journal_through); + state.recovery = Some(crate::pipeline::recovery::Failure::credentials( + position, + error.clone(), + attempt, + )); + if let Err(error) = store_correlation(daemon, key, &state).await { + tracing::warn!(%error,"pending recovery evidence could not be persisted"); + } daemon .pending_sessions .lock() @@ -373,8 +388,22 @@ pub(super) async fn resolve_pending_sessions( continue; }; let forced = state.events.first().is_some_and(|event| force(&event.env)); - if state.delivery_failed && !forced { - continue; + if let Some(failure) = &state.recovery { + use crate::pipeline::{ + recovery::{Recovery, RecoveryOutcome}, + Stage, + }; + if matches!( + (Recovery { + config: None, + version: &daemon.version + }) + .process(failure) + .await, + RecoveryOutcome::Waiting + ) { + continue; + } } if !state.decided() && (state.expired() || forced) { state.standalone = true; diff --git a/bt-daemon/src/server/ingress.rs b/bt-daemon/src/server/ingress.rs index 7f16cbd5..cda3162d 100644 --- a/bt-daemon/src/server/ingress.rs +++ b/bt-daemon/src/server/ingress.rs @@ -73,6 +73,7 @@ pub(super) async fn dispatch_ingress_event(daemon: &Arc, event: PendingE { daemon.settle_session_actors().await; } + let source_env = event.env.clone(); match accept_event(daemon, event).await { Ok(()) => { if let Some(key) = ingress_key { @@ -85,6 +86,17 @@ pub(super) async fn dispatch_ingress_event(daemon: &Arc, event: PendingE tracing::warn!(%error, "journaled ingress event could not be dispatched"); } } + // Authentication and route discovery do not prevent canonical persistence. + match daemon.translation_for(&source_env).await { + Ok(source) => { + if let Err(error) = source.advance(journal_through).await { + tracing::warn!(%error,"source processing could not be queued"); + } + } + Err(error) => { + tracing::warn!(%error,"source journal could not be opened; captured events retained") + } + } } /// Drain events omitted from the bounded in-memory queue. The queue contains diff --git a/bt-daemon/src/server/mod.rs b/bt-daemon/src/server/mod.rs index 8bd27b84..64284699 100644 --- a/bt-daemon/src/server/mod.rs +++ b/bt-daemon/src/server/mod.rs @@ -22,7 +22,7 @@ use recovery::*; use rpc::*; use sessions::*; -use crate::dispatch::{hydrate_transcript_reference, ReplayPlan, Session, SessionOptions}; +use crate::dispatch::{hydrate_transcript_reference, Session, SessionOptions}; use crate::journal::{self, JournalWriter}; use crate::sink::{BraintrustSinkConfig, BraintrustSinkFactory, DebugSinkFactory, SinkFactory}; use crate::translate::Registry; @@ -187,16 +187,13 @@ impl Daemon { *self.last_activity.lock() = Instant::now(); } - /// Queued work, excluding pending sessions whose handoff failed. Those are - /// durable on disk and resume after a restart, so they must not keep an - /// otherwise idle daemon alive forever. + /// Unresolved work keeps automatic recovery alive. fn total_queued(&self) -> u64 { let dispatched: u64 = self.sessions.lock().values().map(|s| s.pending()).sum(); dispatched.saturating_add( self.pending_sessions .lock() .values() - .filter(|state| !state.delivery_failed) .map(|state| state.events.len() as u64) .sum(), ) diff --git a/bt-daemon/src/server/rpc.rs b/bt-daemon/src/server/rpc.rs index 84c7b531..0738a921 100644 --- a/bt-daemon/src/server/rpc.rs +++ b/bt-daemon/src/server/rpc.rs @@ -335,6 +335,12 @@ impl Daemon { source: s.source.clone(), route: serde_json::from_str(&key.route).ok(), queued: s.pending(), + recovery: s + .recovery + .lock() + .unwrap() + .clone() + .or_else(|| s.source_recovery.lock().unwrap().clone()), spans_emitted: s.counters.spans_emitted.load(Ordering::Relaxed), permalink: s.permalink.lock().clone(), last_error: s @@ -359,6 +365,7 @@ impl Daemon { source: source.clone(), route: serde_json::from_str(&key.route).ok(), queued: 0, + recovery: None, spans_emitted: 0, permalink: None, last_error: Some(error.clone()), @@ -390,6 +397,14 @@ impl Daemon { "awaiting parent process confirmation; standalone fallback within 5 seconds" }; if let Some(session) = existing { + session.recovery = state + .recovery + .as_ref() + .map(|failure| failure.info(false)) + .or(session.recovery.take()); + if let Some(failure) = &state.recovery { + session.last_error = Some(failure.message()); + } session.queued = session.queued.saturating_add(state.events.len() as u64); if session.last_error.is_none() { session.last_error = Some(message.to_string()); @@ -400,9 +415,15 @@ impl Daemon { source: first.env.source.clone(), route: route.cloned(), queued: state.events.len() as u64, + recovery: state.recovery.as_ref().map(|failure| failure.info(false)), spans_emitted: 0, permalink: None, - last_error: Some(message.to_string()), + last_error: Some( + state + .recovery + .as_ref() + .map_or_else(|| message.to_string(), |failure| failure.message()), + ), }); } } diff --git a/bt-daemon/src/server/sessions.rs b/bt-daemon/src/server/sessions.rs index 22f2d2e5..6f9a3479 100644 --- a/bt-daemon/src/server/sessions.rs +++ b/bt-daemon/src/server/sessions.rs @@ -33,34 +33,13 @@ impl DeliveryKey { } impl Daemon { - pub(super) async fn session_for( + pub(super) async fn translation_for( &self, env: &Envelope, - key: &DeliveryKey, - replay_through: u64, - ) -> anyhow::Result> { - { - let map = self.sessions.lock(); - if let Some(session) = map.get(key) { - return Ok(session.clone()); - } - } - let route = env - .route - .as_ref() - .ok_or_else(|| anyhow::anyhow!("event is missing its session route"))?; - let config = env.config.clone().ok_or_else(|| { - anyhow::anyhow!("resolved route is missing its session configuration") - })?; - // The actor streams the journal itself, so creating a session stays - // cheap and allocation-free here no matter how long the recorded - // session is. Bound it to what is recorded now, before this event is - // appended, so replay covers prior source observations only. Every - // destination consumes that shared observation stream, with its own - // delivery checkpoint controlling which rows reach its sink. - let journal_path = + ) -> anyhow::Result> { + let event_path = journal::ensure_source_journal(&self.data_dir, &env.source, &env.session_id).await?; - let translator_session_id = + let namespace = if journal::legacy_journal_has_session(&self.data_dir, &env.source, &env.session_id) .await { @@ -68,32 +47,23 @@ impl Daemon { } else { crate::ids::session_namespace(&env.source, &env.session_id) }; - let replay = ReplayPlan { - acknowledged_through: journal::JournalReader::acknowledged_through( - &journal_path, - replay_through, - route, - ) - .await, - journal_path, - }; let journal = self .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().get(&storage_key).cloned(); - let translation = match existing { + let existing = self.translations.lock().unwrap().get(&storage_key).cloned(); + Ok(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(), + namespace: namespace.clone(), version: self.version.clone(), data_dir: self.data_dir.clone(), events: journal.clone(), - event_path: replay.journal_path.clone(), + event_path: event_path.clone(), registry: self.translators.clone(), correlation: self.correlation.clone(), correlation_changed: self.correlation_changed.clone(), @@ -102,10 +72,44 @@ impl Daemon { .await?; self.translations .lock() + .unwrap() .insert(storage_key, translation.clone()); translation } - }; + }) + } + + pub(super) async fn session_for( + &self, + env: &Envelope, + key: &DeliveryKey, + replay_through: u64, + ) -> anyhow::Result> { + { + let map = self.sessions.lock(); + if let Some(session) = map.get(key) { + return Ok(session.clone()); + } + } + let route = env + .route + .as_ref() + .ok_or_else(|| anyhow::anyhow!("event is missing its session route"))?; + let config = env.config.clone().ok_or_else(|| { + anyhow::anyhow!("resolved route is missing its session configuration") + })?; + // The actor streams the journal itself, so creating a session stays + // cheap and allocation-free here no matter how long the recorded + // session is. Bound it to what is recorded now, before this event is + // appended, so replay covers prior source observations only. Every + // destination consumes that shared observation stream, with its own + // delivery checkpoint controlling which rows reach its sink. + let journal_path = + journal::ensure_source_journal(&self.data_dir, &env.source, &env.session_id).await?; + let legacy_through = + journal::JournalReader::acknowledged_through(&journal_path, replay_through, route) + .await; + let translation = self.translation_for(env).await?; translation .observe(crate::pipeline::translation::Observation { key: key.correlation_key(), @@ -116,11 +120,8 @@ impl Daemon { // 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(); + let cursor = translation.journal.subscribe(route, legacy_through).await?; + let mut map = self.sessions.lock().unwrap(); if let Some(s) = map.get(key) { return Ok(s.clone()); } @@ -128,6 +129,7 @@ impl Daemon { SessionOptions { translation, cursor, + auth_provider: self.auth_provider.clone(), session_id: env.session_id.clone(), source: env.source.clone(), bt_version: self.version.clone(), diff --git a/bt-daemon/src/span_processor.rs b/bt-daemon/src/span_processor.rs index 55c33db2..cd75eabe 100644 --- a/bt-daemon/src/span_processor.rs +++ b/bt-daemon/src/span_processor.rs @@ -13,7 +13,7 @@ use std::collections::{BTreeMap, HashMap}; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; -use std::time::{Duration, Instant, SystemTime}; +use std::time::{Duration, Instant}; const MEMORY_LIMIT_BYTES: usize = 64 * 1024 * 1024; const STACK_LIMIT_BYTES: usize = 512 * 1024; @@ -42,6 +42,7 @@ struct Engine { modules: HashMap, failed_plugins: HashMap, env: BTreeMap, + last_fingerprint: Option, context: Context, started: Instant, deadline_ms: Arc, @@ -50,8 +51,7 @@ struct Engine { } struct CachedModule { - modified: Option, - len: u64, + fingerprint: PluginFingerprint, function: Persistent>, } @@ -63,8 +63,7 @@ struct FailedPlugin { #[derive(Clone, Copy, PartialEq, Eq)] struct PluginFingerprint { - modified: Option, - len: u64, + digest: [u8; 32], } impl Engine { @@ -85,6 +84,7 @@ impl Engine { modules: HashMap::new(), failed_plugins: HashMap::new(), env: environment(), + last_fingerprint: None, context, started, deadline_ms, @@ -93,16 +93,18 @@ impl Engine { } fn load(&mut self, path: &Path) -> anyhow::Result>> { - let metadata = std::fs::metadata(path) - .map_err(|error| anyhow::anyhow!("failed to inspect {}: {error}", path.display()))?; - let modified = metadata.modified().ok(); + self.last_fingerprint = None; + let source = std::fs::read(path) + .map_err(|error| anyhow::anyhow!("failed to read {}: {error}", path.display()))?; + let fingerprint = PluginFingerprint { + digest: Sha256::digest(&source).into(), + }; + self.last_fingerprint = Some(fingerprint); if let Some(module) = self.modules.get(path) { - if module.modified == modified && module.len == metadata.len() { + if module.fingerprint == fingerprint { return Ok(module.function.clone()); } } - let source = std::fs::read(path) - .map_err(|error| anyhow::anyhow!("failed to read {}: {error}", path.display()))?; let digest = Sha256::digest(&source); let name = format!("bt-span-plugin:{digest:x}"); self.arm_deadline(); @@ -128,8 +130,7 @@ impl Engine { self.modules.insert( path.to_path_buf(), CachedModule { - modified, - len: metadata.len(), + fingerprint, function: function.clone(), }, ); @@ -179,7 +180,7 @@ impl Engine { } } -#[derive(Debug)] +#[derive(Debug, Clone)] pub struct PluginFailure { pub path: PathBuf, pub message: String, @@ -203,6 +204,25 @@ pub fn process( op: &SpanOp, source: &str, session_id: &str, +) -> anyhow::Result { + process_inner(plugins, op, source, session_id, true) +} + +pub(crate) fn process_isolated( + plugins: &[PathBuf], + op: &SpanOp, + source: &str, + session_id: &str, +) -> anyhow::Result { + process_inner(plugins, op, source, session_id, false) +} + +fn process_inner( + plugins: &[PathBuf], + op: &SpanOp, + source: &str, + session_id: &str, + use_failed_cache: bool, ) -> anyhow::Result { if plugins.is_empty() { return Ok(ProcessResult { @@ -228,7 +248,12 @@ pub fn process( .as_mut() .ok_or_else(|| anyhow::anyhow!("span processor engine was not initialized"))?; for plugin in plugins { - if let Some(failed) = engine.failed_plugins.get(plugin).cloned() { + if let Some(failed) = engine + .failed_plugins + .get(plugin) + .cloned() + .filter(|_| use_failed_cache) + { if failed.fingerprint == plugin_fingerprint(plugin) { failure = Some(PluginFailure { path: plugin.clone(), @@ -256,9 +281,10 @@ pub fn process( } Ok(_) => { let message = "changed immutable span identity fields".to_owned(); - engine - .failed_plugins - .insert(plugin.clone(), failed_plugin(plugin, &message)); + engine.failed_plugins.insert( + plugin.clone(), + failed_plugin(engine.last_fingerprint, &message), + ); failure = Some(PluginFailure { path: plugin.clone(), message, @@ -268,9 +294,10 @@ pub fn process( } Err(error) => { let message = error.to_string(); - engine - .failed_plugins - .insert(plugin.clone(), failed_plugin(plugin, &message)); + engine.failed_plugins.insert( + plugin.clone(), + failed_plugin(engine.last_fingerprint, &message), + ); failure = Some(PluginFailure { path: plugin.clone(), message, @@ -297,18 +324,31 @@ pub fn process( }) } -fn failed_plugin(path: &Path, message: &str) -> FailedPlugin { +fn failed_plugin(fingerprint: Option, message: &str) -> FailedPlugin { FailedPlugin { - fingerprint: plugin_fingerprint(path), + fingerprint, message: message.to_owned(), } } fn plugin_fingerprint(path: &Path) -> Option { - let metadata = std::fs::metadata(path).ok()?; Some(PluginFingerprint { - modified: metadata.modified().ok(), - len: metadata.len(), + digest: Sha256::digest(std::fs::read(path).ok()?).into(), + }) +} + +pub(crate) fn file_digest(path: &Path) -> Option<[u8; 32]> { + plugin_fingerprint(path).map(|fingerprint| fingerprint.digest) +} + +pub(crate) fn failed_digest(path: &Path) -> Option<[u8; 32]> { + ENGINE.with_borrow(|engine| { + engine + .as_ref()? + .failed_plugins + .get(path)? + .fingerprint + .map(|f| f.digest) }) } diff --git a/bt-daemon/src/trace_runtime.rs b/bt-daemon/src/trace_runtime.rs index 114f1614..174354cd 100644 --- a/bt-daemon/src/trace_runtime.rs +++ b/bt-daemon/src/trace_runtime.rs @@ -356,7 +356,12 @@ async fn diagnose_daemon( .sessions .into_iter() .filter(|session| is_same_agent(source, &session.source)) - .filter_map(|session| session.last_error); + .filter_map(|session| { + session.last_error.map(|error| match session.recovery { + Some(recovery) => format!("{error}; retry eligible: {}", recovery.eligible), + None => error, + }) + }); for error in errors { if !session_errors.contains(&error) { session_errors.push(error); diff --git a/bt-daemon/src/wire/methods.rs b/bt-daemon/src/wire/methods.rs index f3a4a782..7461d03e 100644 --- a/bt-daemon/src/wire/methods.rs +++ b/bt-daemon/src/wire/methods.rs @@ -91,12 +91,23 @@ pub struct StatusResult { pub sessions: Vec, } +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RecoveryInfo { + pub stage: String, + pub position: u64, + pub reason: String, + pub retry: String, + pub eligible: bool, +} + #[derive(Debug, Clone, Serialize, Deserialize)] pub struct SessionStatus { pub session_id: String, pub source: String, #[serde(default, skip_serializing_if = "Option::is_none")] pub route: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub recovery: Option, pub queued: u64, pub spans_emitted: u64, #[serde(default, skip_serializing_if = "Option::is_none")] diff --git a/bt-daemon/src/wire/mod.rs b/bt-daemon/src/wire/mod.rs index 666b4a82..48e9dfa4 100644 --- a/bt-daemon/src/wire/mod.rs +++ b/bt-daemon/src/wire/mod.rs @@ -16,7 +16,7 @@ pub use envelope::{ pub use methods::{ method, AuthDiagnoseParams, AuthDiagnoseResult, Capabilities, ClientInfo, EventLogResult, FlushParams, FlushResult, InitializeParams, InitializeResult, ManagedRunFlushParams, - SessionStatus, ShutdownResult, StatusParams, StatusResult, + RecoveryInfo, SessionStatus, ShutdownResult, StatusParams, StatusResult, }; pub use rpc::{error_code, Message, Request, RequestId, Response, RpcError}; diff --git a/bt-daemon/tests/pipeline.rs b/bt-daemon/tests/pipeline.rs index 78d0de47..8c143e5c 100644 --- a/bt-daemon/tests/pipeline.rs +++ b/bt-daemon/tests/pipeline.rs @@ -587,7 +587,7 @@ async fn shutdown(socket: &Path) { /// A failed flush consumes its batch, just like checked SDK delivery. An empty /// subsequent flush must not acknowledge that lost batch's source events. #[tokio::test] -async fn rejected_flush_blocks_checkpoints_until_journal_replay() { +async fn rejected_flush_retries_the_span_journal_before_checkpointing() { use std::sync::atomic::{AtomicUsize, Ordering}; #[derive(Default)] @@ -656,24 +656,39 @@ async fn rejected_flush_blocks_checkpoints_until_journal_replay() { .await .unwrap(); flush_session(session, &socket, 5_000).await.unwrap(); + tokio::time::timeout(Duration::from_secs(5), async { + loop { + let (head, tail) = head_and_tail(&data_dir); + if head == tail && tail > 0 { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .unwrap(); + assert!( + sink.emitted.load(Ordering::SeqCst) >= 5, + "rejected initial rows must be resubmitted" + ); shutdown(&socket).await; handle.await.unwrap(); let journal_path = source_journal_path(&data_dir, "debug", session); let journal = std::fs::read_to_string(&journal_path).unwrap(); assert!(!journal.contains("delivery_checkpoint"), "{journal}"); - assert_eq!(sink.flushes.load(Ordering::SeqCst), 1); + assert!(sink.flushes.load(Ordering::SeqCst) > 1); let emitted_before_restart = sink.emitted.load(Ordering::SeqCst); let handle = start(); wait_for(&socket).await; flush_session(session, &socket, 5_000).await.unwrap(); - assert!(sink.emitted.load(Ordering::SeqCst) > emitted_before_restart); + assert_eq!(sink.emitted.load(Ordering::SeqCst), emitted_before_restart); let (head, tail) = head_and_tail(&data_dir); assert!(tail > 0); assert_eq!( head, tail, - "restart completes the durable SpanOp subscription" + "automatic retry completed the durable subscription before restart" ); shutdown(&socket).await; handle.await.unwrap(); @@ -1989,7 +2004,7 @@ async fn span_plugins_transform_live_and_replayed_rows_with_daemon_environment() } #[tokio::test] -async fn a_failing_span_plugin_discards_rows_and_persists_the_raw_exception() { +async fn a_failing_span_plugin_holds_rows_and_persists_the_raw_exception() { let (data_dir, socket, handle, tmp) = start_daemon().await; let plugin = tmp.path().join("bad.mjs"); let later_plugin = tmp.path().join("later.mjs"); @@ -2027,7 +2042,7 @@ async fn a_failing_span_plugin_discards_rows_and_persists_the_raw_exception() { status.sessions[0] .last_error .as_deref() - .is_some_and(|error| error.contains("span operations are being discarded")), + .is_some_and(|error| error.contains("delivery is held")), "unexpected plugin status: {:?}", status.sessions[0].last_error ); @@ -2061,6 +2076,20 @@ async fn a_failing_span_plugin_discards_rows_and_persists_the_raw_exception() { .await .unwrap(); + tokio::time::timeout(Duration::from_secs(5), async { + loop { + if flush_session("plugin-failure", &socket, 5000) + .await + .unwrap() + .flushed + { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .unwrap(); let spans = std::fs::read_to_string(data_dir.join("spans/plugin-failure.ndjson")).unwrap(); assert!( spans.contains("after-failure:"), @@ -3376,6 +3405,125 @@ async fn status_counts_rows_emitted_before_a_plugin_failure() { handle.await.unwrap(); } +#[tokio::test] +async fn missing_credentials_recover_without_another_native_event() { + use std::sync::atomic::{AtomicBool, Ordering}; + struct AvailableAuth { + ready: Arc, + inner: TestAuthProvider, + } + #[async_trait] + impl AuthProvider for AvailableAuth { + async fn resolve( + &self, + selection: &AuthSelection, + reason: AuthResolveReason, + ) -> anyhow::Result { + anyhow::ensure!(self.ready.load(Ordering::SeqCst), "credential unavailable"); + self.inner.resolve(selection, reason).await + } + } + let tmp = tempfile::tempdir().unwrap(); + let data_dir = tmp.path().join("data"); + let socket = test_endpoint(tmp.path()); + let ready = Arc::new(AtomicBool::new(false)); + let mut options = debug_serve_options("test", &data_dir); + options.auth_provider = Some(Arc::new(AvailableAuth { + ready: ready.clone(), + inner: TestAuthProvider { + calls: Mutex::new(Vec::new()), + fail: false, + first_lease_expired: false, + }, + })); + 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, options).await.unwrap() }); + wait_for(&socket).await; + forward_envelope( + &envelope("auth-recovery", "SessionStart", 1), + &socket, + &dummy_host(), + false, + ) + .await + .unwrap(); + flush_session("auth-recovery", &socket, 5000).await.unwrap(); + let status = run_status(StatusArgs { + socket: Some(socket.clone()), + session_id: Some("auth-recovery".into()), + }) + .await + .unwrap() + .unwrap(); + assert!(status.sessions.iter().any(|s| s + .recovery + .as_ref() + .is_some_and(|r| r.stage == "credentials"))); + tokio::time::timeout(Duration::from_secs(5), async { + loop { + if std::fs::read_dir(data_dir.join("pipeline")) + .ok() + .is_some_and(|mut entries| entries.next().is_some()) + { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .unwrap(); + let db_path = std::fs::read_dir(data_dir.join("pipeline")) + .unwrap() + .filter_map(Result::ok) + .map(|e| e.path()) + .find(|p| p.extension().is_some_and(|e| e == "sqlite")) + .unwrap(); + tokio::time::timeout(Duration::from_secs(5), async { + loop { + let db = rusqlite::Connection::open(&db_path).unwrap(); + let count: u64 = db + .query_row( + "SELECT count(*) FROM ops JOIN batches USING(input) WHERE committed=1", + [], + |r| r.get(0), + ) + .unwrap(); + if count >= 2 { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .unwrap(); + ready.store(true, Ordering::SeqCst); + tokio::time::timeout(Duration::from_secs(5), async { + loop { + if std::fs::read_to_string(data_dir.join("spans/auth-recovery.ndjson")) + .is_ok_and(|s| s.lines().count() >= 2) + { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .unwrap(); + assert!( + flush_session("auth-recovery", &socket, 5000) + .await + .unwrap() + .flushed + ); + 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}; From 1a8c2722a2d2827c59bd38cc7346ec2794543d5f Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Wed, 14 Oct 2026 13:13:22 +0800 Subject: [PATCH 2/4] Recover plugin chains when predecessor contents change --- bt-daemon/src/pipeline/consumer.rs | 52 ++++++++++++++++++++++++++++++ bt-daemon/src/pipeline/plugins.rs | 14 ++++++-- bt-daemon/src/pipeline/recovery.rs | 14 ++++++++ bt-daemon/src/span_processor.rs | 28 ++++++++++++++++ 4 files changed, 106 insertions(+), 2 deletions(-) diff --git a/bt-daemon/src/pipeline/consumer.rs b/bt-daemon/src/pipeline/consumer.rs index ed7f93e3..03955f47 100644 --- a/bt-daemon/src/pipeline/consumer.rs +++ b/bt-daemon/src/pipeline/consumer.rs @@ -751,4 +751,56 @@ mod tests { assert_eq!(record.submitted.load(Ordering::SeqCst), 2); consumer.shutdown().await; } + #[tokio::test] + async fn editing_a_predecessor_releases_a_later_plugins_hold() { + let tmp = tempfile::tempdir().unwrap(); + let first = tmp.path().join("first.mjs"); + let last = tmp.path().join("last.mjs"); + std::fs::write(&first, "export default span => ({...span,name:'bad'})").unwrap(); + std::fs::write(&last,"export default span => {if(span.name==='bad') throw new Error('bad input'); return span}").unwrap(); + let last_digest = crate::span_processor::file_digest(&last); + let journal = SpanJournal::open(&tmp.path().join("spans.sqlite")) + .await + .unwrap(); + stored(&journal).await; + let route = SessionRoute { + span_plugins: vec![first.clone(), last.clone()], + ..Default::default() + }; + let record = Arc::new(Recording::default()); + let consumer = Consumer::spawn( + journal.clone(), + journal.subscribe(&route, 0).await.unwrap(), + options(tmp.path(), &route, record.clone()), + ); + wait_for(|| consumer.error.lock().unwrap().is_some()).await; + let failure: Failure = serde_json::from_str( + &journal + .failure(&serde_json::to_string(&route).unwrap()) + .await + .unwrap() + .unwrap(), + ) + .unwrap(); + std::fs::write( + tmp.path().join("unrelated.mjs"), + "export default span => span", + ) + .unwrap(); + let config = consumer.config.borrow().clone(); + assert!(matches!( + (Recovery { + config: Some(&config), + version: "test" + }) + .process(&failure) + .await, + RecoveryOutcome::Waiting + )); + std::fs::write(&first, "export default span => ({...span,name:'good'})").unwrap(); + wait_for(|| *consumer.progress.borrow() == 1).await; + assert_eq!(crate::span_processor::file_digest(&last), last_digest); + assert_eq!(record.accepted.load(Ordering::SeqCst), 1); + consumer.shutdown().await; + } } diff --git a/bt-daemon/src/pipeline/plugins.rs b/bt-daemon/src/pipeline/plugins.rs index 6f8143ce..9fa2e1cd 100644 --- a/bt-daemon/src/pipeline/plugins.rs +++ b/bt-daemon/src/pipeline/plugins.rs @@ -21,6 +21,7 @@ pub(crate) struct Plugins { pub(crate) struct PluginRejected { pub failure: crate::span_processor::PluginFailure, pub digest: Option<[u8; 32]>, + pub processors: Vec>, } impl std::fmt::Display for PluginRejected { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { @@ -53,7 +54,11 @@ impl Stage for Plugins { async fn process(&mut self, input: &SpanOp) -> PluginOutcome { if let Some((paths, failure)) = &self.blocked { if paths == &self.paths - && span_processor::file_digest(&failure.failure.path) == failure.digest + && paths + .iter() + .map(|path| span_processor::file_digest(path)) + .collect::>() + == failure.processors { let mut failure = failure.clone(); failure.failure.newly_seen = false; @@ -65,7 +70,12 @@ impl Stage for Plugins { Ok(result) => match (result.op, result.failure) { (_, Some(failure)) => { let digest = span_processor::failed_digest(&failure.path); - let failure = PluginRejected { failure, digest }; + let processors = span_processor::chain_digests(&self.paths, &failure.path); + let failure = PluginRejected { + failure, + digest, + processors, + }; self.blocked = Some((self.paths.clone(), failure.clone())); PluginOutcome::Failed(failure) } diff --git a/bt-daemon/src/pipeline/recovery.rs b/bt-daemon/src/pipeline/recovery.rs index f0b43149..42038c61 100644 --- a/bt-daemon/src/pipeline/recovery.rs +++ b/bt-daemon/src/pipeline/recovery.rs @@ -37,6 +37,8 @@ pub(crate) enum Condition { path: PathBuf, digest: Option<[u8; 32]>, chain: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + processors: Option>>, }, Configuration { signature: String, @@ -135,6 +137,7 @@ impl Failure { path: plugin.failure.path.clone(), digest: plugin.digest, chain: chain(config), + processors: Some(plugin.processors.clone()), }; } else if let Some(error) = error.downcast_ref::() { use braintrust_sdk_rust::BraintrustError; @@ -251,9 +254,20 @@ impl Stage for Recovery<'_> { path, digest, chain: previous, + processors, } => { self.config.is_some_and(|config| chain(config) != *previous) || crate::span_processor::file_digest(path) != *digest + || processors.as_ref().is_some_and(|previous| { + self.config.is_some_and(|config| { + config + .span_plugins + .iter() + .map(|path| crate::span_processor::file_digest(path)) + .collect::>() + != *previous + }) + }) } Condition::Configuration { signature, diff --git a/bt-daemon/src/span_processor.rs b/bt-daemon/src/span_processor.rs index cd75eabe..f706496e 100644 --- a/bt-daemon/src/span_processor.rs +++ b/bt-daemon/src/span_processor.rs @@ -352,6 +352,34 @@ pub(crate) fn failed_digest(path: &Path) -> Option<[u8; 32]> { }) } +// Preserve the bytes actually loaded by predecessors, not a later filesystem +// sample: an edit during a failing call must already qualify for recovery. +pub(crate) fn chain_digests(paths: &[PathBuf], failed: &Path) -> Vec> { + ENGINE.with_borrow(|engine| { + let mut reached_failure = false; + paths + .iter() + .map(|path| { + if path == failed { + reached_failure = true; + engine + .as_ref() + .and_then(|engine| engine.failed_plugins.get(path)) + .and_then(|failed| failed.fingerprint) + .map(|fingerprint| fingerprint.digest) + } else if !reached_failure { + engine + .as_ref() + .and_then(|engine| engine.modules.get(path)) + .map(|module| module.fingerprint.digest) + } else { + file_digest(path) + } + }) + .collect() + }) +} + /// Compile each module and verify that it default-exports a function. Explicit /// CLI commands call this before persisting or launching with a plugin chain. pub fn validate(plugins: &[PathBuf]) -> anyhow::Result<()> { From effdf9a11ae985e1c9aef687aae7dab9cb55ed9f Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Wed, 14 Oct 2026 18:28:57 +0800 Subject: [PATCH 3/4] Share route completion between delivery and recovery --- bt-daemon/src/pipeline/consumer.rs | 542 ++++++++++++----------------- 1 file changed, 220 insertions(+), 322 deletions(-) diff --git a/bt-daemon/src/pipeline/consumer.rs b/bt-daemon/src/pipeline/consumer.rs index 03955f47..bd33ee1e 100644 --- a/bt-daemon/src/pipeline/consumer.rs +++ b/bt-daemon/src/pipeline/consumer.rs @@ -25,10 +25,7 @@ enum Work { Flush, Complete(Delivery<()>), } -enum State { - Processing, - Recovering(Failure, Work), -} +type Held = Box<(Failure, Work)>; pub(crate) struct Consumer { task: Mutex>>, @@ -117,6 +114,96 @@ impl ConsumerOptions { config.auth = lease.auth; Ok(()) } + + fn storage(&self, error: &anyhow::Error, cursor: &Cursor, attempt: u32, work: Work) -> Held { + let mut failure = Failure::storage( + error, + cursor.completed_position() + 1, + &self.version, + attempt, + ); + failure.boundary = match work { + Work::Complete(_) => Boundary::Completion, + Work::Flush => Boundary::Ledger, + Work::Resubmit => failure.boundary, + }; + Box::new((failure, work)) + } + + fn routing( + &self, + error: &anyhow::Error, + cursor: &Cursor, + config: &SessionConfig, + attempt: u32, + ) -> Held { + if let Some(partial) = error.downcast_ref::() { + self.counters + .spans_emitted + .fetch_add(partial.emitted, Ordering::Relaxed); + if partial.report { + tracing::warn!(session_id=%self.session_id, %error, "route processing failed"); + } + } + Box::new(( + Failure::routing(error, cursor.completed_position() + 1, config, attempt), + Work::Resubmit, + )) + } + + // Normal delivery and recovery share the same acceptance/commit path. Only a + // rejected batch is resubmitted; accepted work retains its local retry point. + async fn finish( + &self, + cursor: &mut Cursor, + sink: &mut Option, + config: &SessionConfig, + attempt: u32, + work: Work, + ) -> Result { + let completed = !matches!(work, Work::Resubmit); + let completion = match work { + Work::Resubmit => { + // The SDK consumed the rejected batch. Recreate its ledger too, + // so in-memory filtering cannot turn a retry into an empty ack. + *sink = None; + None + } + Work::Complete(delivery) => Some(delivery), + Work::Flush => { + let sink = sink.as_mut().expect("accepted delivery retains its sink"); + sink.flush().await.map_err(|error| { + if error.downcast_ref::().is_some() { + self.storage(&error, cursor, attempt, Work::Flush) + } else { + self.routing(&error, cursor, config, attempt) + } + })?; + *self.permalink.lock().unwrap() = sink.permalink(); + // Deferred roots keep their prefix unresolved until a child + // releases them; the cursor can read ahead without buffering ops. + if sink.has_pending_delivery() { + return Ok(false); + } + Some(cursor.pending_prefix()) + } + }; + if let Some(delivery) = completion { + delivery.into_completed().await.map_err(|unfinished| { + self.storage( + &unfinished.error, + cursor, + attempt, + Work::Complete(unfinished.delivery), + ) + })?; + } + cursor + .rewind() + .await + .map_err(|error| self.storage(&error, cursor, attempt, Work::Resubmit))?; + Ok(completed) + } } impl Consumer { @@ -132,342 +219,153 @@ impl Consumer { error: options.error.clone(), }); let task = tokio::spawn(async move { - let mut sink: Option = None; - let restored = journal.failure(&cursor.route).await; - let mut state = match restored { - Ok(Some(raw)) => State::Recovering( + let mut sink = None; + let mut held = match journal.failure(&cursor.route).await { + Ok(Some(raw)) => Some(Box::new(( serde_json::from_str(&raw).unwrap_or_else(|_| { Failure::legacy(raw, cursor.completed_position() + 1, &options.config) }), Work::Resubmit, - ), - Ok(None) => State::Processing, - Err(error) => State::Recovering( - Failure::storage(&error, cursor.completed_position() + 1, &options.version, 0), - Work::Resubmit, - ), + ))), + Ok(None) => None, + Err(error) => Some(options.storage(&error, &cursor, 0, Work::Resubmit)), }; let mut attempts = 0; let mut recorded = None; let mut record_retry = 0; loop { - match state { - State::Recovering(mut failure, work) => { - if matches!(failure.boundary, Boundary::Completion) - && cursor.completed_position() - >= failure.through.unwrap_or(failure.position) - && journal.set_failure(&cursor.route, None).await.is_ok() - { - *options.error.lock().unwrap() = None; - *options.recovery.lock().unwrap() = None; - state = State::Processing; - continue; - } - failure.through.get_or_insert(cursor.position()); - let encoded = serde_json::to_string(&failure).expect("failure evidence"); - if recorded.as_ref() != Some(&encoded) - && super::recovery::now_ms() >= record_retry - { - if options.hold(&journal, &cursor, &mut failure).await { - recorded = Some(encoded.clone()); - } else { - record_retry = super::recovery::now_ms().saturating_add(1_000); - } - } - progress.send_replace(cursor.completed_position()); - tokio::select! { - _=stop.notified()=>break, - _=tokio::time::sleep(std::time::Duration::from_millis(250))=>{}, - result=settings.changed()=>if result.is_err() { break; }, - } - let mut current = settings.borrow_and_update().clone(); - if recorded.as_ref() != Some(&encoded) - || matches!( - (Recovery { - config: Some(¤t), - version: &options.version - }) - .process(&failure) - .await, - RecoveryOutcome::Waiting - ) - { - state = State::Recovering(failure, work); - continue; - } - *options.recovery.lock().unwrap() = Some(failure.info(true)); - attempts = failure.attempt.saturating_add(1); - if matches!(failure.boundary, Boundary::Credentials) { - if let Err(error) = - options.credentials(&cursor.route, &mut current).await - { - state = State::Recovering( - Failure::credentials( - cursor.completed_position() + 1, - error.to_string(), - attempts, - ), - work, - ); - continue; - } - config.send_replace(current.clone()); - } - let retry = match work { - Work::Complete(delivery) => match delivery.into_completed().await { - Ok(()) => Ok(true), - Err(unfinished) => { - let mut failure = Failure::storage( - &unfinished.error, - cursor.completed_position() + 1, - &options.version, - attempts, - ); - failure.boundary = Boundary::Completion; - state = State::Recovering( - failure, - Work::Complete(unfinished.delivery), - ); - continue; - } - }, - Work::Flush => sink - .as_mut() - .expect("accepted delivery retains its sink") - .flush() - .await - .map(|()| true), - Work::Resubmit => { - sink = None; - Ok(false) - } - }; - match retry { - Ok(completed) => { - if completed - && !sink - .as_ref() - .is_some_and(|sink| sink.has_pending_delivery()) - { - if let Err(unfinished) = - cursor.pending_prefix().into_completed().await - { - let mut failure = Failure::storage( - &unfinished.error, - cursor.completed_position() + 1, - &options.version, - attempts, - ); - failure.boundary = Boundary::Completion; - state = State::Recovering( - failure, - Work::Complete(unfinished.delivery), - ); - continue; - } - } - if let Err(error) = cursor.rewind().await { - state = State::Recovering( - Failure::storage( - &error, - cursor.completed_position() + 1, - &options.version, - attempts, - ), - Work::Resubmit, - ); - continue; - } - progress.send_replace(cursor.completed_position()); - if completed { - *options.error.lock().unwrap() = None; - *options.recovery.lock().unwrap() = None; - recorded = None; - } - state = State::Processing; - } - Err(error) if error.downcast_ref::().is_some() => { - let mut failure = Failure::storage( - &error, - cursor.completed_position() + 1, - &options.version, - attempts, - ); - failure.boundary = Boundary::Ledger; - state = State::Recovering(failure, Work::Flush); - } - Err(error) => { - state = State::Recovering( - Failure::routing( - &error, - cursor.completed_position() + 1, - ¤t, - attempts, - ), - Work::Resubmit, - ); - } - } + if let Some(pending) = held.take() { + let (mut failure, work) = *pending; + if matches!(failure.boundary, Boundary::Completion) + && cursor.completed_position() + >= failure.through.unwrap_or(failure.position) + && journal.set_failure(&cursor.route, None).await.is_ok() + { + *options.error.lock().unwrap() = None; + *options.recovery.lock().unwrap() = None; + continue; } - State::Processing => { - let tail = match journal.tail().await { - Ok(tail) => tail, - Err(error) => { - state = State::Recovering( - Failure::storage( - &error, - cursor.completed_position() + 1, - &options.version, - attempts, - ), - Work::Resubmit, - ); - continue; - } - }; - options.counters.queued.store( - tail.saturating_sub(cursor.completed_position()), - Ordering::Relaxed, - ); - let received = tokio::select! { - _=stop.notified()=>break, - result=InputPort::receive(&mut cursor)=>result, - result=settings.changed()=> { if result.is_err() {break;} state=State::Processing; continue; }, - }; - let delivery = match received { - Ok(Some(delivery)) => delivery, - Ok(None) => break, - Err(error) => { - state = State::Recovering( - Failure::storage( - &error, - cursor.completed_position() + 1, - &options.version, - attempts, - ), - Work::Resubmit, - ); - continue; - } - }; - let current = settings.borrow_and_update().clone(); - if sink.is_none() { - match options.sink(¤t).await { - Ok(created) => sink = Some(created), - Err(error) => { - state = State::Recovering( - Failure::routing( - &error, - cursor.completed_position() + 1, - ¤t, - attempts, - ), - Work::Resubmit, - ); - continue; - } - } + failure.through.get_or_insert(cursor.position()); + let encoded = serde_json::to_string(&failure).expect("failure evidence"); + if recorded.as_ref() != Some(&encoded) + && super::recovery::now_ms() >= record_retry + { + if options.hold(&journal, &cursor, &mut failure).await { + recorded = Some(encoded.clone()); + } else { + record_retry = super::recovery::now_ms().saturating_add(1_000); } - let sink = sink.as_mut().expect("configured sink"); - sink.configure(¤t); - let emitting = tokio::select! { _=stop.notified()=>break,result=sink.emit(delivery.value())=>result }; - match emitting { - Ok(count) => { - options - .counters - .spans_emitted - .fetch_add(count, Ordering::Relaxed); - } - Err(error) => { - if let Some(partial) = - error.downcast_ref::() - { - options - .counters - .spans_emitted - .fetch_add(partial.emitted, Ordering::Relaxed); - if partial.report { - tracing::warn!(%error,session_id=%options.session_id,"route processing failed"); - } - } - state = State::Recovering( - Failure::routing( - &error, - cursor.completed_position() + 1, - ¤t, - attempts, - ), - Work::Resubmit, - ); - continue; - } - } - let flushing = - tokio::select! { _=stop.notified()=>break,result=sink.flush()=>result }; - if let Err(error) = flushing { - state = if error.downcast_ref::().is_some() { - let mut failure = Failure::storage( - &error, + } + progress.send_replace(cursor.completed_position()); + tokio::select! { + _=stop.notified()=>break, + _=tokio::time::sleep(std::time::Duration::from_millis(250))=>{}, + result=settings.changed()=>if result.is_err() { break; }, + } + let mut current = settings.borrow_and_update().clone(); + if recorded.as_ref() != Some(&encoded) + || matches!( + (Recovery { + config: Some(¤t), + version: &options.version + }) + .process(&failure) + .await, + RecoveryOutcome::Waiting + ) + { + held = Some(Box::new((failure, work))); + continue; + } + *options.recovery.lock().unwrap() = Some(failure.info(true)); + attempts = failure.attempt.saturating_add(1); + if matches!(failure.boundary, Boundary::Credentials) { + if let Err(error) = options.credentials(&cursor.route, &mut current).await { + held = Some(Box::new(( + Failure::credentials( cursor.completed_position() + 1, - &options.version, + error.to_string(), attempts, - ); - failure.boundary = Boundary::Ledger; - State::Recovering(failure, Work::Flush) - } else { - State::Recovering( - Failure::routing( - &error, - cursor.completed_position() + 1, - ¤t, - attempts, - ), - Work::Resubmit, - ) - }; + ), + work, + ))); continue; } - *options.permalink.lock().unwrap() = sink.permalink(); - drop(delivery); - if !sink.has_pending_delivery() { - match cursor.pending_prefix().into_completed().await { - Ok(()) => { - if let Err(error) = cursor.rewind().await { - state = State::Recovering( - Failure::storage( - &error, - cursor.completed_position() + 1, - &options.version, - attempts, - ), - Work::Resubmit, - ); - continue; - } - progress.send_replace(cursor.completed_position()); - *options.error.lock().unwrap() = None; - *options.recovery.lock().unwrap() = None; - attempts = 0; - recorded = None; - } - Err(unfinished) => { - let mut failure = Failure::storage( - &unfinished.error, - cursor.completed_position() + 1, - &options.version, - attempts, - ); - failure.boundary = Boundary::Completion; - state = State::Recovering( - failure, - Work::Complete(unfinished.delivery), - ); - continue; - } + config.send_replace(current.clone()); + } + match options + .finish(&mut cursor, &mut sink, ¤t, attempts, work) + .await + { + Ok(completed) => { + progress.send_replace(cursor.completed_position()); + if completed { + *options.error.lock().unwrap() = None; + *options.recovery.lock().unwrap() = None; + recorded = None; } } - state = State::Processing; + Err(failure) => held = Some(failure), + } + continue; + } + let attempt = async { + let tail = journal.tail().await.map_err(|error| { + options.storage(&error, &cursor, attempts, Work::Resubmit) + })?; + options.counters.queued.store( + tail.saturating_sub(cursor.completed_position()), + Ordering::Relaxed, + ); + let received = tokio::select! { + result=InputPort::receive(&mut cursor)=>result, + result=settings.changed()=> { + return if result.is_err() { Ok(None) } else { Ok(Some(false)) }; + }, + }; + let delivery = received.map_err(|error| { + options.storage(&error, &cursor, attempts, Work::Resubmit) + })?; + let Some(delivery) = delivery else { + return Ok(None); + }; + let current = settings.borrow_and_update().clone(); + if sink.is_none() { + sink = Some(options.sink(¤t).await.map_err(|error| { + options.routing(&error, &cursor, ¤t, attempts) + })?); + } + let routing = sink.as_mut().expect("configured sink"); + routing.configure(¤t); + let count = routing + .emit(delivery.value()) + .await + .map_err(|error| options.routing(&error, &cursor, ¤t, attempts))?; + options + .counters + .spans_emitted + .fetch_add(count, Ordering::Relaxed); + drop(delivery); + options + .finish(&mut cursor, &mut sink, ¤t, attempts, Work::Flush) + .await + .map(Some) + }; + let result = tokio::select! { + _=stop.notified()=>break, + result=attempt=>result, + }; + match result { + Ok(None) => break, + Ok(Some(false)) => {} + Ok(Some(true)) => { + progress.send_replace(cursor.completed_position()); + *options.error.lock().unwrap() = None; + *options.recovery.lock().unwrap() = None; + attempts = 0; + recorded = None; } + Err(failure) => held = Some(failure), } } }); From 8b9e7a863ec9e47460809723c00f419a913950eb Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Sat, 17 Oct 2026 00:02:35 +0800 Subject: [PATCH 4/4] Align recovery with append-only ledger completion and panic bans --- bt-daemon/src/delivery_ledger.rs | 73 ++++++++++++++++-- bt-daemon/src/pipeline/consumer.rs | 104 ++++++++++++++++++++------ bt-daemon/src/pipeline/recovery.rs | 14 ++-- bt-daemon/src/pipeline/translation.rs | 50 ++++++++----- bt-daemon/src/server/rpc.rs | 3 +- bt-daemon/src/server/sessions.rs | 5 +- 6 files changed, 193 insertions(+), 56 deletions(-) diff --git a/bt-daemon/src/delivery_ledger.rs b/bt-daemon/src/delivery_ledger.rs index 814687a2..6b7a7ce8 100644 --- a/bt-daemon/src/delivery_ledger.rs +++ b/bt-daemon/src/delivery_ledger.rs @@ -430,11 +430,15 @@ impl Sink for LedgerSink { async fn flush(&mut self) -> anyhow::Result<()> { self.inner.flush().await?; if let Some(ledger) = &mut self.ledger { - let lock = ledger.lock().await?; - ledger - .refresh(&mut self.span_origins, &mut self.pending_origins, &lock) - .await?; - ledger.commit(&lock).await?; + async { + let lock = ledger.lock().await?; + ledger + .refresh(&mut self.span_origins, &mut self.pending_origins, &lock) + .await?; + ledger.commit(&lock).await + } + .await + .map_err(|error| anyhow::Error::new(CommitFailure(error)))?; } Ok(()) } @@ -529,6 +533,65 @@ mod tests { }) } + #[tokio::test] + async fn post_acceptance_ledger_failures_retain_pending_completion() { + for phase in ["lock", "refresh", "append"] { + let tmp = tempfile::tempdir().unwrap(); + let config = config("accepted"); + let mut sink = LedgerSink::new( + Box::new(RecordingSink::default()), + Some(tmp.path()), + "debug", + "test", + Some(&config), + "test", + ) + .await; + sink.emit(&[terminal("one")]).await.unwrap(); + let path = sink.ledger.as_ref().unwrap().path.clone(); + let original = std::fs::read(&path).unwrap(); + let permissions = std::fs::metadata(&path).unwrap().permissions(); + match phase { + "lock" => { + std::fs::remove_file(path.with_extension("lock")).unwrap(); + std::fs::create_dir(path.with_extension("lock")).unwrap(); + } + "refresh" => { + use std::io::Write; + std::fs::OpenOptions::new() + .append(true) + .open(&path) + .unwrap() + .write_all(b"invalid JSON\n") + .unwrap(); + } + "append" => { + let mut readonly = permissions.clone(); + readonly.set_readonly(true); + std::fs::set_permissions(&path, readonly).unwrap(); + } + _ => unreachable!(), + } + let error = sink.flush().await.unwrap_err(); + assert!( + error.downcast_ref::().is_some(), + "{phase}: {error:#}" + ); + assert!(sink.ledger.as_ref().unwrap().pending.contains("one")); + match phase { + "lock" => std::fs::remove_dir(path.with_extension("lock")).unwrap(), + "refresh" => std::fs::write(&path, &original).unwrap(), + "append" => std::fs::set_permissions(&path, permissions).unwrap(), + _ => unreachable!(), + } + sink.flush().await.unwrap(); + let (restored, _) = DeliveryLedger::load(tmp.path(), "debug", "test", &config) + .await + .unwrap(); + assert!(restored.known.contains("one"), "{phase}"); + } + } + #[tokio::test] async fn open_span_snapshots_survive_recovery_before_delivery_checkpoint() { let temp = tempfile::tempdir().unwrap(); diff --git a/bt-daemon/src/pipeline/consumer.rs b/bt-daemon/src/pipeline/consumer.rs index bd33ee1e..81f74333 100644 --- a/bt-daemon/src/pipeline/consumer.rs +++ b/bt-daemon/src/pipeline/consumer.rs @@ -72,14 +72,13 @@ impl ConsumerOptions { } async fn hold(&self, journal: &SpanJournal, cursor: &Cursor, failure: &mut Failure) -> bool { failure.through.get_or_insert(cursor.position()); - *self.error.lock().unwrap() = Some(failure.message()); - *self.recovery.lock().unwrap() = Some(failure.info(false)); - if let Err(error) = journal - .set_failure( - &cursor.route, - Some(serde_json::to_string(failure).expect("failure evidence")), - ) - .await + *self.error.lock() = Some(failure.message()); + *self.recovery.lock() = Some(failure.info(false)); + if let Err(error) = async { + let encoded = serde_json::to_string(failure)?; + journal.set_failure(&cursor.route, Some(encoded)).await + } + .await { tracing::warn!(%error,"route recovery evidence could not be persisted; original failure retained"); return false; @@ -171,7 +170,14 @@ impl ConsumerOptions { } Work::Complete(delivery) => Some(delivery), Work::Flush => { - let sink = sink.as_mut().expect("accepted delivery retains its sink"); + let sink = sink.as_mut().ok_or_else(|| { + self.storage( + &anyhow::anyhow!("accepted delivery has no sink"), + cursor, + attempt, + Work::Flush, + ) + })?; sink.flush().await.map_err(|error| { if error.downcast_ref::().is_some() { self.storage(&error, cursor, attempt, Work::Flush) @@ -179,7 +185,7 @@ impl ConsumerOptions { self.routing(&error, cursor, config, attempt) } })?; - *self.permalink.lock().unwrap() = sink.permalink(); + *self.permalink.lock() = sink.permalink(); // Deferred roots keep their prefix unresolved until a child // releases them; the cursor can read ahead without buffering ops. if sink.has_pending_delivery() { @@ -241,12 +247,29 @@ impl Consumer { >= failure.through.unwrap_or(failure.position) && journal.set_failure(&cursor.route, None).await.is_ok() { - *options.error.lock().unwrap() = None; - *options.recovery.lock().unwrap() = None; + *options.error.lock() = None; + *options.recovery.lock() = None; continue; } failure.through.get_or_insert(cursor.position()); - let encoded = serde_json::to_string(&failure).expect("failure evidence"); + let encoded = match serde_json::to_string(&failure) { + Ok(encoded) => encoded, + Err(error) => { + held = Some(Box::new(( + Failure::revision( + Boundary::Journal, + failure.position, + format!( + "{}; recovery evidence cannot be encoded: {error}", + failure.message() + ), + &options.version, + ), + work, + ))); + continue; + } + }; if recorded.as_ref() != Some(&encoded) && super::recovery::now_ms() >= record_retry { @@ -277,7 +300,7 @@ impl Consumer { held = Some(Box::new((failure, work))); continue; } - *options.recovery.lock().unwrap() = Some(failure.info(true)); + *options.recovery.lock() = Some(failure.info(true)); attempts = failure.attempt.saturating_add(1); if matches!(failure.boundary, Boundary::Credentials) { if let Err(error) = options.credentials(&cursor.route, &mut current).await { @@ -300,8 +323,8 @@ impl Consumer { Ok(completed) => { progress.send_replace(cursor.completed_position()); if completed { - *options.error.lock().unwrap() = None; - *options.recovery.lock().unwrap() = None; + *options.error.lock() = None; + *options.recovery.lock() = None; recorded = None; } } @@ -335,7 +358,14 @@ impl Consumer { options.routing(&error, &cursor, ¤t, attempts) })?); } - let routing = sink.as_mut().expect("configured sink"); + let routing = sink.as_mut().ok_or_else(|| { + options.storage( + &anyhow::anyhow!("configured sink is missing"), + &cursor, + attempts, + Work::Resubmit, + ) + })?; routing.configure(¤t); let count = routing .emit(delivery.value()) @@ -360,8 +390,8 @@ impl Consumer { Ok(Some(false)) => {} Ok(Some(true)) => { progress.send_replace(cursor.completed_position()); - *options.error.lock().unwrap() = None; - *options.recovery.lock().unwrap() = None; + *options.error.lock() = None; + *options.recovery.lock() = None; attempts = 0; recorded = None; } @@ -420,6 +450,7 @@ mod tests { accepted: AtomicU64, reject: AtomicBool, unauthorized: AtomicBool, + commit_fail: AtomicBool, } struct Factory(Arc); struct CheckedSink { @@ -458,6 +489,9 @@ mod tests { .into()); } self.record.accepted.fetch_add(pending, Ordering::SeqCst); + if self.record.commit_fail.swap(false, Ordering::SeqCst) { + return Err(CommitFailure(anyhow::anyhow!("injected ledger failure")).into()); + } Ok(()) } } @@ -531,7 +565,7 @@ mod tests { wait_for(|| record.accepted.load(Ordering::SeqCst) == 1).await; wait_for(|| *consumer.progress.borrow() == 1).await; assert_eq!(record.submitted.load(Ordering::SeqCst), 2); - assert!(consumer.error.lock().unwrap().is_none()); + assert!(consumer.error.lock().is_none()); assert!(journal .failure(&serde_json::to_string(&route).unwrap()) .await @@ -539,6 +573,30 @@ mod tests { .is_none()); consumer.shutdown().await; } + #[tokio::test] + async fn ledger_recovery_does_not_resubmit_accepted_work() { + let tmp = tempfile::tempdir().unwrap(); + let journal = SpanJournal::open(&tmp.path().join("spans.sqlite")) + .await + .unwrap(); + stored(&journal).await; + let route = SessionRoute::default(); + let record = Arc::new(Recording::default()); + record.commit_fail.store(true, Ordering::SeqCst); + let consumer = Consumer::spawn( + journal.clone(), + journal.subscribe(&route, 0).await.unwrap(), + options(tmp.path(), &route, record.clone()), + ); + wait_for(|| consumer.error.lock().is_some()).await; + assert_eq!(*consumer.progress.borrow(), 0); + assert_eq!(record.accepted.load(Ordering::SeqCst), 1); + wait_for(|| *consumer.progress.borrow() == 1).await; + assert_eq!(record.submitted.load(Ordering::SeqCst), 1); + assert_eq!(record.accepted.load(Ordering::SeqCst), 1); + consumer.shutdown().await; + } + #[tokio::test] async fn head_write_recovery_retries_only_completion() { let tmp = tempfile::tempdir().unwrap(); @@ -555,7 +613,7 @@ mod tests { cursor, options(tmp.path(), &route, record.clone()), ); - wait_for(|| consumer.error.lock().unwrap().is_some()).await; + wait_for(|| consumer.error.lock().is_some()).await; assert_eq!(*consumer.progress.borrow(), 0); assert_eq!(record.accepted.load(Ordering::SeqCst), 1); db.execute_batch("DROP TRIGGER fail_head").unwrap(); @@ -584,7 +642,7 @@ mod tests { journal.subscribe(&route, 0).await.unwrap(), options(tmp.path(), &route, record.clone()), ); - wait_for(|| consumer.error.lock().unwrap().is_some()).await; + wait_for(|| consumer.error.lock().is_some()).await; consumer.shutdown().await; let evidence = journal.failure(&key).await.unwrap().unwrap(); assert!(!evidence.contains("never-persist-this")); @@ -671,7 +729,7 @@ mod tests { journal.subscribe(&route, 0).await.unwrap(), options(tmp.path(), &route, record.clone()), ); - wait_for(|| consumer.error.lock().unwrap().is_some()).await; + wait_for(|| consumer.error.lock().is_some()).await; let failure: Failure = serde_json::from_str( &journal .failure(&serde_json::to_string(&route).unwrap()) diff --git a/bt-daemon/src/pipeline/recovery.rs b/bt-daemon/src/pipeline/recovery.rs index 42038c61..97b82412 100644 --- a/bt-daemon/src/pipeline/recovery.rs +++ b/bt-daemon/src/pipeline/recovery.rs @@ -73,11 +73,8 @@ pub(crate) fn now_ms() -> i64 { .as_millis() .min(i64::MAX as u128) as i64 } -fn digest(value: &impl Serialize) -> String { - format!( - "{:x}", - Sha256::digest(serde_json::to_vec(value).expect("serializable condition evidence")) - ) +fn digest(value: &impl std::fmt::Debug) -> String { + format!("{:x}", Sha256::digest(format!("{value:?}"))) } fn chain(config: &SessionConfig) -> String { digest(&config.span_plugins) @@ -86,7 +83,12 @@ fn configuration(config: &SessionConfig) -> String { let processors: Vec<_> = config .span_plugins .iter() - .map(|path| (path, crate::span_processor::file_digest(path))) + .map(|path| { + ( + format!("{path:?}"), + crate::span_processor::file_digest(path), + ) + }) .collect(); digest( &serde_json::json!({"destination":config.destination,"metadata":config.additional_metadata, diff --git a/bt-daemon/src/pipeline/translation.rs b/bt-daemon/src/pipeline/translation.rs index 89c37a56..ea98f134 100644 --- a/bt-daemon/src/pipeline/translation.rs +++ b/bt-daemon/src/pipeline/translation.rs @@ -201,9 +201,23 @@ impl Producer { let mut attempts = 0; loop { if let State::Recovering(mut failure) = self.state { - *self.error.lock().unwrap() = Some(failure.message()); - *self.recovery.lock().unwrap() = Some(failure.info(false)); - let encoded = serde_json::to_string(&failure).expect("source failure evidence"); + *self.error.lock() = Some(failure.message()); + *self.recovery.lock() = Some(failure.info(false)); + let encoded = match serde_json::to_string(&failure) { + Ok(encoded) => encoded, + Err(error) => { + self.state = State::Recovering(Failure::revision( + Boundary::Journal, + failure.position, + format!( + "{}; recovery evidence cannot be encoded: {error}", + failure.message() + ), + &self.options.version, + )); + continue; + } + }; if recorded.as_ref() != Some(&encoded) && super::recovery::now_ms() >= record_retry { match self.journal.set_source_failure(Some(encoded.clone())).await { @@ -242,7 +256,7 @@ impl Producer { { continue; } - *self.recovery.lock().unwrap() = Some(failure.info(true)); + *self.recovery.lock() = Some(failure.info(true)); attempts = failure.attempt.saturating_add(1); self.state = State::Processing; let result = async { @@ -267,8 +281,8 @@ impl Producer { self.state = State::Recovering(self.failure(error, attempts)); continue; } - *self.error.lock().unwrap() = None; - *self.recovery.lock().unwrap() = None; + *self.error.lock() = None; + *self.recovery.lock() = None; recorded = None; attempts = 0; continue; @@ -513,7 +527,10 @@ impl Producer { /// Each chunk stays owned until its local write succeeds. A commit retry /// cannot call translation or advance to the following event. async fn publish(&mut self) -> anyhow::Result<()> { - let mut pending = self.publication.take().expect("pending publication"); + let mut pending = self + .publication + .take() + .ok_or_else(|| anyhow::anyhow!("pending publication is missing"))?; let result = async { let publish = pending.input > self.committed; let ctx = SessionCtx { @@ -532,16 +549,14 @@ impl Producer { for op in ops.iter_mut() { 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) = &pending.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(|| { + pending.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(); } @@ -731,6 +746,7 @@ mod tests { reader: None, observers: Vec::new(), origins: HashMap::new(), + capture: None, bounds: HashMap::new(), position: 0, committed: 0, diff --git a/bt-daemon/src/server/rpc.rs b/bt-daemon/src/server/rpc.rs index 0738a921..0f16c2d1 100644 --- a/bt-daemon/src/server/rpc.rs +++ b/bt-daemon/src/server/rpc.rs @@ -338,9 +338,8 @@ impl Daemon { recovery: s .recovery .lock() - .unwrap() .clone() - .or_else(|| s.source_recovery.lock().unwrap().clone()), + .or_else(|| s.source_recovery.lock().clone()), spans_emitted: s.counters.spans_emitted.load(Ordering::Relaxed), permalink: s.permalink.lock().clone(), last_error: s diff --git a/bt-daemon/src/server/sessions.rs b/bt-daemon/src/server/sessions.rs index 6f9a3479..dc2ff2f3 100644 --- a/bt-daemon/src/server/sessions.rs +++ b/bt-daemon/src/server/sessions.rs @@ -51,7 +51,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(); Ok(match existing { Some(translation) => translation, None => { @@ -72,7 +72,6 @@ impl Daemon { .await?; self.translations .lock() - .unwrap() .insert(storage_key, translation.clone()); translation } @@ -121,7 +120,7 @@ impl Daemon { translation.advance(replay_through).await?; let _ = translation.barrier().await; let cursor = translation.journal.subscribe(route, legacy_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()); }