Recover pipeline failures at durable boundaries

bt-daemon/src/delivery_ledger.rs

@@ -248,6 +248,20 @@ fn late_merge_id(row: &SpanRow, key: &str) -> String {
     format!("{}\u{1f}{key}", row.span_id)
 }
 
+/// The backend accepted delivery; only the local ledger write needs retrying.
+#[derive(Debug)]
+pub(crate) struct CommitFailure(pub anyhow::Error);
+impl std::fmt::Display for CommitFailure {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        self.0.fmt(f)
+    }
+}
+impl std::error::Error for CommitFailure {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        Some(self.0.as_ref())
+    }
+}
+
 /// Wrap a sink with destination-scoped terminal-span suppression. If durable
 /// ledger state cannot be loaded, callers retain normal delivery rather than
 /// risking lost trace data.
@@ -416,11 +430,15 @@ impl Sink for LedgerSink {
     async fn flush(&mut self) -> anyhow::Result<()> {
         self.inner.flush().await?;
         if let Some(ledger) = &mut self.ledger {
-            let lock = ledger.lock().await?;
-            ledger
-                .refresh(&mut self.span_origins, &mut self.pending_origins, &lock)
-                .await?;
-            ledger.commit(&lock).await?;
+            async {
+                let lock = ledger.lock().await?;
+                ledger
+                    .refresh(&mut self.span_origins, &mut self.pending_origins, &lock)
+                    .await?;
+                ledger.commit(&lock).await
+            }
+            .await
+            .map_err(|error| anyhow::Error::new(CommitFailure(error)))?;
         }
         Ok(())
     }
@@ -515,6 +533,65 @@ mod tests {
         })
     }
 
+    #[tokio::test]
+    async fn post_acceptance_ledger_failures_retain_pending_completion() {
+        for phase in ["lock", "refresh", "append"] {
+            let tmp = tempfile::tempdir().unwrap();
+            let config = config("accepted");
+            let mut sink = LedgerSink::new(
+                Box::new(RecordingSink::default()),
+                Some(tmp.path()),
+                "debug",
+                "test",
+                Some(&config),
+                "test",
+            )
+            .await;
+            sink.emit(&[terminal("one")]).await.unwrap();
+            let path = sink.ledger.as_ref().unwrap().path.clone();
+            let original = std::fs::read(&path).unwrap();
+            let permissions = std::fs::metadata(&path).unwrap().permissions();
+            match phase {
+                "lock" => {
+                    std::fs::remove_file(path.with_extension("lock")).unwrap();
+                    std::fs::create_dir(path.with_extension("lock")).unwrap();
+                }
+                "refresh" => {
+                    use std::io::Write;
+                    std::fs::OpenOptions::new()
+                        .append(true)
+                        .open(&path)
+                        .unwrap()
+                        .write_all(b"invalid JSON\n")
+                        .unwrap();
+                }
+                "append" => {
+                    let mut readonly = permissions.clone();
+                    readonly.set_readonly(true);
+                    std::fs::set_permissions(&path, readonly).unwrap();
+                }
+                _ => unreachable!(),
+            }
+            let error = sink.flush().await.unwrap_err();
+            assert!(
+                error.downcast_ref::<CommitFailure>().is_some(),
+                "{phase}: {error:#}"
+            );
+            assert!(sink.ledger.as_ref().unwrap().pending.contains("one"));
+            match phase {
+                "lock" => std::fs::remove_dir(path.with_extension("lock")).unwrap(),
+                "refresh" => std::fs::write(&path, &original).unwrap(),
+                "append" => std::fs::set_permissions(&path, permissions).unwrap(),
+                _ => unreachable!(),
+            }
+            sink.flush().await.unwrap();
+            let (restored, _) = DeliveryLedger::load(tmp.path(), "debug", "test", &config)
+                .await
+                .unwrap();
+            assert!(restored.known.contains("one"), "{phase}");
+        }
+    }
+
     #[tokio::test]
     async fn open_span_snapshots_survive_recovery_before_delivery_checkpoint() {
         let temp = tempfile::tempdir().unwrap();

bt-daemon/src/dispatch.rs

@@ -43,14 +43,6 @@ enum SessionMsg {
 }
 
 /// Where to rebuild a session's translator state from, streamed at startup.
-pub struct ReplayPlan {
-    pub journal_path: PathBuf,
-    /// Events through this offset reached the backend before the prior daemon
-    /// stopped. Replay still rebuilds translator state from them, but must not
-    /// deliver them again.
-    pub acknowledged_through: u64,
-}
-
 pub(crate) struct SessionOptions {
     pub session_id: String,
     pub source: String,
@@ -59,6 +51,7 @@ pub(crate) struct SessionOptions {
     pub data_dir: PathBuf,
     pub translation: Arc<Translation>,
     pub cursor: crate::pipeline::journal::Cursor,
+    pub auth_provider: Option<Arc<dyn crate::AuthProvider>>,
 }
 
 /// Handle to one live session: its queue plus observable counters/state.
@@ -68,6 +61,8 @@ pub struct Session {
     pub counters: Arc<Counters>,
     pub last_error: Arc<Mutex<Option<String>>>,
     source_progress: Arc<crate::pipeline::translation::Progress>,
+    pub recovery: Arc<Mutex<Option<crate::wire::RecoveryInfo>>>,
+    pub source_recovery: Arc<Mutex<Option<crate::wire::RecoveryInfo>>>,
     pub source_error: Arc<Mutex<Option<String>>>,
     pub permalink: Arc<Mutex<Option<String>>>,
     last_activity: Mutex<Instant>,
@@ -82,6 +77,7 @@ impl Session {
     ) -> Arc<Session> {
         let (tx, mut rx) = mpsc::channel(QUEUE_CAPACITY);
         let counters = Arc::new(Counters::default());
+        let recovery = Arc::new(Mutex::new(None));
         let last_error = Arc::new(Mutex::new(None));
         let permalink = Arc::new(Mutex::new(None));
         let source = options.source.clone();
@@ -98,10 +94,13 @@ impl Session {
                 error: last_error.clone(),
                 permalink: permalink.clone(),
                 factory: sink_factory,
+                auth_provider: options.auth_provider,
+                recovery: recovery.clone(),
             },
         );
         let producer = options.translation;
         let source_error = producer.error.clone();
+        let source_recovery = producer.recovery.clone();
         let source_progress = producer.progress.clone();
         let errors = last_error.clone();
         let queued = counters.clone();
@@ -190,6 +189,8 @@ impl Session {
             last_error,
             source_error,
             source_progress,
+            recovery,
+            source_recovery,
             permalink,
             last_activity: Mutex::new(Instant::now()),
         })
@@ -586,6 +587,8 @@ mod tests {
             counters: Arc::new(Counters::default()),
             last_error: Arc::new(Mutex::new(None)),
             source_error: Arc::new(Mutex::new(None)),
+            recovery: Arc::new(Mutex::new(None)),
+            source_recovery: Arc::new(Mutex::new(None)),
             source_progress: Arc::new(crate::pipeline::translation::Progress::default()),
             permalink: Arc::new(Mutex::new(None)),
             last_activity: Mutex::new(Instant::now()),