From a16caaa6519df701c2b0528ae5a2811beb9db489 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Thu, 15 Oct 2026 20:51:46 +0800 Subject: [PATCH 1/4] Hide route journal behind input port --- bt-daemon/src/dispatch.rs | 9 +- bt-daemon/src/pipeline/consumer.rs | 162 ++++++++++++-------------- bt-daemon/src/pipeline/delivery.rs | 8 ++ bt-daemon/src/pipeline/journal.rs | 110 ++++++++++++++++- bt-daemon/src/pipeline/translation.rs | 14 ++- bt-daemon/src/server/sessions.rs | 4 +- 6 files changed, 209 insertions(+), 98 deletions(-) diff --git a/bt-daemon/src/dispatch.rs b/bt-daemon/src/dispatch.rs index 04ffb939..c331a7ad 100644 --- a/bt-daemon/src/dispatch.rs +++ b/bt-daemon/src/dispatch.rs @@ -46,7 +46,7 @@ pub(crate) struct SessionOptions { pub config: crate::wire::SessionConfig, pub data_dir: PathBuf, pub translation: Arc, - pub cursor: crate::pipeline::journal::Cursor, + pub input: Box, pub auth_provider: Option>, } @@ -75,11 +75,12 @@ impl Session { let source = options.source.clone(); let observer_key = format!( "{}\u{1f}{}\u{1f}{}", - options.source, options.session_id, options.cursor.route + options.source, + options.session_id, + options.input.route_key() ); let route = Consumer::spawn( - options.translation.journal.clone(), - options.cursor, + options.input, ConsumerOptions { source: options.source.clone(), session_id: options.session_id.clone(), diff --git a/bt-daemon/src/pipeline/consumer.rs b/bt-daemon/src/pipeline/consumer.rs index 850ff6eb..bb12ca97 100644 --- a/bt-daemon/src/pipeline/consumer.rs +++ b/bt-daemon/src/pipeline/consumer.rs @@ -1,7 +1,7 @@ //! One route owns processing and recovery at its durable SpanOp head. use super::{ delivery::InputPort, - journal::{Cursor, SpanJournal}, + journal::RouteInputPort, recovery::{Boundary, Failure, Recovery, RecoveryOutcome}, routing::RoutingSink, Delivery, Stage, @@ -70,16 +70,11 @@ impl ConsumerOptions { sink.configure(config); Ok(sink) } - async fn hold(&self, journal: &SpanJournal, cursor: &Cursor, failure: &mut Failure) -> bool { - failure.through.get_or_insert(cursor.position()); + async fn hold(&self, input: &dyn RouteInputPort, failure: &mut Failure) -> bool { + failure.through.get_or_insert(input.position()); *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 - { + if let Err(error) = input.set_recovery(Some(failure)).await { tracing::warn!(%error,"route recovery evidence could not be persisted; original failure retained"); return false; } @@ -114,10 +109,16 @@ impl ConsumerOptions { Ok(()) } - fn storage(&self, error: &anyhow::Error, cursor: &Cursor, attempt: u32, work: Work) -> Held { + fn storage( + &self, + error: &anyhow::Error, + input: &dyn RouteInputPort, + attempt: u32, + work: Work, + ) -> Held { let mut failure = Failure::storage( error, - cursor.completed_position() + 1, + input.completed_position() + 1, &self.version, attempt, ); @@ -132,7 +133,7 @@ impl ConsumerOptions { fn routing( &self, error: &anyhow::Error, - cursor: &Cursor, + input: &dyn RouteInputPort, config: &SessionConfig, attempt: u32, ) -> Held { @@ -145,7 +146,7 @@ impl ConsumerOptions { } } Box::new(( - Failure::routing(error, cursor.completed_position() + 1, config, attempt), + Failure::routing(error, input.completed_position() + 1, config, attempt), Work::Resubmit, )) } @@ -154,7 +155,7 @@ impl ConsumerOptions { // rejected batch is resubmitted; accepted work retains its local retry point. async fn finish( &self, - cursor: &mut Cursor, + input: &mut dyn RouteInputPort, sink: &mut Option, config: &SessionConfig, attempt: u32, @@ -180,9 +181,9 @@ impl ConsumerOptions { })?; sink.flush().await.map_err(|error| { if error.downcast_ref::().is_some() { - self.storage(&error, cursor, attempt, Work::Flush) + self.storage(&error, input, attempt, Work::Flush) } else { - self.routing(&error, cursor, config, attempt) + self.routing(&error, input, config, attempt) } })?; *self.permalink.lock() = sink.permalink(); @@ -191,31 +192,31 @@ impl ConsumerOptions { if sink.has_pending_delivery() { return Ok(false); } - Some(cursor.pending_prefix()) + Some(input.pending_prefix()) } }; if let Some(delivery) = completion { delivery.into_completed().await.map_err(|unfinished| { self.storage( &unfinished.error, - cursor, + input, attempt, Work::Complete(unfinished.delivery), ) })?; } - cursor + input .rewind() .await - .map_err(|error| self.storage(&error, cursor, attempt, Work::Resubmit))?; + .map_err(|error| self.storage(&error, input, attempt, Work::Resubmit))?; Ok(completed) } } impl Consumer { - pub fn spawn(journal: SpanJournal, mut cursor: Cursor, options: ConsumerOptions) -> Arc { + pub fn spawn(mut input: Box, options: ConsumerOptions) -> Arc { let (config, mut settings) = watch::channel(options.config.clone()); - let (progress, receive) = watch::channel(cursor.position()); + let (progress, receive) = watch::channel(input.position()); let stop = Arc::new(Notify::new()); let handle = Arc::new(Self { task: Mutex::new(None), @@ -226,15 +227,10 @@ impl Consumer { }); let task = tokio::spawn(async move { 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, - ))), + let mut held = match input.recovery(&options.config).await { + Ok(Some(failure)) => Some(Box::new((failure, Work::Resubmit))), Ok(None) => None, - Err(error) => Some(options.storage(&error, &cursor, 0, Work::Resubmit)), + Err(error) => Some(options.storage(&error, &input, 0, Work::Resubmit)), }; let mut attempts = 0; let mut recorded = None; @@ -243,15 +239,14 @@ impl Consumer { 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() + && input.completed_position() >= failure.through.unwrap_or(failure.position) + && input.set_recovery(None).await.is_ok() { *options.error.lock() = None; *options.recovery.lock() = None; continue; } - failure.through.get_or_insert(cursor.position()); + failure.through.get_or_insert(input.position()); let encoded = match serde_json::to_string(&failure) { Ok(encoded) => encoded, Err(error) => { @@ -273,13 +268,13 @@ impl Consumer { if recorded.as_ref() != Some(&encoded) && super::recovery::now_ms() >= record_retry { - if options.hold(&journal, &cursor, &mut failure).await { + if options.hold(&input, &mut failure).await { recorded = Some(encoded.clone()); } else { record_retry = super::recovery::now_ms().saturating_add(1_000); } } - progress.send_replace(cursor.completed_position()); + progress.send_replace(input.completed_position()); tokio::select! { _=stop.notified()=>break, _=tokio::time::sleep(std::time::Duration::from_millis(250))=>{}, @@ -303,10 +298,12 @@ impl Consumer { *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 { + if let Err(error) = + options.credentials(input.route_key(), &mut current).await + { held = Some(Box::new(( Failure::credentials( - cursor.completed_position() + 1, + input.completed_position() + 1, error.to_string(), attempts, ), @@ -317,11 +314,11 @@ impl Consumer { config.send_replace(current.clone()); } match options - .finish(&mut cursor, &mut sink, ¤t, attempts, work) + .finish(&mut input, &mut sink, ¤t, attempts, work) .await { Ok(completed) => { - progress.send_replace(cursor.completed_position()); + progress.send_replace(input.completed_position()); if completed { *options.error.lock() = None; *options.recovery.lock() = None; @@ -333,57 +330,54 @@ impl Consumer { continue; } let attempt = async { - let tail = journal.tail().await.map_err(|error| { - options.storage(&error, &cursor, attempts, Work::Resubmit) + let tail = input.tail().await.map_err(|error| { + options.storage(&error, &input, attempts, Work::Resubmit) })?; options.counters.queued.store( - tail.saturating_sub(cursor.completed_position()), + tail.saturating_sub(input.completed_position()), Ordering::Relaxed, ); let received = tokio::select! { - result=InputPort::receive(&mut cursor)=>result, + result=InputPort::receive(&mut input)=>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) + options.storage(&error, &input, attempts, Work::Resubmit) })?; let Some(delivery) = delivery else { return Ok(None); }; let current = settings.borrow_and_update().clone(); if sink.is_none() { - journal - .bind(&cursor.route, ¤t) - .await - .map_err(|error| { - if error - .downcast_ref::() - .is_some() - { - Box::new(( - Failure::legacy( - error.to_string(), - cursor.completed_position() + 1, - ¤t, - ), - Work::Resubmit, - )) - } else { - options.storage(&error, &cursor, attempts, Work::Resubmit) - } - })?; + input.bind(¤t).await.map_err(|error| { + if error + .downcast_ref::() + .is_some() + { + Box::new(( + Failure::legacy( + error.to_string(), + input.completed_position() + 1, + ¤t, + ), + Work::Resubmit, + )) + } else { + options.storage(&error, &input, attempts, Work::Resubmit) + } + })?; sink = Some(options.sink(¤t).await.map_err(|error| { if error.downcast_ref::().is_some() || error.downcast_ref::().is_some() { let mut held = - options.storage(&error, &cursor, attempts, Work::Resubmit); + options.storage(&error, &input, attempts, Work::Resubmit); held.0.boundary = Boundary::Ledger; held } else { - options.routing(&error, &cursor, ¤t, attempts) + options.routing(&error, &input, ¤t, attempts) } })?); } @@ -399,14 +393,14 @@ impl Consumer { let count = routing .emit(delivery.value()) .await - .map_err(|error| options.routing(&error, &cursor, ¤t, attempts))?; + .map_err(|error| options.routing(&error, &input, ¤t, attempts))?; options .counters .spans_emitted .fetch_add(count, Ordering::Relaxed); drop(delivery); options - .finish(&mut cursor, &mut sink, ¤t, attempts, Work::Flush) + .finish(&mut input, &mut sink, ¤t, attempts, Work::Flush) .await .map(Some) }; @@ -418,7 +412,7 @@ impl Consumer { Ok(None) => break, Ok(Some(false)) => {} Ok(Some(true)) => { - progress.send_replace(cursor.completed_position()); + progress.send_replace(input.completed_position()); *options.error.lock() = None; *options.recovery.lock() = None; attempts = 0; @@ -468,6 +462,7 @@ impl Consumer { mod tests { use super::*; use crate::{ + pipeline::journal::SpanJournal, translate::{SpanOp, SpanRow}, wire::BackendAuth, }; @@ -587,8 +582,7 @@ mod tests { record.reject.store(true, Ordering::SeqCst); let cursor = journal.subscribe(&route, 0).await.unwrap(); let consumer = Consumer::spawn( - journal.clone(), - cursor, + Box::new(cursor), options(tmp.path(), &route, record.clone()), ); wait_for(|| record.accepted.load(Ordering::SeqCst) == 1).await; @@ -638,8 +632,7 @@ mod tests { 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, + Box::new(cursor), options(tmp.path(), &route, record.clone()), ); wait_for(|| consumer.error.lock().is_some()).await; @@ -667,8 +660,7 @@ mod tests { 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(), + Box::new(journal.subscribe(&route, 0).await.unwrap()), options(tmp.path(), &route, record.clone()), ); wait_for(|| consumer.error.lock().is_some()).await; @@ -676,8 +668,7 @@ mod tests { 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(), + Box::new(journal.subscribe(&route, 0).await.unwrap()), options(tmp.path(), &route, record.clone()), ); tokio::time::sleep(std::time::Duration::from_millis(600)).await; @@ -725,11 +716,7 @@ mod tests { 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, - ); + let consumer = Consumer::spawn(Box::new(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"); @@ -773,11 +760,7 @@ mod tests { crate::BraintrustSinkConfig::default(), )); let mut config = opts.config.clone(); - let consumer = Consumer::spawn( - journal.clone(), - journal.subscribe(&route, 0).await.unwrap(), - opts, - ); + let consumer = Consumer::spawn(Box::new(journal.subscribe(&route, 0).await.unwrap()), opts); wait_for(|| consumer.error.lock().is_some()).await; assert_eq!(*consumer.progress.borrow(), 0); drop(rejection); @@ -817,8 +800,7 @@ mod tests { }; let record = Arc::new(Recording::default()); let consumer = Consumer::spawn( - journal.clone(), - journal.subscribe(&route, 0).await.unwrap(), + Box::new(journal.subscribe(&route, 0).await.unwrap()), options(tmp.path(), &route, record.clone()), ); wait_for(|| consumer.error.lock().is_some()).await; diff --git a/bt-daemon/src/pipeline/delivery.rs b/bt-daemon/src/pipeline/delivery.rs index 41a48e69..7c636009 100644 --- a/bt-daemon/src/pipeline/delivery.rs +++ b/bt-daemon/src/pipeline/delivery.rs @@ -73,6 +73,14 @@ impl InputPort for mpsc::Receiver> { } } +#[async_trait::async_trait] +impl InputPort for Box { + type Item = T::Item; + async fn receive(&mut self) -> anyhow::Result>> { + (**self).receive().await + } +} + #[async_trait::async_trait] impl OutputPort for mpsc::Sender> { type Item = T; diff --git a/bt-daemon/src/pipeline/journal.rs b/bt-daemon/src/pipeline/journal.rs index 8ab057d9..1d30f4cf 100644 --- a/bt-daemon/src/pipeline/journal.rs +++ b/bt-daemon/src/pipeline/journal.rs @@ -1,5 +1,5 @@ //! Concrete durable channel: one ordered SpanOp log and a head per route. -use super::{Completion, Delivery}; +use super::{recovery::Failure, Completion, Delivery}; use crate::{ translate::SpanOp, wire::{SessionConfig, SessionRoute}, @@ -428,6 +428,65 @@ pub(crate) struct Cursor { changes: watch::Receiver, } +/// The route-facing view of the durable SpanOp channel. +/// +/// Consumers depend on delivery and head operations, while `SpanJournal` and +/// SQLite remain private to this adapter. +#[async_trait::async_trait] +pub(crate) trait RouteInputPort: + super::delivery::InputPort> + Sync +{ + fn route_key(&self) -> &str; + fn position(&self) -> u64; + fn completed_position(&self) -> u64; + fn pending_prefix(&self) -> Delivery<()>; + async fn rewind(&mut self) -> anyhow::Result<()>; + async fn tail(&self) -> anyhow::Result; + async fn recovery(&self, config: &SessionConfig) -> anyhow::Result>; + async fn set_recovery(&self, failure: Option<&Failure>) -> anyhow::Result<()>; + async fn bind(&self, config: &SessionConfig) -> anyhow::Result<()>; +} + +#[derive(Debug)] +pub(crate) struct RouteBindingChanged; +impl std::fmt::Display for RouteBindingChanged { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("durable route binding changed") + } +} +impl std::error::Error for RouteBindingChanged {} + +#[async_trait::async_trait] +impl RouteInputPort for Box { + fn route_key(&self) -> &str { + (**self).route_key() + } + fn position(&self) -> u64 { + (**self).position() + } + fn completed_position(&self) -> u64 { + (**self).completed_position() + } + fn pending_prefix(&self) -> Delivery<()> { + (**self).pending_prefix() + } + async fn rewind(&mut self) -> anyhow::Result<()> { + (**self).rewind().await + } + async fn tail(&self) -> anyhow::Result { + (**self).tail().await + } + async fn recovery(&self, config: &SessionConfig) -> anyhow::Result> { + (**self).recovery(config).await + } + async fn set_recovery(&self, failure: Option<&Failure>) -> anyhow::Result<()> { + (**self).set_recovery(failure).await + } + async fn bind(&self, config: &SessionConfig) -> anyhow::Result<()> { + (**self).bind(config).await + } +} + impl Cursor { pub async fn next_batch( &mut self, @@ -520,6 +579,55 @@ impl super::delivery::InputPort for Cursor { } } +#[async_trait::async_trait] +impl RouteInputPort for Cursor { + fn route_key(&self) -> &str { + &self.route + } + fn position(&self) -> u64 { + Cursor::position(self) + } + fn completed_position(&self) -> u64 { + Cursor::completed_position(self) + } + fn pending_prefix(&self) -> Delivery<()> { + Cursor::pending_prefix(self) + } + async fn rewind(&mut self) -> anyhow::Result<()> { + Cursor::rewind(self).await + } + async fn tail(&self) -> anyhow::Result { + self.journal.tail().await + } + async fn recovery(&self, config: &SessionConfig) -> anyhow::Result> { + let Some(raw) = self.journal.failure(&self.route).await? else { + return Ok(None); + }; + Ok(Some(serde_json::from_str(&raw).unwrap_or_else(|_| { + Failure::legacy(raw, self.completed_position() + 1, config) + }))) + } + async fn set_recovery(&self, failure: Option<&Failure>) -> anyhow::Result<()> { + let failure = failure + .map(serde_json::to_string) + .transpose() + .map_err(anyhow::Error::from)?; + self.journal.set_failure(&self.route, failure).await + } + async fn bind(&self, config: &SessionConfig) -> anyhow::Result<()> { + self.journal + .bind(&self.route, config) + .await + .map_err(|error| { + if error.downcast_ref::().is_some() { + anyhow::Error::new(RouteBindingChanged) + } else { + error + } + }) + } +} + struct HeadCompletion { journal: SpanJournal, route: String, diff --git a/bt-daemon/src/pipeline/translation.rs b/bt-daemon/src/pipeline/translation.rs index 0966d709..74bff4fe 100644 --- a/bt-daemon/src/pipeline/translation.rs +++ b/bt-daemon/src/pipeline/translation.rs @@ -51,7 +51,7 @@ impl Progress { pub(crate) struct Translation { tx: mpsc::Sender, task: tokio::sync::Mutex>>, - pub journal: SpanJournal, + journal: SpanJournal, pub error: Arc>>, pub recovery: Arc>>, pub routes: AtomicUsize, @@ -169,6 +169,18 @@ impl Translation { rx.await .map_err(|_| "translation control lost".to_owned())? } + + /// Connect a route to the committed SpanOp stream without exposing its + /// storage adapter to the route actor. + pub async fn subscribe( + &self, + route: &SessionRoute, + legacy_through: u64, + ) -> anyhow::Result> { + Ok(Box::new( + self.journal.subscribe(route, legacy_through).await?, + )) + } } enum State { diff --git a/bt-daemon/src/server/sessions.rs b/bt-daemon/src/server/sessions.rs index 93da136f..763d4aa5 100644 --- a/bt-daemon/src/server/sessions.rs +++ b/bt-daemon/src/server/sessions.rs @@ -119,7 +119,7 @@ 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, legacy_through).await?; + let input = translation.subscribe(route, legacy_through).await?; let mut map = self.sessions.lock(); if let Some(s) = map.get(key) { return Ok(s.clone()); @@ -127,7 +127,7 @@ impl Daemon { let session = Session::spawn( SessionOptions { translation, - cursor, + input, auth_provider: self.route_auth_provider(key), session_id: env.session_id.clone(), source: env.source.clone(), From 6c5d0661ad61b63fc37f6a8253bddb59bf8c02a9 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Thu, 15 Oct 2026 21:03:43 +0800 Subject: [PATCH 2/4] Fix route port clippy lint --- bt-daemon/src/pipeline/journal.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/bt-daemon/src/pipeline/journal.rs b/bt-daemon/src/pipeline/journal.rs index 1d30f4cf..d0409c70 100644 --- a/bt-daemon/src/pipeline/journal.rs +++ b/bt-daemon/src/pipeline/journal.rs @@ -433,6 +433,7 @@ pub(crate) struct Cursor { /// Consumers depend on delivery and head operations, while `SpanJournal` and /// SQLite remain private to this adapter. #[async_trait::async_trait] +#[allow(clippy::double_must_use)] // async_trait marks its boxed future returns as must_use. pub(crate) trait RouteInputPort: super::delivery::InputPort> + Sync { From b0be27460c60d7d920c2ef6d641fdb71bbe47fe8 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Thu, 15 Oct 2026 23:37:48 +0800 Subject: [PATCH 3/4] Make journal replay transparent through InputPort --- bt-daemon/src/dispatch.rs | 48 +- bt-daemon/src/pipeline/consumer.rs | 513 +++++++++++----------- bt-daemon/src/pipeline/delivery.rs | 14 +- bt-daemon/src/pipeline/journal.rs | 385 ++++++++++------ bt-daemon/src/pipeline/recovery.rs | 32 +- bt-daemon/src/pipeline/translation.rs | 8 +- bt-daemon/src/server/correlation_store.rs | 8 +- bt-daemon/src/server/sessions.rs | 1 + 8 files changed, 578 insertions(+), 431 deletions(-) diff --git a/bt-daemon/src/dispatch.rs b/bt-daemon/src/dispatch.rs index c331a7ad..23794a2f 100644 --- a/bt-daemon/src/dispatch.rs +++ b/bt-daemon/src/dispatch.rs @@ -25,7 +25,6 @@ const QUEUE_CAPACITY: usize = 1024; #[derive(Default)] pub struct Counters { - pub queued: AtomicU64, pub spans_emitted: AtomicU64, } @@ -44,9 +43,10 @@ pub(crate) struct SessionOptions { pub source: String, pub bt_version: String, pub config: crate::wire::SessionConfig, + pub route: crate::wire::SessionRoute, pub data_dir: PathBuf, pub translation: Arc, - pub input: Box, + pub input: crate::pipeline::journal::Cursor, pub auth_provider: Option>, } @@ -57,6 +57,7 @@ pub struct Session { pub counters: Arc, pub last_error: Arc>>, source_progress: Arc, + route_progress: crate::pipeline::journal::Progress, pub recovery: Arc>>, pub source_recovery: Arc>>, pub source_error: Arc>>, @@ -73,20 +74,28 @@ impl Session { let last_error = Arc::new(Mutex::new(None)); let permalink = Arc::new(Mutex::new(None)); let source = options.source.clone(); + let (config, settings) = tokio::sync::watch::channel(options.config); + let subscription = options.input.into_input( + settings, + options.bt_version.clone(), + last_error.clone(), + recovery.clone(), + ); let observer_key = format!( "{}\u{1f}{}\u{1f}{}", options.source, options.session_id, - options.input.route_key() + serde_json::to_string(&options.route).expect("serializable route") ); let route = Consumer::spawn( - options.input, + subscription.input, ConsumerOptions { source: options.source.clone(), session_id: options.session_id.clone(), version: options.bt_version, data_dir: options.data_dir, - config: options.config, + config, + route: options.route, counters: counters.clone(), error: last_error.clone(), permalink: permalink.clone(), @@ -101,7 +110,8 @@ impl Session { let source_progress = producer.progress.clone(); let errors = last_error.clone(); let source_errors = source_error.clone(); - let queued = counters.clone(); + let route_progress = subscription.progress; + let progress = route_progress.clone(); tokio::spawn(async move { while let Some(message) = rx.recv().await { let finalize = matches!(&message, SessionMsg::Finalize(_)); @@ -129,24 +139,23 @@ impl Session { crate::journal::ControlKind::Checkpoint }; let target = producer.control(kind).await; - let route = route.clone(); - let queued = queued.clone(); + let progress = progress.clone(); let errors = errors.clone(); let source_errors = source_errors.clone(); tokio::spawn(async move { let mut reply = reply; let waiting = async { match target { - Ok(tail) => match route.through(tail).await { + Ok(tail) => match progress.through(tail, &errors).await { Ok(()) => 0, Err(error) => { *errors.lock() = Some(error); - queued.queued.load(Ordering::Relaxed).max(1) + progress.pending().max(1) } }, Err(error) => { *source_errors.lock() = Some(error); - queued.queued.load(Ordering::Relaxed).max(1) + progress.pending().max(1) } } }; @@ -166,7 +175,7 @@ impl Session { if let Some(tail) = target { let _ = tokio::time::timeout( std::time::Duration::from_secs(2), - route.through(tail), + progress.through(tail, &errors), ) .await; } @@ -189,6 +198,7 @@ impl Session { last_error, source_error, source_progress, + route_progress, recovery, source_recovery, permalink, @@ -197,9 +207,8 @@ impl Session { } pub(crate) fn pending(&self) -> u64 { - self.counters - .queued - .load(Ordering::Relaxed) + self.route_progress + .pending() .saturating_add(self.source_progress.pending()) } @@ -246,11 +255,11 @@ impl Session { /// Returns `(flushed, pending)`. pub async fn flush(&self, timeout: std::time::Duration) -> (bool, u64) { let Ok(reply_rx) = self.enqueue_flush().await else { - return (false, self.counters.queued.load(Ordering::Relaxed)); + return (false, self.pending()); }; match tokio::time::timeout(timeout, reply_rx).await { Ok(Ok(pending)) => (pending == 0, pending), - _ => (false, self.counters.queued.load(Ordering::Relaxed)), + _ => (false, self.pending()), } } @@ -260,11 +269,11 @@ impl Session { self.touch(); let (reply_tx, reply_rx) = oneshot::channel(); if self.tx.send(SessionMsg::Finalize(reply_tx)).await.is_err() { - return (false, self.counters.queued.load(Ordering::Relaxed)); + return (false, self.pending()); } match tokio::time::timeout(timeout, reply_rx).await { Ok(Ok(pending)) => (pending == 0, pending), - _ => (false, self.counters.queued.load(Ordering::Relaxed)), + _ => (false, self.pending()), } } @@ -590,6 +599,7 @@ mod tests { recovery: Arc::new(Mutex::new(None)), source_recovery: Arc::new(Mutex::new(None)), source_progress: Arc::new(crate::pipeline::translation::Progress::default()), + route_progress: crate::pipeline::journal::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 bb12ca97..6be625e2 100644 --- a/bt-daemon/src/pipeline/consumer.rs +++ b/bt-daemon/src/pipeline/consumer.rs @@ -1,7 +1,6 @@ -//! One route owns processing and recovery at its durable SpanOp head. +//! Route processing consumes managed deliveries independently of their source. use super::{ delivery::InputPort, - journal::RouteInputPort, recovery::{Boundary, Failure, Recovery, RecoveryOutcome}, routing::RoutingSink, Delivery, Stage, @@ -10,36 +9,36 @@ use crate::{ delivery_ledger::{CommitFailure, LedgerSink}, dispatch::Counters, sink::{Sink, SinkFactory}, + translate::SpanOp, wire::{RecoveryInfo, SessionConfig, SessionRoute}, AuthProvider, AuthResolveReason, }; -use parking_lot::Mutex; use std::{ + collections::VecDeque, path::PathBuf, - sync::{atomic::Ordering, Arc}, + sync::{atomic::Ordering, Arc, Mutex}, }; use tokio::sync::{watch, Notify}; enum Work { Resubmit, Flush, - Complete(Delivery<()>), + Complete, } type Held = Box<(Failure, Work)>; pub(crate) struct Consumer { task: Mutex>>, config: watch::Sender, - progress: watch::Receiver, stop: Arc, - error: Arc>>, } pub(crate) struct ConsumerOptions { pub source: String, pub session_id: String, pub version: String, pub data_dir: PathBuf, - pub config: SessionConfig, + pub route: SessionRoute, + pub config: watch::Sender, pub counters: Arc, pub error: Arc>>, pub recovery: Arc>>, @@ -70,22 +69,25 @@ impl ConsumerOptions { sink.configure(config); Ok(sink) } - async fn hold(&self, input: &dyn RouteInputPort, failure: &mut Failure) -> bool { - failure.through.get_or_insert(input.position()); - *self.error.lock() = Some(failure.message()); - *self.recovery.lock() = Some(failure.info(false)); - if let Err(error) = input.set_recovery(Some(failure)).await { - tracing::warn!(%error,"route recovery evidence could not be persisted; original failure retained"); + async fn hold(&self, pending: &mut VecDeque>, failure: &mut Failure) -> bool { + let result = match pending.front_mut() { + Some(delivery) => delivery.failed(failure).await, + None => Ok(()), // Input errors are reported by the source itself. + }; + *self.error.lock().unwrap() = Some(failure.message()); + *self.recovery.lock().unwrap() = Some(failure.info(false)); + if let Err(error) = result { + tracing::warn!(%error,"delivery recovery evidence could not be reported; original failure retained"); return false; } true } - async fn credentials(&self, route: &str, config: &mut SessionConfig) -> anyhow::Result<()> { + async fn credentials(&self, 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 route = &self.route; let lease = crate::server::resolve_route_auth( provider.as_ref(), &route.auth, @@ -109,34 +111,17 @@ impl ConsumerOptions { Ok(()) } - fn storage( - &self, - error: &anyhow::Error, - input: &dyn RouteInputPort, - attempt: u32, - work: Work, - ) -> Held { - let mut failure = Failure::storage( - error, - input.completed_position() + 1, - &self.version, - attempt, - ); + fn storage(&self, error: &anyhow::Error, attempt: u32, work: Work) -> Held { + let mut failure = Failure::storage(error, &self.version, attempt); failure.boundary = match work { - Work::Complete(_) => Boundary::Completion, + Work::Complete => Boundary::Completion, Work::Flush => Boundary::Ledger, Work::Resubmit => failure.boundary, }; Box::new((failure, work)) } - fn routing( - &self, - error: &anyhow::Error, - input: &dyn RouteInputPort, - config: &SessionConfig, - attempt: u32, - ) -> Held { + fn routing(&self, error: &anyhow::Error, config: &SessionConfig, attempt: u32) -> Held { if let Some(partial) = error.downcast_ref::() { self.counters .spans_emitted @@ -145,136 +130,87 @@ impl ConsumerOptions { tracing::warn!(session_id=%self.session_id, %error, "route processing failed"); } } - Box::new(( - Failure::routing(error, input.completed_position() + 1, config, attempt), - Work::Resubmit, - )) + Box::new((Failure::routing(error, 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. + // Keep the original acknowledgement obligations until checked delivery has + // resolved every deferred row. Completion failures retry only acknowledgement. async fn finish( &self, - input: &mut dyn RouteInputPort, + pending: &mut VecDeque>, sink: &mut Option, config: &SessionConfig, attempt: u32, work: Work, ) -> Result { - let completed = !matches!(work, Work::Resubmit); - let completion = match work { + 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. + // Dropping unresolved obligations asks their source to replay. + // Recreate the consumed SDK batch and its in-memory ledger too. *sink = None; - None + pending.clear(); + return Ok(false); } - Work::Complete(delivery) => Some(delivery), Work::Flush => { - let sink = sink.as_mut().ok_or_else(|| { - self.storage( - &anyhow::anyhow!("accepted delivery has no sink"), - cursor, - attempt, - 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, input, attempt, Work::Flush) + self.storage(&error, attempt, Work::Flush) } else { - self.routing(&error, input, config, attempt) + self.routing(&error, config, attempt) } })?; - *self.permalink.lock() = sink.permalink(); - // Deferred roots keep their prefix unresolved until a child - // releases them; the cursor can read ahead without buffering ops. + *self.permalink.lock().unwrap() = sink.permalink(); if sink.has_pending_delivery() { return Ok(false); } - Some(input.pending_prefix()) } - }; - if let Some(delivery) = completion { - delivery.into_completed().await.map_err(|unfinished| { - self.storage( - &unfinished.error, - input, - attempt, - Work::Complete(unfinished.delivery), - ) - })?; + Work::Complete => {} } - input - .rewind() - .await - .map_err(|error| self.storage(&error, input, attempt, Work::Resubmit))?; - Ok(completed) + while let Some(delivery) = pending.pop_front() { + if let Err(unfinished) = delivery.into_completed().await { + pending.push_front(unfinished.delivery); + return Err(self.storage(&unfinished.error, attempt, Work::Complete)); + } + } + Ok(true) } } impl Consumer { - pub fn spawn(mut input: Box, options: ConsumerOptions) -> Arc { - let (config, mut settings) = watch::channel(options.config.clone()); - let (progress, receive) = watch::channel(input.position()); + pub fn spawn( + mut input: Box>>, + options: ConsumerOptions, + ) -> Arc { + let config = options.config.clone(); + let mut settings = config.subscribe(); let stop = Arc::new(Notify::new()); let handle = Arc::new(Self { task: Mutex::new(None), config: config.clone(), - progress: receive, stop: stop.clone(), - error: options.error.clone(), }); let task = tokio::spawn(async move { let mut sink = None; - let mut held = match input.recovery(&options.config).await { - Ok(Some(failure)) => Some(Box::new((failure, Work::Resubmit))), - Ok(None) => None, - Err(error) => Some(options.storage(&error, &input, 0, Work::Resubmit)), - }; + let mut pending = VecDeque::new(); + let mut held: Option = None; let mut attempts = 0; let mut recorded = None; let mut record_retry = 0; loop { - if let Some(pending) = held.take() { - let (mut failure, work) = *pending; - if matches!(failure.boundary, Boundary::Completion) - && input.completed_position() >= failure.through.unwrap_or(failure.position) - && input.set_recovery(None).await.is_ok() - { - *options.error.lock() = None; - *options.recovery.lock() = None; - continue; - } - failure.through.get_or_insert(input.position()); - 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 let Some(waiting) = held.take() { + let (mut failure, work) = *waiting; + let mut encoded = serde_json::to_string(&failure).expect("failure evidence"); if recorded.as_ref() != Some(&encoded) && super::recovery::now_ms() >= record_retry { - if options.hold(&input, &mut failure).await { + if options.hold(&mut pending, &mut failure).await { + encoded = serde_json::to_string(&failure).expect("failure evidence"); recorded = Some(encoded.clone()); } else { record_retry = super::recovery::now_ms().saturating_add(1_000); } } - progress.send_replace(input.completed_position()); tokio::select! { _=stop.notified()=>break, _=tokio::time::sleep(std::time::Duration::from_millis(250))=>{}, @@ -295,18 +231,12 @@ impl Consumer { held = Some(Box::new((failure, work))); continue; } - *options.recovery.lock() = Some(failure.info(true)); + *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(input.route_key(), &mut current).await - { + if let Err(error) = options.credentials(&mut current).await { held = Some(Box::new(( - Failure::credentials( - input.completed_position() + 1, - error.to_string(), - attempts, - ), + Failure::credentials(error.to_string(), attempts), work, ))); continue; @@ -314,14 +244,13 @@ impl Consumer { config.send_replace(current.clone()); } match options - .finish(&mut input, &mut sink, ¤t, attempts, work) + .finish(&mut pending, &mut sink, ¤t, attempts, work) .await { Ok(completed) => { - progress.send_replace(input.completed_position()); if completed { - *options.error.lock() = None; - *options.recovery.lock() = None; + *options.error.lock().unwrap() = None; + *options.recovery.lock().unwrap() = None; recorded = None; } } @@ -330,77 +259,52 @@ impl Consumer { continue; } let attempt = async { - let tail = input.tail().await.map_err(|error| { - options.storage(&error, &input, attempts, Work::Resubmit) - })?; - options.counters.queued.store( - tail.saturating_sub(input.completed_position()), - Ordering::Relaxed, - ); let received = tokio::select! { - result=InputPort::receive(&mut input)=>result, + result=input.receive()=>result, result=settings.changed()=> { return if result.is_err() { Ok(None) } else { Ok(Some(false)) }; }, }; - let delivery = received.map_err(|error| { - options.storage(&error, &input, attempts, Work::Resubmit) + let delivery = received.map_err(|error| match error.downcast::() { + Ok(failure) => Box::new((failure, Work::Resubmit)), + Err(error) => options.storage(&error, attempts, Work::Resubmit), })?; let Some(delivery) = delivery else { return Ok(None); }; let current = settings.borrow_and_update().clone(); - if sink.is_none() { - input.bind(¤t).await.map_err(|error| { - if error - .downcast_ref::() - .is_some() - { - Box::new(( - Failure::legacy( - error.to_string(), - input.completed_position() + 1, - ¤t, - ), - Work::Resubmit, - )) - } else { - options.storage(&error, &input, attempts, Work::Resubmit) - } - })?; - sink = Some(options.sink(¤t).await.map_err(|error| { - if error.downcast_ref::().is_some() - || error.downcast_ref::().is_some() - { - let mut held = - options.storage(&error, &input, attempts, Work::Resubmit); - held.0.boundary = Boundary::Ledger; - held - } else { - options.routing(&error, &input, ¤t, attempts) - } - })?); + let ops = delivery.value().as_slice(); + let result = async { + if sink.is_none() { + sink = Some(options.sink(¤t).await.map_err(|error| { + if error.downcast_ref::().is_some() + || error.downcast_ref::().is_some() + { + let mut failure = + options.storage(&error, attempts, Work::Resubmit); + failure.0.boundary = Boundary::Ledger; + failure + } else { + options.routing(&error, ¤t, attempts) + } + })?); + } + let routing = sink.as_mut().expect("configured sink"); + routing.configure(¤t); + routing + .emit(ops) + .await + .map_err(|error| options.routing(&error, ¤t, attempts)) } - 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()) - .await - .map_err(|error| options.routing(&error, &input, ¤t, attempts))?; + .await; + pending.push_back(delivery.map(|_| ())); + let count = result?; options .counters .spans_emitted .fetch_add(count, Ordering::Relaxed); - drop(delivery); options - .finish(&mut input, &mut sink, ¤t, attempts, Work::Flush) + .finish(&mut pending, &mut sink, ¤t, attempts, Work::Flush) .await .map(Some) }; @@ -412,9 +316,8 @@ impl Consumer { Ok(None) => break, Ok(Some(false)) => {} Ok(Some(true)) => { - progress.send_replace(input.completed_position()); - *options.error.lock() = None; - *options.recovery.lock() = None; + *options.error.lock().unwrap() = None; + *options.recovery.lock().unwrap() = None; attempts = 0; recorded = None; } @@ -422,7 +325,7 @@ impl Consumer { } } }); - *handle.task.lock() = Some(task); + *handle.task.lock().unwrap() = Some(task); handle } pub fn configure(&self, config: SessionConfig) { @@ -436,26 +339,11 @@ impl Consumer { } pub async fn shutdown(&self) { self.stop.notify_one(); - let task = self.task.lock().take(); + let task = self.task.lock().unwrap().take(); if let Some(task) = task { let _ = task.await; } } - pub async fn through(&self, tail: u64) -> Result<(), String> { - let mut progress = self.progress.clone(); - loop { - if *progress.borrow_and_update() >= tail { - return Ok(()); - } - if let Some(error) = self.error.lock().clone() { - return Err(error); - } - progress - .changed() - .await - .map_err(|_| "routing actor is gone".to_owned())?; - } - } } #[cfg(test)] @@ -474,7 +362,7 @@ mod tests { accepted: AtomicU64, reject: AtomicBool, unauthorized: AtomicBool, - commit_fail: AtomicBool, + defer_until: AtomicU64, } struct Factory(Arc); struct CheckedSink { @@ -499,6 +387,9 @@ mod tests { Ok(ops.len() as u64) } async fn flush(&mut self) -> anyhow::Result<()> { + if self.pending < self.record.defer_until.load(Ordering::SeqCst) { + return Ok(()); + } // 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) { @@ -513,11 +404,11 @@ 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(()) } + fn has_pending_delivery(&self) -> bool { + self.pending > 0 + } } fn options( dir: &std::path::Path, @@ -529,13 +420,15 @@ mod tests { session_id: "test".into(), version: "test".into(), data_dir: dir.into(), - config: route.with_auth(BackendAuth { + route: route.clone(), + config: watch::channel(route.with_auth(BackendAuth { token: "never-persist-this".into(), api_url: None, app_url: None, org_id: None, org_name: None, - }), + })) + .0, counters: Arc::new(Counters::default()), error: Arc::new(Mutex::new(None)), recovery: Arc::new(Mutex::new(None)), @@ -544,6 +437,27 @@ mod tests { auth_provider: None, } } + fn spawn( + cursor: crate::pipeline::journal::Cursor, + options: ConsumerOptions, + ) -> ( + Arc, + crate::pipeline::journal::Progress, + Arc>>, + ) { + let subscription = cursor.into_input( + options.config.subscribe(), + options.version.clone(), + options.error.clone(), + options.recovery.clone(), + ); + let error = options.error.clone(); + ( + Consumer::spawn(subscription.input, options), + subscription.progress, + error, + ) + } async fn wait_for(mut predicate: impl FnMut() -> bool) { tokio::time::timeout(std::time::Duration::from_secs(5), async { while !predicate() { @@ -581,42 +495,104 @@ mod tests { let record = Arc::new(Recording::default()); record.reject.store(true, Ordering::SeqCst); let cursor = journal.subscribe(&route, 0).await.unwrap(); - let consumer = Consumer::spawn( - Box::new(cursor), - options(tmp.path(), &route, record.clone()), + let opts = options(tmp.path(), &route, record.clone()); + let subscription = cursor.into_input( + opts.config.subscribe(), + opts.version.clone(), + opts.error.clone(), + opts.recovery.clone(), ); + let progress = subscription.progress; + let mut input = subscription.input; + let (tx, rx) = tokio::sync::mpsc::channel(1); + let forward = tokio::spawn(async move { + while let Ok(Some(delivery)) = input.receive().await { + if tx.send(delivery).await.is_err() { + break; + } + } + }); + let error = opts.error.clone(); + let consumer = Consumer::spawn(Box::new(rx), opts); wait_for(|| record.accepted.load(Ordering::SeqCst) == 1).await; - wait_for(|| *consumer.progress.borrow() == 1).await; + wait_for(|| progress.completed() == 1).await; assert_eq!(record.submitted.load(Ordering::SeqCst), 2); - assert!(consumer.error.lock().is_none()); + assert!(error.lock().unwrap().is_none()); assert!(journal .failure(&serde_json::to_string(&route).unwrap()) .await .unwrap() .is_none()); consumer.shutdown().await; + forward.abort(); + let _ = forward.await; + journal.close().await.unwrap(); } + struct Acknowledged(Arc); + #[async_trait::async_trait] + impl crate::pipeline::Completion for Acknowledged { + async fn complete(&mut self) -> anyhow::Result<()> { + self.0.fetch_add(1, Ordering::SeqCst); + Ok(()) + } + } + #[tokio::test] - async fn ledger_recovery_does_not_resubmit_accepted_work() { + async fn plain_channel_uses_the_same_consumer_and_completion_contract() { + let tmp = tempfile::tempdir().unwrap(); + let record = Arc::new(Recording::default()); + let acknowledged = Arc::new(AtomicU64::new(0)); + let opts = options(tmp.path(), &SessionRoute::default(), record.clone()); + let (tx, rx) = tokio::sync::mpsc::channel(1); + let consumer = Consumer::spawn(Box::new(rx), opts); + tx.send(Delivery::new( + vec![SpanOp::Insert(SpanRow { + span_id: "one".into(), + name: "one".into(), + root_span_id: "one".into(), + end_ms: Some(1), + ..Default::default() + })], + Acknowledged(acknowledged.clone()), + )) + .await + .unwrap_or_else(|_| panic!("consumer stopped")); + wait_for(|| acknowledged.load(Ordering::SeqCst) == 1).await; + assert_eq!(record.accepted.load(Ordering::SeqCst), 1); + consumer.shutdown().await; + } + + #[tokio::test] + async fn deferred_delivery_can_read_ahead_and_acknowledge_original_receipts() { let tmp = tempfile::tempdir().unwrap(); let journal = SpanJournal::open(&tmp.path().join("spans.sqlite")) .await .unwrap(); - stored(&journal).await; + let ops: Vec<_> = (0..129) + .map(|index| { + SpanOp::Insert(SpanRow { + span_id: index.to_string(), + name: index.to_string(), + root_span_id: index.to_string(), + end_ms: Some(1), + ..Default::default() + }) + }) + .collect(); + journal.begin(1, true).await.unwrap(); + journal.append(1, &ops).await.unwrap(); + journal.commit(1).await.unwrap(); let route = SessionRoute::default(); let record = Arc::new(Recording::default()); - record.commit_fail.store(true, Ordering::SeqCst); - let consumer = Consumer::spawn( - journal.clone(), + record.defer_until.store(129, Ordering::SeqCst); + let (consumer, progress, _error) = spawn( 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); + wait_for(|| progress.completed() == 129).await; + assert_eq!(record.submitted.load(Ordering::SeqCst), 129); + assert_eq!(record.accepted.load(Ordering::SeqCst), 129); + assert_eq!(progress.pending(), 0); consumer.shutdown().await; } @@ -631,15 +607,13 @@ mod tests { 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( - Box::new(cursor), - options(tmp.path(), &route, record.clone()), - ); - wait_for(|| consumer.error.lock().is_some()).await; - assert_eq!(*consumer.progress.borrow(), 0); + let (consumer, progress, error) = + spawn(cursor, options(tmp.path(), &route, record.clone())); + wait_for(|| error.lock().unwrap().is_some()).await; + assert_eq!(progress.completed(), 0); assert_eq!(record.accepted.load(Ordering::SeqCst), 1); db.execute_batch("DROP TRIGGER fail_head").unwrap(); - wait_for(|| *consumer.progress.borrow() == 1).await; + wait_for(|| progress.completed() == 1).await; assert_eq!(record.submitted.load(Ordering::SeqCst), 1); assert_eq!(record.accepted.load(Ordering::SeqCst), 1); consumer.shutdown().await; @@ -659,25 +633,44 @@ mod tests { }; let key = serde_json::to_string(&route).unwrap(); let record = Arc::new(Recording::default()); - let consumer = Consumer::spawn( - Box::new(journal.subscribe(&route, 0).await.unwrap()), + let (consumer, _progress, error) = spawn( + journal.subscribe(&route, 0).await.unwrap(), options(tmp.path(), &route, record.clone()), ); - wait_for(|| consumer.error.lock().is_some()).await; + wait_for(|| 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( - Box::new(journal.subscribe(&route, 0).await.unwrap()), - options(tmp.path(), &route, record.clone()), + // Restart with an ordinary channel between the durable source and the + // same consumer. Historical failure gating belongs to the source. + let opts = options(tmp.path(), &route, record.clone()); + let subscription = journal.subscribe(&route, 0).await.unwrap().into_input( + opts.config.subscribe(), + opts.version.clone(), + opts.error.clone(), + opts.recovery.clone(), ); + let progress = subscription.progress; + let mut input = subscription.input; + let (tx, rx) = tokio::sync::mpsc::channel(1); + let forward = tokio::spawn(async move { + while let Ok(Some(delivery)) = input.receive().await { + if tx.send(delivery).await.is_err() { + break; + } + } + }); + let consumer = Consumer::spawn(Box::new(rx), opts); 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; + wait_for(|| progress.completed() == 1).await; assert_eq!(record.accepted.load(Ordering::SeqCst), 1); assert!(journal.failure(&key).await.unwrap().is_none()); consumer.shutdown().await; + forward.abort(); + let _ = forward.await; + journal.close().await.unwrap(); } struct RefreshAuth(Arc); #[async_trait::async_trait] @@ -716,8 +709,8 @@ mod tests { 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(Box::new(journal.subscribe(&route, 0).await.unwrap()), opts); - wait_for(|| *consumer.progress.borrow() == 1).await; + let (consumer, progress, _error) = spawn(journal.subscribe(&route, 0).await.unwrap(), opts); + wait_for(|| progress.completed() == 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); @@ -754,15 +747,17 @@ mod tests { ..Default::default() }; let mut opts = options(tmp.path(), &route, Arc::new(Recording::default())); - opts.config.auth.api_url = Some(server.uri()); - opts.config.auth.app_url = Some(server.uri()); + opts.config.send_modify(|config| { + config.auth.api_url = Some(server.uri()); + config.auth.app_url = Some(server.uri()); + }); opts.factory = Arc::new(crate::BraintrustSinkFactory::new( crate::BraintrustSinkConfig::default(), )); - let mut config = opts.config.clone(); - let consumer = Consumer::spawn(Box::new(journal.subscribe(&route, 0).await.unwrap()), opts); - wait_for(|| consumer.error.lock().is_some()).await; - assert_eq!(*consumer.progress.borrow(), 0); + let mut config = opts.config.borrow().clone(); + let (consumer, progress, error) = spawn(journal.subscribe(&route, 0).await.unwrap(), opts); + wait_for(|| error.lock().unwrap().is_some()).await; + assert_eq!(progress.completed(), 0); drop(rejection); Mock::given(method("POST")) .and(path("/logs3")) @@ -774,11 +769,11 @@ mod tests { config.auth.token = "replacement-token".into(); consumer.configure(config.clone()); tokio::time::sleep(std::time::Duration::from_millis(500)).await; - assert_eq!(*consumer.progress.borrow(), 0); + assert_eq!(progress.completed(), 0); config.tags.push("corrected".into()); consumer.configure(config); - wait_for(|| *consumer.progress.borrow() == 1).await; - assert!(consumer.error.lock().is_none()); + wait_for(|| progress.completed() == 1).await; + assert!(error.lock().unwrap().is_none()); consumer.shutdown().await; } @@ -799,11 +794,11 @@ mod tests { ..Default::default() }; let record = Arc::new(Recording::default()); - let consumer = Consumer::spawn( - Box::new(journal.subscribe(&route, 0).await.unwrap()), + let (consumer, progress, error) = spawn( + journal.subscribe(&route, 0).await.unwrap(), options(tmp.path(), &route, record.clone()), ); - wait_for(|| consumer.error.lock().is_some()).await; + wait_for(|| error.lock().unwrap().is_some()).await; let failure: Failure = serde_json::from_str( &journal .failure(&serde_json::to_string(&route).unwrap()) @@ -828,7 +823,7 @@ mod tests { RecoveryOutcome::Waiting )); std::fs::write(&first, "export default span => ({...span,name:'good'})").unwrap(); - wait_for(|| *consumer.progress.borrow() == 1).await; + wait_for(|| progress.completed() == 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/delivery.rs b/bt-daemon/src/pipeline/delivery.rs index 7c636009..3b9536a6 100644 --- a/bt-daemon/src/pipeline/delivery.rs +++ b/bt-daemon/src/pipeline/delivery.rs @@ -1,6 +1,9 @@ +use super::recovery::Failure; use tokio::sync::mpsc; /// A message owns its completion obligation until terminal handling succeeds. +/// Mapping or forwarding preserves it; dropping it never acknowledges the input. +/// Recoverable sources may redeliver abandoned input. pub(crate) struct Delivery { value: T, completion: Box, @@ -11,6 +14,12 @@ pub(crate) struct Delivery { #[async_trait::async_trait] pub(crate) trait Completion: Send { async fn complete(&mut self) -> anyhow::Result<()>; + + /// Report a processing failure through the same obligation. Durable sources + /// retain the evidence; ordinary channels may simply discard the report. + async fn failed(&mut self, _failure: &mut Failure) -> anyhow::Result<()> { + Ok(()) + } } pub(crate) struct Unfinished { @@ -30,7 +39,6 @@ impl Delivery { &self.value } - #[cfg_attr(not(test), allow(dead_code))] pub fn map(self, map: impl FnOnce(T) -> U) -> Delivery { Delivery { value: map(self.value), @@ -38,6 +46,10 @@ impl Delivery { } } + pub async fn failed(&mut self, failure: &mut Failure) -> anyhow::Result<()> { + self.completion.failed(failure).await + } + pub async fn into_completed(mut self) -> Result> { match self.completion.complete().await { Ok(()) => Ok(self.value), diff --git a/bt-daemon/src/pipeline/journal.rs b/bt-daemon/src/pipeline/journal.rs index d0409c70..6fd58a06 100644 --- a/bt-daemon/src/pipeline/journal.rs +++ b/bt-daemon/src/pipeline/journal.rs @@ -1,14 +1,21 @@ //! Concrete durable channel: one ordered SpanOp log and a head per route. -use super::{recovery::Failure, Completion, Delivery}; +use super::{ + delivery::InputPort, + recovery::{Boundary, Failure, Recovery, RecoveryOutcome}, + Completion, Delivery, Stage, +}; use crate::{ translate::SpanOp, - wire::{SessionConfig, SessionRoute}, + wire::{RecoveryInfo, SessionConfig, SessionRoute}, }; use rusqlite::{params, Connection, OptionalExtension}; use serde::{Deserialize, Serialize}; use std::{ path::Path, - sync::{Arc, Mutex}, + sync::{ + atomic::{AtomicBool, Ordering}, + Arc, Mutex, + }, }; use tokio::sync::watch; @@ -68,10 +75,8 @@ impl StoredOp { impl SpanJournal { pub async fn open(path: &Path) -> anyhow::Result { let path = path.to_owned(); - let connection = tokio::task::spawn_blocking(move || -> anyhow::Result { - let parent = path - .parent() - .ok_or_else(|| anyhow::anyhow!("SpanOp journal path has no parent"))?; + let connection = tokio::task::spawn_blocking(move || -> anyhow::Result<(Connection, u64)> { + let parent = path.parent().ok_or_else(|| anyhow::anyhow!("SpanOp journal path has no parent"))?; std::fs::create_dir_all(parent)?; let db = Connection::open(path)?; db.busy_timeout(std::time::Duration::from_secs(5))?; @@ -102,10 +107,12 @@ impl SpanJournal { db.execute_batch("ALTER TABLE heads ADD COLUMN binding TEXT")?; db.execute_batch("PRAGMA user_version=3; COMMIT")?; } - Ok(db) + let tail = db.query_row("SELECT coalesce(max(position),0) FROM ops JOIN batches USING(input) WHERE committed=1", [], |row| row.get(0))?; + Ok((db, tail)) }) .await??; - let (changed, _) = watch::channel(0); + let (connection, tail) = connection; + let (changed, _) = watch::channel(tail); Ok(Self { connection: Arc::new(Mutex::new(Some(connection))), changed, @@ -239,8 +246,7 @@ impl SpanJournal { Ok(tail) }) .await?; - self.changed - .send_modify(|version| *version = version.wrapping_add(1)); + self.changed.send_replace(tail); Ok(tail) } @@ -273,12 +279,20 @@ impl SpanJournal { Ok(position) }) .await?; + let (progress, _) = watch::channel(position); Ok(Cursor { journal: self.clone(), route, read: position, completed: position, changes: self.changed.subscribe(), + progress, + abandoned: Arc::new(AtomicBool::new(false)), + settings: None, + version: String::new(), + restored: false, + error: Arc::new(Mutex::new(None)), + recovery: Arc::new(Mutex::new(None)), }) } @@ -426,69 +440,83 @@ pub(crate) struct Cursor { read: u64, completed: u64, changes: watch::Receiver, + progress: watch::Sender, + abandoned: Arc, + settings: Option>, + version: String, + restored: bool, + error: Arc>>, + recovery: Arc>>, } -/// The route-facing view of the durable SpanOp channel. -/// -/// Consumers depend on delivery and head operations, while `SpanJournal` and -/// SQLite remain private to this adapter. -#[async_trait::async_trait] -#[allow(clippy::double_must_use)] // async_trait marks its boxed future returns as must_use. -pub(crate) trait RouteInputPort: - super::delivery::InputPort> + Sync -{ - fn route_key(&self) -> &str; - fn position(&self) -> u64; - fn completed_position(&self) -> u64; - fn pending_prefix(&self) -> Delivery<()>; - async fn rewind(&mut self) -> anyhow::Result<()>; - async fn tail(&self) -> anyhow::Result; - async fn recovery(&self, config: &SessionConfig) -> anyhow::Result>; - async fn set_recovery(&self, failure: Option<&Failure>) -> anyhow::Result<()>; - async fn bind(&self, config: &SessionConfig) -> anyhow::Result<()>; +/// The composition layer keeps this status handle; downstream stages only get +/// the InputPort and managed deliveries. +#[derive(Clone)] +pub(crate) struct Progress { + completed: watch::Receiver, + published: watch::Receiver, } - -#[derive(Debug)] -pub(crate) struct RouteBindingChanged; -impl std::fmt::Display for RouteBindingChanged { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.write_str("durable route binding changed") +#[cfg(test)] +impl Default for Progress { + fn default() -> Self { + Self { + completed: watch::channel(0).1, + published: watch::channel(0).1, + } } } -impl std::error::Error for RouteBindingChanged {} - -#[async_trait::async_trait] -impl RouteInputPort for Box { - fn route_key(&self) -> &str { - (**self).route_key() +impl Progress { + pub fn completed(&self) -> u64 { + *self.completed.borrow() } - fn position(&self) -> u64 { - (**self).position() + pub fn pending(&self) -> u64 { + self.published.borrow().saturating_sub(self.completed()) } - fn completed_position(&self) -> u64 { - (**self).completed_position() - } - fn pending_prefix(&self) -> Delivery<()> { - (**self).pending_prefix() - } - async fn rewind(&mut self) -> anyhow::Result<()> { - (**self).rewind().await - } - async fn tail(&self) -> anyhow::Result { - (**self).tail().await - } - async fn recovery(&self, config: &SessionConfig) -> anyhow::Result> { - (**self).recovery(config).await - } - async fn set_recovery(&self, failure: Option<&Failure>) -> anyhow::Result<()> { - (**self).set_recovery(failure).await - } - async fn bind(&self, config: &SessionConfig) -> anyhow::Result<()> { - (**self).bind(config).await + + pub async fn through(&self, target: u64, error: &Mutex>) -> Result<(), String> { + let mut completed = self.completed.clone(); + loop { + if *completed.borrow_and_update() >= target { + return Ok(()); + } + if let Some(error) = error.lock().unwrap().clone() { + return Err(error); + } + tokio::select! { + changed = completed.changed() => changed.map_err(|_| "journal input is gone".to_owned())?, + _ = tokio::time::sleep(std::time::Duration::from_millis(250)) => {}, + } + } } } +pub(crate) struct Subscription { + pub input: Box>>, + pub progress: Progress, +} + impl Cursor { + pub fn into_input( + mut self, + settings: watch::Receiver, + version: String, + error: Arc>>, + recovery: Arc>>, + ) -> Subscription { + let progress = Progress { + completed: self.progress.subscribe(), + published: self.changes.clone(), + }; + self.settings = Some(settings); + self.version = version; + self.error = error; + self.recovery = recovery; + Subscription { + input: Box::new(self), + progress, + } + } + pub async fn next_batch( &mut self, limit: usize, @@ -523,28 +551,19 @@ impl Cursor { route: self.route.clone(), expected: after, through, + progress: self.progress.clone(), + abandoned: self.abandoned.clone(), + acknowledged: false, + error: self.error.clone(), + recovery: self.recovery.clone(), }, ))) } - pub fn position(&self) -> u64 { - self.read - } + #[cfg(test)] pub fn completed_position(&self) -> u64 { self.completed } - /// Terminal handling can aggregate its fully handled contiguous read prefix. - pub fn pending_prefix(&self) -> Delivery<()> { - Delivery::new( - (), - HeadCompletion { - journal: self.journal.clone(), - route: self.route.clone(), - expected: self.completed, - through: self.read, - }, - ) - } pub async fn rewind(&mut self) -> anyhow::Result<()> { let key = self.route.clone(); @@ -559,73 +578,116 @@ impl Cursor { }) .await?; self.read = self.completed; + self.progress.send_replace(self.completed); Ok(()) } + #[cfg(test)] pub async fn changed(&mut self) { let _ = self.changes.changed().await; } } #[async_trait::async_trait] -impl super::delivery::InputPort for Cursor { +impl InputPort for Cursor { type Item = Vec; async fn receive(&mut self) -> anyhow::Result>> { - loop { - if let Some(delivery) = self.next_batch(128).await? { - return Ok(Some(delivery)); + let result = self.receive_managed().await; + match result { + Ok(delivery) => Ok(Some(delivery)), + Err(error) => { + let error = match error.downcast::() { + Ok(failure) => return Err(anyhow::Error::new(failure)), + Err(error) => error, + }; + let failure = Failure::storage(&error, &self.version, 0).at(self.completed + 1); + // An input failure has no delivery receipt to report through. + // Its source still owns persistence of the recovery evidence. + if let Err(error) = self + .journal + .set_failure(&self.route, Some(serde_json::to_string(&failure)?)) + .await + { + tracing::warn!(%error, "input recovery evidence could not be persisted"); + } + Err(anyhow::Error::new(failure)) } - self.changed().await; } } } -#[async_trait::async_trait] -impl RouteInputPort for Cursor { - fn route_key(&self) -> &str { - &self.route - } - fn position(&self) -> u64 { - Cursor::position(self) - } - fn completed_position(&self) -> u64 { - Cursor::completed_position(self) - } - fn pending_prefix(&self) -> Delivery<()> { - Cursor::pending_prefix(self) - } - async fn rewind(&mut self) -> anyhow::Result<()> { - Cursor::rewind(self).await - } - async fn tail(&self) -> anyhow::Result { - self.journal.tail().await - } - async fn recovery(&self, config: &SessionConfig) -> anyhow::Result> { - let Some(raw) = self.journal.failure(&self.route).await? else { - return Ok(None); - }; - Ok(Some(serde_json::from_str(&raw).unwrap_or_else(|_| { - Failure::legacy(raw, self.completed_position() + 1, config) - }))) - } - async fn set_recovery(&self, failure: Option<&Failure>) -> anyhow::Result<()> { - let failure = failure - .map(serde_json::to_string) - .transpose() - .map_err(anyhow::Error::from)?; - self.journal.set_failure(&self.route, failure).await - } - async fn bind(&self, config: &SessionConfig) -> anyhow::Result<()> { - self.journal - .bind(&self.route, config) - .await - .map_err(|error| { - if error.downcast_ref::().is_some() { - anyhow::Error::new(RouteBindingChanged) - } else { - error +impl Cursor { + async fn receive_managed(&mut self) -> anyhow::Result>> { + let mut acknowledgements = self.progress.subscribe(); + loop { + if self.abandoned.load(Ordering::Acquire) { + self.rewind().await?; + self.abandoned.store(false, Ordering::Release); + } + self.completed = *self.progress.borrow(); + self.read = self.read.max(self.completed); + let config = self + .settings + .as_ref() + .map(|settings| settings.borrow().clone()); + if !self.restored { + if let Some(raw) = self.journal.failure(&self.route).await? { + let failure = match serde_json::from_str::(&raw) { + Ok(failure) => failure, + Err(error) => match &config { + Some(config) => Failure::legacy(raw, self.completed + 1, config), + None => return Err(error.into()), + }, + }; + if matches!(failure.boundary, Boundary::Completion) + && self.completed >= failure.through.unwrap_or(failure.position) + { + self.journal.set_failure(&self.route, None).await?; + } else { + *self.error.lock().unwrap() = Some(failure.message()); + *self.recovery.lock().unwrap() = Some(failure.info(false)); + if matches!( + (Recovery { + config: config.as_ref(), + version: &self.version + }) + .process(&failure) + .await, + RecoveryOutcome::Waiting + ) { + tokio::time::sleep(std::time::Duration::from_millis(250)).await; + continue; + } + *self.recovery.lock().unwrap() = Some(failure.info(true)); + } } - }) + } + if let Some(config) = &config { + if let Err(error) = self.journal.bind(&self.route, config).await { + if error.downcast_ref::().is_some() { + let failure = + Failure::legacy(error.to_string(), self.completed + 1, config); + self.journal + .set_failure(&self.route, Some(serde_json::to_string(&failure)?)) + .await?; + return Err(anyhow::Error::new(failure)); + } + return Err(error); + } + } + if let Some(delivery) = self.next_batch(128).await? { + if self.abandoned.load(Ordering::Acquire) { + drop(delivery); + continue; + } + self.restored = true; + return Ok(delivery); + } + tokio::select! { + _ = self.changes.changed() => {}, + _ = acknowledgements.changed() => {}, + } + } } } @@ -634,6 +696,19 @@ struct HeadCompletion { route: String, expected: u64, through: u64, + progress: watch::Sender, + abandoned: Arc, + acknowledged: bool, + error: Arc>>, + recovery: Arc>>, +} +impl Drop for HeadCompletion { + fn drop(&mut self) { + if !self.acknowledged { + self.abandoned.store(true, Ordering::Release); + self.progress.send_modify(|_| {}); + } + } } #[async_trait::async_trait] impl Completion for HeadCompletion { @@ -647,7 +722,7 @@ impl Completion for HeadCompletion { r.get(0) })?; if head >= through { - return Ok(()); + return Ok(head); } anyhow::ensure!(head >= expected, "completion cannot skip unresolved input"); tx.execute( @@ -655,9 +730,25 @@ impl Completion for HeadCompletion { params![through, route], )?; tx.commit()?; - Ok(()) + Ok(through) }) .await + .map(|head| { + self.acknowledged = true; + self.progress.send_replace(head); + *self.error.lock().unwrap() = None; + *self.recovery.lock().unwrap() = None; + }) + } + + async fn failed(&mut self, failure: &mut Failure) -> anyhow::Result<()> { + failure.position = self.expected + 1; + failure.through = Some(self.through); + *self.error.lock().unwrap() = Some(failure.message()); + *self.recovery.lock().unwrap() = Some(failure.info(false)); + self.journal + .set_failure(&self.route, Some(serde_json::to_string(failure)?)) + .await } } @@ -688,6 +779,40 @@ mod tests { } } + #[tokio::test] + async fn input_port_replays_abandoned_delivery_without_cursor_control() { + use crate::pipeline::delivery::InputPort; + let tmp = tempfile::tempdir().unwrap(); + let journal = SpanJournal::open(&tmp.path().join("spans.sqlite")) + .await + .unwrap(); + journal.begin(1, true).await.unwrap(); + journal.append(1, &[op("one")]).await.unwrap(); + journal.commit(1).await.unwrap(); + let mut input: Box>> = + Box::new(journal.subscribe(&route("a"), 0).await.unwrap()); + let delivery = input.receive().await.unwrap().unwrap(); + drop(delivery); + // Cancel a receive while its storage read is pending, as a concurrent + // configuration update does. The replay obligation must survive it. + let storage = journal.connection.lock().unwrap(); + let mut cancelled = Box::pin(input.receive()); + let mut context = std::task::Context::from_waker(std::task::Waker::noop()); + assert!(std::future::Future::poll(cancelled.as_mut(), &mut context).is_pending()); + drop(cancelled); + drop(storage); + let replay = tokio::time::timeout(std::time::Duration::from_millis(250), input.receive()) + .await + .expect("unacknowledged input must be replayed") + .unwrap() + .unwrap(); + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = &replay.value()[0]; + assert_eq!(row.name, "one"); + replay.into_completed().await.ok().unwrap(); + assert!(journal.route_complete(&route("a"), 1).await.unwrap()); + assert!(!journal.route_complete(&route("b"), 1).await.unwrap()); + } + #[tokio::test] async fn preparatory_schema_upgrades_without_changing_committed_positions() { let tmp = tempfile::tempdir().unwrap(); diff --git a/bt-daemon/src/pipeline/recovery.rs b/bt-daemon/src/pipeline/recovery.rs index 97b82412..71a029ac 100644 --- a/bt-daemon/src/pipeline/recovery.rs +++ b/bt-daemon/src/pipeline/recovery.rs @@ -98,6 +98,12 @@ fn configuration(config: &SessionConfig) -> String { } impl Failure { + /// Source adapters attach their own diagnostic location to stage failures. + pub fn at(mut self, position: u64) -> Self { + self.position = position; + self + } + 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 { @@ -123,16 +129,11 @@ impl Failure { }, } } - pub fn routing( - error: &anyhow::Error, - position: u64, - config: &SessionConfig, - attempt: u32, - ) -> Self { + pub fn routing(error: &anyhow::Error, 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); + let mut failure = Self::delay(Boundary::Sink, 0, format!("{error:#}"), attempt); if let Some(plugin) = error.downcast_ref::() { failure.boundary = Boundary::Plugin; failure.condition = Condition::Plugin { @@ -178,8 +179,8 @@ impl Failure { } failure } - pub fn credentials(position: u64, reason: String, attempt: u32) -> Self { - let mut failure = Self::delay(Boundary::Credentials, position, reason, attempt); + pub fn credentials(reason: String, attempt: u32) -> Self { + let mut failure = Self::delay(Boundary::Credentials, 0, reason, attempt); if let Condition::Delay { recheck_ms } = failure.condition { failure.condition = Condition::Credentials { recheck_ms }; } @@ -198,12 +199,12 @@ impl Failure { }, } } - pub fn storage(error: &anyhow::Error, position: u64, version: &str, attempt: u32) -> Self { + pub fn storage(error: &anyhow::Error, 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) } + Self::revision(Boundary::Journal,0,error.to_string(),version) + } else { Self::delay(Boundary::Journal,0,error.to_string(),attempt) } } pub fn info(&self, eligible: bool) -> RecoveryInfo { let retry = match &self.condition { @@ -238,6 +239,13 @@ impl Failure { } } +impl std::fmt::Display for Failure { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(&self.message()) + } +} +impl std::error::Error for Failure {} + pub(crate) enum RecoveryOutcome { Waiting, Retry, diff --git a/bt-daemon/src/pipeline/translation.rs b/bt-daemon/src/pipeline/translation.rs index 74bff4fe..a60437a0 100644 --- a/bt-daemon/src/pipeline/translation.rs +++ b/bt-daemon/src/pipeline/translation.rs @@ -176,10 +176,8 @@ impl Translation { &self, route: &SessionRoute, legacy_through: u64, - ) -> anyhow::Result> { - Ok(Box::new( - self.journal.subscribe(route, legacy_through).await?, - )) + ) -> anyhow::Result { + self.journal.subscribe(route, legacy_through).await } } @@ -368,7 +366,7 @@ impl Producer { &self.options.version, ) } else { - Failure::storage(&error, position, &self.options.version, attempt) + Failure::storage(&error, &self.options.version, attempt).at(position) }; failure.through = self .publication diff --git a/bt-daemon/src/server/correlation_store.rs b/bt-daemon/src/server/correlation_store.rs index c6520eef..4607e142 100644 --- a/bt-daemon/src/server/correlation_store.rs +++ b/bt-daemon/src/server/correlation_store.rs @@ -354,11 +354,9 @@ pub(super) async fn finish_correlation( .events .first() .map_or(0, |event| event.journal_through); - state.recovery = Some(crate::pipeline::recovery::Failure::credentials( - position, - error.clone(), - attempt, - )); + state.recovery = Some( + crate::pipeline::recovery::Failure::credentials(error.clone(), attempt).at(position), + ); if let Err(error) = store_correlation(daemon, key, &state).await { tracing::warn!(%error,"pending recovery evidence could not be persisted"); } diff --git a/bt-daemon/src/server/sessions.rs b/bt-daemon/src/server/sessions.rs index 763d4aa5..b722fd21 100644 --- a/bt-daemon/src/server/sessions.rs +++ b/bt-daemon/src/server/sessions.rs @@ -128,6 +128,7 @@ impl Daemon { SessionOptions { translation, input, + route: route.clone(), auth_provider: self.route_auth_provider(key), session_id: env.session_id.clone(), source: env.source.clone(), From 618f17829e00261f95ba547d50bd7e805d810332 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Sat, 17 Oct 2026 00:51:04 +0800 Subject: [PATCH 4/4] Preserve typed recovery and completion through input ports --- bt-daemon/src/dispatch.rs | 8 +-- bt-daemon/src/pipeline/consumer.rs | 88 ++++++++++++++++++++++-------- bt-daemon/src/pipeline/journal.rs | 25 +++++---- bt-daemon/src/pipeline/recovery.rs | 6 +- bt-daemon/src/server/sessions.rs | 1 + 5 files changed, 83 insertions(+), 45 deletions(-) diff --git a/bt-daemon/src/dispatch.rs b/bt-daemon/src/dispatch.rs index 23794a2f..47baecb1 100644 --- a/bt-daemon/src/dispatch.rs +++ b/bt-daemon/src/dispatch.rs @@ -39,6 +39,7 @@ enum SessionMsg { /// Where to rebuild a session's translator state from, streamed at startup. pub(crate) struct SessionOptions { + pub observer_key: String, pub session_id: String, pub source: String, pub bt_version: String, @@ -81,12 +82,7 @@ impl Session { last_error.clone(), recovery.clone(), ); - let observer_key = format!( - "{}\u{1f}{}\u{1f}{}", - options.source, - options.session_id, - serde_json::to_string(&options.route).expect("serializable route") - ); + let observer_key = options.observer_key; let route = Consumer::spawn( subscription.input, ConsumerOptions { diff --git a/bt-daemon/src/pipeline/consumer.rs b/bt-daemon/src/pipeline/consumer.rs index 6be625e2..fec51b26 100644 --- a/bt-daemon/src/pipeline/consumer.rs +++ b/bt-daemon/src/pipeline/consumer.rs @@ -13,10 +13,11 @@ use crate::{ wire::{RecoveryInfo, SessionConfig, SessionRoute}, AuthProvider, AuthResolveReason, }; +use parking_lot::Mutex; use std::{ collections::VecDeque, path::PathBuf, - sync::{atomic::Ordering, Arc, Mutex}, + sync::{atomic::Ordering, Arc}, }; use tokio::sync::{watch, Notify}; @@ -74,8 +75,8 @@ impl ConsumerOptions { Some(delivery) => delivery.failed(failure).await, None => Ok(()), // Input errors are reported by the source itself. }; - *self.error.lock().unwrap() = Some(failure.message()); - *self.recovery.lock().unwrap() = Some(failure.info(false)); + *self.error.lock() = Some(failure.message()); + *self.recovery.lock() = Some(failure.info(false)); if let Err(error) = result { tracing::warn!(%error,"delivery recovery evidence could not be reported; original failure retained"); return false; @@ -152,7 +153,13 @@ impl ConsumerOptions { return Ok(false); } 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"), + attempt, + Work::Flush, + ) + })?; sink.flush().await.map_err(|error| { if error.downcast_ref::().is_some() { self.storage(&error, attempt, Work::Flush) @@ -160,7 +167,7 @@ impl ConsumerOptions { self.routing(&error, config, attempt) } })?; - *self.permalink.lock().unwrap() = sink.permalink(); + *self.permalink.lock() = sink.permalink(); if sink.has_pending_delivery() { return Ok(false); } @@ -200,13 +207,13 @@ impl Consumer { loop { if let Some(waiting) = held.take() { let (mut failure, work) = *waiting; - let mut encoded = serde_json::to_string(&failure).expect("failure evidence"); - if recorded.as_ref() != Some(&encoded) + let mut reported = failure.clone(); + if recorded.as_ref() != Some(&reported) && super::recovery::now_ms() >= record_retry { if options.hold(&mut pending, &mut failure).await { - encoded = serde_json::to_string(&failure).expect("failure evidence"); - recorded = Some(encoded.clone()); + reported = failure.clone(); + recorded = Some(reported.clone()); } else { record_retry = super::recovery::now_ms().saturating_add(1_000); } @@ -217,7 +224,7 @@ impl Consumer { result=settings.changed()=>if result.is_err() { break; }, } let mut current = settings.borrow_and_update().clone(); - if recorded.as_ref() != Some(&encoded) + if recorded.as_ref() != Some(&reported) || matches!( (Recovery { config: Some(¤t), @@ -231,7 +238,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(&mut current).await { @@ -249,8 +256,8 @@ impl Consumer { { Ok(completed) => { if completed { - *options.error.lock().unwrap() = None; - *options.recovery.lock().unwrap() = None; + *options.error.lock() = None; + *options.recovery.lock() = None; recorded = None; } } @@ -289,7 +296,13 @@ impl Consumer { } })?); } - let routing = sink.as_mut().expect("configured sink"); + let routing = sink.as_mut().ok_or_else(|| { + options.storage( + &anyhow::anyhow!("configured sink is missing"), + attempts, + Work::Resubmit, + ) + })?; routing.configure(¤t); routing .emit(ops) @@ -316,8 +329,8 @@ impl Consumer { Ok(None) => break, Ok(Some(false)) => {} Ok(Some(true)) => { - *options.error.lock().unwrap() = None; - *options.recovery.lock().unwrap() = None; + *options.error.lock() = None; + *options.recovery.lock() = None; attempts = 0; recorded = None; } @@ -325,7 +338,7 @@ impl Consumer { } } }); - *handle.task.lock().unwrap() = Some(task); + *handle.task.lock() = Some(task); handle } pub fn configure(&self, config: SessionConfig) { @@ -339,7 +352,7 @@ impl Consumer { } pub async fn shutdown(&self) { self.stop.notify_one(); - let task = self.task.lock().unwrap().take(); + let task = self.task.lock().take(); if let Some(task) = task { let _ = task.await; } @@ -362,6 +375,7 @@ mod tests { accepted: AtomicU64, reject: AtomicBool, unauthorized: AtomicBool, + commit_fail: AtomicBool, defer_until: AtomicU64, } struct Factory(Arc); @@ -404,6 +418,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(()) } fn has_pending_delivery(&self) -> bool { @@ -517,7 +534,7 @@ mod tests { wait_for(|| record.accepted.load(Ordering::SeqCst) == 1).await; wait_for(|| progress.completed() == 1).await; assert_eq!(record.submitted.load(Ordering::SeqCst), 2); - assert!(error.lock().unwrap().is_none()); + assert!(error.lock().is_none()); assert!(journal .failure(&serde_json::to_string(&route).unwrap()) .await @@ -596,6 +613,29 @@ mod tests { 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, progress, error) = spawn( + journal.subscribe(&route, 0).await.unwrap(), + options(tmp.path(), &route, record.clone()), + ); + wait_for(|| error.lock().is_some()).await; + assert_eq!(progress.completed(), 0); + assert_eq!(record.accepted.load(Ordering::SeqCst), 1); + wait_for(|| progress.completed() == 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(); @@ -609,7 +649,7 @@ mod tests { let record = Arc::new(Recording::default()); let (consumer, progress, error) = spawn(cursor, options(tmp.path(), &route, record.clone())); - wait_for(|| error.lock().unwrap().is_some()).await; + wait_for(|| error.lock().is_some()).await; assert_eq!(progress.completed(), 0); assert_eq!(record.accepted.load(Ordering::SeqCst), 1); db.execute_batch("DROP TRIGGER fail_head").unwrap(); @@ -637,7 +677,7 @@ mod tests { journal.subscribe(&route, 0).await.unwrap(), options(tmp.path(), &route, record.clone()), ); - wait_for(|| error.lock().unwrap().is_some()).await; + wait_for(|| error.lock().is_some()).await; consumer.shutdown().await; let evidence = journal.failure(&key).await.unwrap().unwrap(); assert!(!evidence.contains("never-persist-this")); @@ -756,7 +796,7 @@ mod tests { )); let mut config = opts.config.borrow().clone(); let (consumer, progress, error) = spawn(journal.subscribe(&route, 0).await.unwrap(), opts); - wait_for(|| error.lock().unwrap().is_some()).await; + wait_for(|| error.lock().is_some()).await; assert_eq!(progress.completed(), 0); drop(rejection); Mock::given(method("POST")) @@ -773,7 +813,7 @@ mod tests { config.tags.push("corrected".into()); consumer.configure(config); wait_for(|| progress.completed() == 1).await; - assert!(error.lock().unwrap().is_none()); + assert!(error.lock().is_none()); consumer.shutdown().await; } @@ -798,7 +838,7 @@ mod tests { journal.subscribe(&route, 0).await.unwrap(), options(tmp.path(), &route, record.clone()), ); - wait_for(|| error.lock().unwrap().is_some()).await; + wait_for(|| 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/journal.rs b/bt-daemon/src/pipeline/journal.rs index 6fd58a06..3c336c08 100644 --- a/bt-daemon/src/pipeline/journal.rs +++ b/bt-daemon/src/pipeline/journal.rs @@ -8,13 +8,14 @@ use crate::{ translate::SpanOp, wire::{RecoveryInfo, SessionConfig, SessionRoute}, }; +use parking_lot::Mutex; use rusqlite::{params, Connection, OptionalExtension}; use serde::{Deserialize, Serialize}; use std::{ path::Path, sync::{ atomic::{AtomicBool, Ordering}, - Arc, Mutex, + Arc, Mutex as StorageMutex, }, }; use tokio::sync::watch; @@ -32,7 +33,7 @@ impl std::error::Error for BindingChanged {} #[derive(Clone)] pub(crate) struct SpanJournal { - connection: Arc>>, + connection: Arc>>, changed: watch::Sender, } @@ -114,7 +115,7 @@ impl SpanJournal { let (connection, tail) = connection; let (changed, _) = watch::channel(tail); Ok(Self { - connection: Arc::new(Mutex::new(Some(connection))), + connection: Arc::new(StorageMutex::new(Some(connection))), changed, }) } @@ -479,7 +480,7 @@ impl Progress { if *completed.borrow_and_update() >= target { return Ok(()); } - if let Some(error) = error.lock().unwrap().clone() { + if let Some(error) = error.lock().clone() { return Err(error); } tokio::select! { @@ -644,8 +645,8 @@ impl Cursor { { self.journal.set_failure(&self.route, None).await?; } else { - *self.error.lock().unwrap() = Some(failure.message()); - *self.recovery.lock().unwrap() = Some(failure.info(false)); + *self.error.lock() = Some(failure.message()); + *self.recovery.lock() = Some(failure.info(false)); if matches!( (Recovery { config: config.as_ref(), @@ -658,7 +659,7 @@ impl Cursor { tokio::time::sleep(std::time::Duration::from_millis(250)).await; continue; } - *self.recovery.lock().unwrap() = Some(failure.info(true)); + *self.recovery.lock() = Some(failure.info(true)); } } } @@ -736,16 +737,16 @@ impl Completion for HeadCompletion { .map(|head| { self.acknowledged = true; self.progress.send_replace(head); - *self.error.lock().unwrap() = None; - *self.recovery.lock().unwrap() = None; + *self.error.lock() = None; + *self.recovery.lock() = None; }) } async fn failed(&mut self, failure: &mut Failure) -> anyhow::Result<()> { failure.position = self.expected + 1; failure.through = Some(self.through); - *self.error.lock().unwrap() = Some(failure.message()); - *self.recovery.lock().unwrap() = Some(failure.info(false)); + *self.error.lock() = Some(failure.message()); + *self.recovery.lock() = Some(failure.info(false)); self.journal .set_failure(&self.route, Some(serde_json::to_string(failure)?)) .await @@ -795,7 +796,7 @@ mod tests { drop(delivery); // Cancel a receive while its storage read is pending, as a concurrent // configuration update does. The replay obligation must survive it. - let storage = journal.connection.lock().unwrap(); + let storage = journal.connection.lock(); let mut cancelled = Box::pin(input.receive()); let mut context = std::task::Context::from_waker(std::task::Waker::noop()); assert!(std::future::Future::poll(cancelled.as_mut(), &mut context).is_pending()); diff --git a/bt-daemon/src/pipeline/recovery.rs b/bt-daemon/src/pipeline/recovery.rs index 71a029ac..8030bf10 100644 --- a/bt-daemon/src/pipeline/recovery.rs +++ b/bt-daemon/src/pipeline/recovery.rs @@ -5,7 +5,7 @@ use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; use std::path::PathBuf; -#[derive(Debug, Clone, Copy, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub(crate) enum Boundary { Plugin, @@ -30,7 +30,7 @@ impl std::fmt::Display for Boundary { } } -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(tag = "when", rename_all = "snake_case")] pub(crate) enum Condition { Plugin { @@ -55,7 +55,7 @@ pub(crate) enum Condition { }, } -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub(crate) struct Failure { pub boundary: Boundary, pub position: u64, diff --git a/bt-daemon/src/server/sessions.rs b/bt-daemon/src/server/sessions.rs index b722fd21..d3b86e02 100644 --- a/bt-daemon/src/server/sessions.rs +++ b/bt-daemon/src/server/sessions.rs @@ -126,6 +126,7 @@ impl Daemon { } let session = Session::spawn( SessionOptions { + observer_key: key.correlation_key(), translation, input, route: route.clone(),