diff --git a/bt-daemon/src/dispatch.rs b/bt-daemon/src/dispatch.rs index 04ffb939..47baecb1 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, } @@ -40,13 +39,15 @@ 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, pub config: crate::wire::SessionConfig, + pub route: crate::wire::SessionRoute, pub data_dir: PathBuf, pub translation: Arc, - pub cursor: crate::pipeline::journal::Cursor, + pub input: crate::pipeline::journal::Cursor, pub auth_provider: Option>, } @@ -57,6 +58,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,19 +75,23 @@ impl Session { let last_error = Arc::new(Mutex::new(None)); let permalink = Arc::new(Mutex::new(None)); let source = options.source.clone(); - let observer_key = format!( - "{}\u{1f}{}\u{1f}{}", - options.source, options.session_id, options.cursor.route + 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 = options.observer_key; let route = Consumer::spawn( - options.translation.journal.clone(), - options.cursor, + 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(), @@ -100,7 +106,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(_)); @@ -128,24 +135,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) } } }; @@ -165,7 +171,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; } @@ -188,6 +194,7 @@ impl Session { last_error, source_error, source_progress, + route_progress, recovery, source_recovery, permalink, @@ -196,9 +203,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()) } @@ -245,11 +251,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()), } } @@ -259,11 +265,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()), } } @@ -589,6 +595,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 850ff6eb..fec51b26 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::{Cursor, SpanJournal}, recovery::{Boundary, Failure, Recovery, RecoveryOutcome}, routing::RoutingSink, Delivery, Stage, @@ -10,11 +9,13 @@ 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}, }; @@ -23,23 +24,22 @@ 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,27 +70,25 @@ 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, 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() = 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"); + 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, @@ -114,28 +112,17 @@ impl ConsumerOptions { 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, - ); + 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, - cursor: &Cursor, - 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 @@ -144,149 +131,100 @@ impl ConsumerOptions { tracing::warn!(session_id=%self.session_id, %error, "route processing failed"); } } - Box::new(( - Failure::routing(error, cursor.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, - cursor: &mut Cursor, + 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, ) })?; sink.flush().await.map_err(|error| { if error.downcast_ref::().is_some() { - self.storage(&error, cursor, attempt, Work::Flush) + self.storage(&error, attempt, Work::Flush) } else { - self.routing(&error, cursor, 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. 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), - ) - })?; + Work::Complete => {} } - cursor - .rewind() - .await - .map_err(|error| self.storage(&error, cursor, 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(journal: SpanJournal, mut cursor: Cursor, options: ConsumerOptions) -> Arc { - let (config, mut settings) = watch::channel(options.config.clone()); - let (progress, receive) = watch::channel(cursor.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 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) => None, - Err(error) => Some(options.storage(&error, &cursor, 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) - && cursor.completed_position() - >= failure.through.unwrap_or(failure.position) - && journal.set_failure(&cursor.route, None).await.is_ok() - { - *options.error.lock() = None; - *options.recovery.lock() = None; - continue; - } - failure.through.get_or_insert(cursor.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 recorded.as_ref() != Some(&encoded) + if let Some(waiting) = held.take() { + let (mut failure, work) = *waiting; + let mut reported = failure.clone(); + if recorded.as_ref() != Some(&reported) && super::recovery::now_ms() >= record_retry { - if options.hold(&journal, &cursor, &mut failure).await { - recorded = Some(encoded.clone()); + if options.hold(&mut pending, &mut failure).await { + reported = failure.clone(); + recorded = Some(reported.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) + if recorded.as_ref() != Some(&reported) || matches!( (Recovery { config: Some(¤t), @@ -303,13 +241,9 @@ 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(&mut current).await { held = Some(Box::new(( - Failure::credentials( - cursor.completed_position() + 1, - error.to_string(), - attempts, - ), + Failure::credentials(error.to_string(), attempts), work, ))); continue; @@ -317,11 +251,10 @@ impl Consumer { config.send_replace(current.clone()); } match options - .finish(&mut cursor, &mut sink, ¤t, attempts, work) + .finish(&mut pending, &mut sink, ¤t, attempts, work) .await { Ok(completed) => { - progress.send_replace(cursor.completed_position()); if completed { *options.error.lock() = None; *options.recovery.lock() = None; @@ -333,80 +266,58 @@ impl Consumer { 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=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, &cursor, 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() { - journal - .bind(&cursor.route, ¤t) - .await - .map_err(|error| { - if error - .downcast_ref::() - .is_some() + 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() { - Box::new(( - Failure::legacy( - error.to_string(), - cursor.completed_position() + 1, - ¤t, - ), - Work::Resubmit, - )) + let mut failure = + options.storage(&error, attempts, Work::Resubmit); + failure.0.boundary = Boundary::Ledger; + failure } else { - options.storage(&error, &cursor, attempts, Work::Resubmit) + options.routing(&error, ¤t, attempts) } - })?; - 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); - held.0.boundary = Boundary::Ledger; - held - } else { - options.routing(&error, &cursor, ¤t, attempts) - } - })?); + })?); + } + 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) + .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, &cursor, ¤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 cursor, &mut sink, ¤t, attempts, Work::Flush) + .finish(&mut pending, &mut sink, ¤t, attempts, Work::Flush) .await .map(Some) }; @@ -418,7 +329,6 @@ impl Consumer { Ok(None) => break, Ok(Some(false)) => {} Ok(Some(true)) => { - progress.send_replace(cursor.completed_position()); *options.error.lock() = None; *options.recovery.lock() = None; attempts = 0; @@ -447,27 +357,13 @@ impl Consumer { 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)] mod tests { use super::*; use crate::{ + pipeline::journal::SpanJournal, translate::{SpanOp, SpanRow}, wire::BackendAuth, }; @@ -480,6 +376,7 @@ mod tests { reject: AtomicBool, unauthorized: AtomicBool, commit_fail: AtomicBool, + defer_until: AtomicU64, } struct Factory(Arc); struct CheckedSink { @@ -504,6 +401,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) { @@ -523,6 +423,9 @@ mod tests { } Ok(()) } + fn has_pending_delivery(&self) -> bool { + self.pending > 0 + } } fn options( dir: &std::path::Path, @@ -534,13 +437,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)), @@ -549,6 +454,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() { @@ -586,22 +512,107 @@ 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( - journal.clone(), - 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().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 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(); + 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.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(|| 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; + } + #[tokio::test] async fn ledger_recovery_does_not_resubmit_accepted_work() { let tmp = tempfile::tempdir().unwrap(); @@ -612,15 +623,14 @@ mod tests { let route = SessionRoute::default(); let record = Arc::new(Recording::default()); record.commit_fail.store(true, Ordering::SeqCst); - let consumer = Consumer::spawn( - journal.clone(), + 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); + wait_for(|| error.lock().is_some()).await; + assert_eq!(progress.completed(), 0); assert_eq!(record.accepted.load(Ordering::SeqCst), 1); - 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; @@ -637,16 +647,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( - journal.clone(), - 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().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; @@ -666,27 +673,44 @@ mod tests { }; let key = serde_json::to_string(&route).unwrap(); let record = Arc::new(Recording::default()); - let consumer = Consumer::spawn( - journal.clone(), + 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().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()), + // 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] @@ -725,12 +749,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( - journal.clone(), - 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); @@ -767,19 +787,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( - journal.clone(), - 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().is_some()).await; + assert_eq!(progress.completed(), 0); drop(rejection); Mock::given(method("POST")) .and(path("/logs3")) @@ -791,11 +809,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().is_none()); consumer.shutdown().await; } @@ -816,12 +834,11 @@ mod tests { ..Default::default() }; let record = Arc::new(Recording::default()); - let consumer = Consumer::spawn( - journal.clone(), + 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().is_some()).await; let failure: Failure = serde_json::from_str( &journal .failure(&serde_json::to_string(&route).unwrap()) @@ -846,7 +863,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 41a48e69..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), @@ -73,6 +85,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..3c336c08 100644 --- a/bt-daemon/src/pipeline/journal.rs +++ b/bt-daemon/src/pipeline/journal.rs @@ -1,14 +1,22 @@ //! Concrete durable channel: one ordered SpanOp log and a head per route. -use super::{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 parking_lot::Mutex; use rusqlite::{params, Connection, OptionalExtension}; use serde::{Deserialize, Serialize}; use std::{ path::Path, - sync::{Arc, Mutex}, + sync::{ + atomic::{AtomicBool, Ordering}, + Arc, Mutex as StorageMutex, + }, }; use tokio::sync::watch; @@ -25,7 +33,7 @@ impl std::error::Error for BindingChanged {} #[derive(Clone)] pub(crate) struct SpanJournal { - connection: Arc>>, + connection: Arc>>, changed: watch::Sender, } @@ -68,10 +76,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,12 +108,14 @@ 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))), + connection: Arc::new(StorageMutex::new(Some(connection))), changed, }) } @@ -239,8 +247,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 +280,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,9 +441,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 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, +} +#[cfg(test)] +impl Default for Progress { + fn default() -> Self { + Self { + completed: watch::channel(0).1, + published: watch::channel(0).1, + } + } +} +impl Progress { + pub fn completed(&self) -> u64 { + *self.completed.borrow() + } + pub fn pending(&self) -> u64 { + self.published.borrow().saturating_sub(self.completed()) + } + + 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().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, @@ -463,28 +552,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(); @@ -499,23 +579,115 @@ 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>> { + 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)) + } + } + } +} + +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() = Some(failure.message()); + *self.recovery.lock() = 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() = 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? { - return Ok(Some(delivery)); + if self.abandoned.load(Ordering::Acquire) { + drop(delivery); + continue; + } + self.restored = true; + return Ok(delivery); + } + tokio::select! { + _ = self.changes.changed() => {}, + _ = acknowledgements.changed() => {}, } - self.changed().await; } } } @@ -525,6 +697,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 { @@ -538,7 +723,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( @@ -546,8 +731,24 @@ 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() = 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() = Some(failure.message()); + *self.recovery.lock() = Some(failure.info(false)); + self.journal + .set_failure(&self.route, Some(serde_json::to_string(failure)?)) .await } } @@ -579,6 +780,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(); + 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..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, @@ -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 0966d709..a60437a0 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,16 @@ 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 { + self.journal.subscribe(route, legacy_through).await + } } enum State { @@ -356,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 93da136f..d3b86e02 100644 --- a/bt-daemon/src/server/sessions.rs +++ b/bt-daemon/src/server/sessions.rs @@ -119,15 +119,17 @@ 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()); } let session = Session::spawn( SessionOptions { + observer_key: key.correlation_key(), translation, - cursor, + input, + route: route.clone(), auth_provider: self.route_auth_provider(key), session_id: env.session_id.clone(), source: env.source.clone(),