diff --git a/bt-daemon/Cargo.toml b/bt-daemon/Cargo.toml index 26675ca4..b185ac05 100644 --- a/bt-daemon/Cargo.toml +++ b/bt-daemon/Cargo.toml @@ -33,7 +33,7 @@ fs2 = "0.4" regex = "1" rquickjs = "0.12.2" rquickjs-serde = "0.6.1" -serde = { version = "1", features = ["derive"] } +serde = { version = "1", features = ["derive", "rc"] } serde_json = "1" semver = "1" sha2 = "0.10" diff --git a/bt-daemon/README.md b/bt-daemon/README.md index b01c2484..993d23eb 100644 --- a/bt-daemon/README.md +++ b/bt-daemon/README.md @@ -305,49 +305,27 @@ different profiles, organizations, projects, experiments, or parent spans. ## Recovery and storage -Capture requests return after the event is flushed to its journal. Daemon -workers handle authentication, translation, and delivery. Translation commits -canonical SpanOps and its consumed event position to a SQLite journal. Each route -has an independent durable head that advances only after checked sink acceptance -or intentional filtering. Plugin, authentication, and delivery recovery resumes -from unresolved SpanOps without replaying translation. Retries use deterministic -span IDs, so repeated delivery updates the same spans. - -Provisional session roots stay hidden until their first child arrives. Root -candidates are saved with translation's input checkpoint, independently of route -delivery, so an unused root does not hold completed turn batches open. When work -arrives, the root's saved operations are published before its child. - -Failures retain their reason and retry condition, including across restarts. -The daemon waits for the relevant plugin/configuration change, fresh credentials, -or retry delay before trying again. `bt trace doctor` reports the affected stage, -pending position, reason, and retry eligibility. - -A later event with corrected delivery settings also resumes configuration-held -subscriptions for the same source session, authentication selection, and -destination. Each subscription keeps its original durable head and pinned backend; -the correction is saved for restart recovery. Healthy routes and other destinations -retain their own settings. - -Claude, Codex, and Cursor events reference daemon-owned transcript mirrors. Cursor -preserves separate immutable generations when its native transcript is rewritten. -Grok uses -separate updates and events mirrors. Replay does not depend on the original -transcript staying at its old path. - -The daemon streams journals and transcripts, applies queue backpressure, and -retires idle sessions. Retirement preserves unresolved work and route heads; -startup resumes pending subscriptions without waiting for another event. -Translators rebuild their in-memory state from committed input when more -translation is needed, suppressing already committed output. Journals and mirrors -preserve conversation content; credentials are excluded from journaled routing data. - -Completed, inactive session partitions become eligible for cleanup after seven -days without modification. Cleanup also requires translation to be caught up, -all captured routes to have registered heads, every head to be complete, and no -pending recovery. Retained event logs protect the transcript mirrors and -correlation state needed for reconstruction. Cleanup runs at startup and hourly; -the retention period is fixed in `src/server/maintenance.rs`. +The daemon saves captured events locally before acknowledging them, then uploads +traces in the background. Pending work resumes after a daemon restart without +requiring another event from the agent. Retries update the same spans rather than +creating duplicate traces. + +Authentication, configuration, or plugin failures can delay uploads. The daemon +retries when credentials or settings change, or after a retry delay for temporary +failures. Run `bt trace doctor` to see what needs attention. Delivery to each +destination recovers independently. To send older, completed session history to a +new destination, use transcript import. + +Uploads are batched for up to 30 seconds, with earlier uploads when the batch +reaches its size limit. Turn endings, explicit flushes, and daemon shutdown drain +available traces without waiting for that interval. + +Local event logs and transcript copies contain conversation content. Credentials +are excluded from saved routing settings. Storage can grow as sessions accumulate +messages or uploads remain pending. Space used by completed delivery records is +reused, so a database file may keep its previous size after delivery catches up. +Completed sessions become eligible for cleanup after seven days without changes; +unfinished work is retained for recovery. ## Platform and integration coverage diff --git a/bt-daemon/src/correlation.rs b/bt-daemon/src/correlation.rs index 0ec46ea3..7f6f6bbb 100644 --- a/bt-daemon/src/correlation.rs +++ b/bt-daemon/src/correlation.rs @@ -311,10 +311,9 @@ impl CorrelationRegistry { } if row.end_ms.is_some() { if let Some(tools) = state.active_tools.get_mut(key) { - if let Some(tool) = tools.get_mut(&row.span_id) { - tool.active = false; - changed = true; - } + // Completed tools can no longer parent a child session. + // Duplicate terminal operations are harmless after removal. + changed |= tools.remove(&row.span_id).is_some(); } continue; } @@ -731,6 +730,25 @@ mod tests { ); } + #[test] + fn completed_tools_release_their_entries_without_retiring_the_session() { + let registry = parent_registry(); + for i in 0..10_000 { + let id = format!("tool-{i}"); + open_span(®istry, "parent", &id, SpanType::Tool); + close_span(®istry, "parent", &id); + close_span(®istry, "parent", &id); + assert!(registry.state.lock().active_tools["parent"].is_empty()); + } + assert!(registry + .active_parent_snapshot("parent") + .unwrap() + .tools + .is_empty()); + open_span(®istry, "parent", "next-tool", SpanType::Tool); + assert!(registry.has_active_tools("parent")); + } + #[test] fn parent_registry_remains_usable_after_a_reader_unwinds() { let registry = parent_registry(); diff --git a/bt-daemon/src/delivery_ledger.rs b/bt-daemon/src/delivery_ledger.rs index 8464e008..15952221 100644 --- a/bt-daemon/src/delivery_ledger.rs +++ b/bt-daemon/src/delivery_ledger.rs @@ -8,11 +8,14 @@ //! destination still receives the full trace. use crate::sink::Sink; -use crate::translate::{OriginSnapshot, SpanOp, SpanRow}; +use crate::translate::{OriginSnapshot, SpanOp, SpanRow, WeakOrigins}; +mod origins; use crate::wire::SessionConfig; +use origins::OriginIndex; use sha2::{Digest, Sha256}; use std::collections::{HashMap, HashSet}; use std::path::{Path, PathBuf}; +use std::sync::Arc; use tokio::io::{AsyncBufReadExt, AsyncSeekExt, AsyncWriteExt}; #[derive(serde::Serialize, serde::Deserialize, Default)] @@ -30,7 +33,7 @@ struct LedgerFile { )] late_merge_ids: HashSet, #[serde(default, skip_serializing_if = "HashMap::is_empty")] - span_origins: HashMap, + span_origins: HashMap>, } // Closing this file releases the OS lock, including on cancellation or errors. @@ -39,11 +42,11 @@ struct LedgerLock { } struct DeliveryLedger { + origins: OriginIndex, path: PathBuf, journal_len: u64, - known: HashSet, + // Keep only SDK deliveries awaiting a durable acknowledgement in memory. pending: HashSet, - known_late_merges: HashSet, pending_late_merges: HashSet, } @@ -53,7 +56,7 @@ impl DeliveryLedger { source: &str, session_id: &str, config: &SessionConfig, - ) -> anyhow::Result<(Self, HashMap)> { + ) -> anyhow::Result<(Self, HashMap>)> { let fingerprint = serde_json::json!({ "api_url": config.auth.api_url, "org_id": config.auth.org_id, @@ -73,18 +76,37 @@ impl DeliveryLedger { Err(error) if error.kind() == std::io::ErrorKind::NotFound => LedgerFile::default(), Err(error) => return Err(error.into()), }; + let origin_path = path.with_extension("origins.sqlite"); + tokio::fs::create_dir_all(data_dir.join("delivery-ledger")).await?; let mut ledger = Self { + origins: OriginIndex::open(origin_path).await?, path: path.with_extension("ndjson"), journal_len: 0, - known: persisted.completed_span_ids, pending: HashSet::new(), - known_late_merges: persisted.late_merge_ids, pending_late_merges: HashSet::new(), }; let lock = ledger.lock().await?; + ledger + .origins + .rebuild_completed( + persisted.completed_span_ids, + persisted.late_merge_ids, + ledger.path.clone(), + ) + .await?; + ledger + .origins + .remember(persisted.span_origins.clone()) + .await?; ledger .refresh(&mut persisted.span_origins, &mut HashMap::new(), &lock) .await?; + // Rebuild atomically under the ledger lock. The sidecar is only an + // index; an upgraded/repaired checkpoint and log remain authoritative. + ledger + .origins + .rebuild(persisted.span_origins.clone()) + .await?; Ok((ledger, persisted.span_origins)) } @@ -112,8 +134,8 @@ impl DeliveryLedger { async fn refresh( &mut self, - origins: &mut HashMap, - pending_origins: &mut HashMap, + origins: &mut HashMap>, + pending_origins: &mut HashMap>, _lock: &LedgerLock, ) -> anyhow::Result<()> { let file = match tokio::fs::File::open(&self.path).await { @@ -132,6 +154,8 @@ impl DeliveryLedger { .seek(std::io::SeekFrom::Start(self.journal_len)) .await?; let mut line = Vec::new(); + let mut fresh = HashMap::new(); + let mut through = self.journal_len; loop { line.clear(); let read = reader.read_until(b'\n', &mut line).await?; @@ -141,8 +165,12 @@ impl DeliveryLedger { break; } let update: LedgerFile = serde_json::from_slice(&line)?; - self.known.extend(update.completed_span_ids); - self.known_late_merges.extend(update.late_merge_ids); + for (span, origin) in &update.span_origins { + fresh.entry(span.clone()).or_insert_with(|| origin.clone()); + } + self.origins + .remember_completed(update.completed_span_ids, update.late_merge_ids) + .await?; for (span, origin) in update.span_origins { if pending_origins.remove(&span).is_some() { // Another writer persisted first while this writer's @@ -152,30 +180,30 @@ impl DeliveryLedger { origins.entry(span).or_insert(origin); } } - self.journal_len += read as u64; + through += read as u64; } + self.origins.remember(fresh).await?; + self.journal_len = through; Ok(()) } - fn should_emit(&self, op: &SpanOp) -> bool { + fn should_emit( + &self, + op: &SpanOp, + known: &HashSet, + known_late_merges: &HashSet, + ) -> bool { let row = match op { SpanOp::Insert(row) | SpanOp::Merge(row) => row, }; if let Some(key) = &row.late_merge_key { let id = late_merge_id(row, key); - !self.known_late_merges.contains(&id) && !self.pending_late_merges.contains(&id) + !known_late_merges.contains(&id) && !self.pending_late_merges.contains(&id) } else { - !self.known.contains(&row.span_id) && !self.pending.contains(&row.span_id) + !known.contains(&row.span_id) && !self.pending.contains(&row.span_id) } } - fn filter(&self, ops: &[SpanOp]) -> Vec { - ops.iter() - .filter(|op| self.should_emit(op)) - .cloned() - .collect() - } - fn record_emitted(&mut self, ops: &[SpanOp]) { for op in ops { let row = match op { @@ -191,9 +219,16 @@ impl DeliveryLedger { } async fn commit(&mut self, lock: &LedgerLock) -> anyhow::Result<()> { - self.pending.retain(|id| !self.known.contains(id)); + let (known, known_late_merges) = self + .origins + .completed( + self.pending.iter().cloned().collect(), + self.pending_late_merges.iter().cloned().collect(), + ) + .await?; + self.pending.retain(|id| !known.contains(id)); self.pending_late_merges - .retain(|id| !self.known_late_merges.contains(id)); + .retain(|id| !known_late_merges.contains(id)); if self.pending.is_empty() && self.pending_late_merges.is_empty() { return Ok(()); } @@ -208,9 +243,8 @@ impl DeliveryLedger { lock, ) .await?; - self.known.extend(self.pending.drain()); - self.known_late_merges - .extend(self.pending_late_merges.drain()); + self.pending.clear(); + self.pending_late_merges.clear(); Ok(()) } @@ -239,6 +273,13 @@ impl DeliveryLedger { // Match the event journal's process-crash durability: flush to the OS // before delivery without fsync on every hook event. file.flush().await?; + self.origins.remember(update.span_origins.clone()).await?; + self.origins + .remember_completed( + update.completed_span_ids.clone(), + update.late_merge_ids.clone(), + ) + .await?; self.journal_len += line.len() as u64; Ok(()) } @@ -267,8 +308,8 @@ impl std::error::Error for CommitFailure { pub(crate) struct LedgerSink { inner: Box, ledger: Option, - span_origins: HashMap, - pending_origins: HashMap, + span_origins: WeakOrigins, + pending_origins: HashMap>, plugin_version: Option, source_version: Option, bt_version: String, @@ -291,8 +332,8 @@ impl LedgerSink { _ => None, }; let (ledger, span_origins) = match loaded { - Some((ledger, origins)) => (Some(ledger), origins), - None => (None, HashMap::new()), + Some((ledger, _)) => (Some(ledger), WeakOrigins::default()), + None => (None, WeakOrigins::default()), }; Ok(Self { inner, @@ -324,17 +365,30 @@ impl Sink for LedgerSink { } async fn emit(&mut self, ops: &[SpanOp]) -> anyhow::Result { - let mut filtered = self - .ledger - .as_ref() - .map(|ledger| ledger.filter(ops)) - .unwrap_or_else(|| ops.to_vec()); + self.emit_shared( + &ops.iter() + .cloned() + .map(crate::translate::CowSpanOp::new) + .collect::>(), + ) + .await + } + + async fn emit_shared(&mut self, ops: &[crate::translate::CowSpanOp]) -> anyhow::Result { + let mut filtered = ops.to_vec(); if filtered.is_empty() { return Ok(0); } - for op in &mut filtered { - let is_insert = matches!(op, SpanOp::Insert(_)); - let row = match op { + let mut origins = HashMap::new(); + for op in &filtered { + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = op.as_ref(); + if let Some(origin) = self.span_origins.get(&row.span_id) { + origins.insert(row.span_id.clone(), origin); + } + } + for op in &filtered { + let is_insert = matches!(op.as_ref(), SpanOp::Insert(_)); + let row = match op.as_ref() { SpanOp::Insert(row) | SpanOp::Merge(row) => row, }; // Every span emitted by the integration has plugin provenance, @@ -342,13 +396,20 @@ impl Sink for LedgerSink { // an external parent that the session merely attaches to. if is_insert && self.attached_parent.as_deref() != Some(row.span_id.as_str()) { if let std::collections::hash_map::Entry::Vacant(entry) = - self.span_origins.entry(row.span_id.clone()) + origins.entry(row.span_id.clone()) { - let origin = row.origin.clone().unwrap_or_else(|| OriginSnapshot { - plugin_version: self.plugin_version.clone(), - bt_version: self.bt_version.clone(), - source_version: self.source_version.clone(), - }); + let origin = self + .pending_origins + .get(&row.span_id) + .cloned() + .or_else(|| row.origin.clone()) + .unwrap_or_else(|| { + Arc::new(OriginSnapshot { + plugin_version: self.plugin_version.clone(), + bt_version: self.bt_version.clone(), + source_version: self.source_version.clone(), + }) + }); if self.ledger.is_some() { self.pending_origins .insert(row.span_id.clone(), origin.clone()); @@ -362,19 +423,58 @@ impl Sink for LedgerSink { let lock = if let Some(ledger) = &mut self.ledger { let lock = ledger.lock().await?; ledger - .refresh(&mut self.span_origins, &mut self.pending_origins, &lock) + .refresh(&mut origins, &mut self.pending_origins, &lock) + .await?; + let saved = ledger + .origins + .find( + filtered + .iter() + .map(|op| { + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = op.as_ref(); + row.span_id.clone() + }) + .collect(), + ) + .await?; + for (span, origin) in saved { + self.pending_origins.remove(&span); + origins.insert(span, origin); + } + let (known, known_late_merges) = ledger + .origins + .completed( + filtered + .iter() + .map(|op| { + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = op.as_ref(); + row.span_id.clone() + }) + .chain(self.pending_origins.keys().cloned()) + .collect(), + filtered + .iter() + .filter_map(|op| { + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = op.as_ref(); + row.late_merge_key + .as_deref() + .map(|key| late_merge_id(row, key)) + }) + .collect(), + ) .await?; self.pending_origins.retain(|span, _| { - if ledger.known.contains(span) { + if known.contains(span) { // A peer may have acknowledged a legacy span without an // origin. Do not invent provenance for an already-saved row. + origins.remove(span); self.span_origins.remove(span); false } else { true } }); - filtered.retain(|op| ledger.should_emit(op)); + filtered.retain(|op| ledger.should_emit(op, &known, &known_late_merges)); Some(lock) } else { None @@ -383,12 +483,12 @@ impl Sink for LedgerSink { return Ok(0); } for op in &mut filtered { - let row = match op { - SpanOp::Insert(row) | SpanOp::Merge(row) => row, - }; - // Repeat each span's birth snapshot, including stateless updates. - // This also discards provenance supplied by a span-processing plugin. - row.origin = self.span_origins.get(&row.span_id).cloned(); + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = op.as_ref(); + let origin = origins.get(&row.span_id).cloned(); + if row.origin != origin { + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = op.make_mut(); + row.origin = origin; + } } if !self.pending_origins.is_empty() { // Persist birth versions before the SDK can queue a span. A crash @@ -409,13 +509,21 @@ impl Sink for LedgerSink { } // Never hold a cross-process file lock over SDK/network delivery. drop(lock); - let emitted = self.inner.emit(&filtered).await?; + for op in &filtered { + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = op.as_ref(); + if let Some(origin) = &row.origin { + self.span_origins.insert(row.span_id.clone(), origin); + } + } + let emitted = self.inner.emit_shared(&filtered).await?; // A deferred lifecycle root has not reached the backend yet. Do not // record it as delivered: a subsequent cold-worker replay needs to // retain the root until descendant work makes it exportable. if emitted > 0 { if let Some(ledger) = &mut self.ledger { - ledger.record_emitted(&filtered); + for op in &filtered { + ledger.record_emitted(std::slice::from_ref(op.as_ref())); + } } } Ok(emitted) @@ -427,7 +535,7 @@ impl Sink for LedgerSink { async { let lock = ledger.lock().await?; ledger - .refresh(&mut self.span_origins, &mut self.pending_origins, &lock) + .refresh(&mut HashMap::new(), &mut self.pending_origins, &lock) .await?; ledger.commit(&lock).await } @@ -449,6 +557,16 @@ impl Sink for LedgerSink { #[cfg(test)] mod tests { use super::*; + async fn completed(ledger: &DeliveryLedger, id: &str) -> bool { + ledger + .origins + .completed(vec![id.into()], vec![]) + .await + .unwrap() + .0 + .contains(id) + } + use crate::translate::{SpanRow, SpanType}; use crate::wire::{BackendAuth, FlushMode, TraceDestination}; use parking_lot::Mutex; @@ -491,6 +609,123 @@ mod tests { } } + #[tokio::test] + async fn expired_origins_resolve_from_disk_and_rebuild_a_missing_index() { + let temp = tempfile::tempdir().unwrap(); + let config = config("project"); + let emitted = Arc::new(Mutex::new(Vec::new())); + let mut sink = LedgerSink::new( + Box::new(RecordingSink { + emitted: emitted.clone(), + }), + Some(temp.path()), + "codex", + "weak", + Some(&config), + "birth-bt", + ) + .await + .unwrap(); + sink.set_capture_versions(Some("birth-plugin"), Some("birth-source")); + sink.emit(&[SpanOp::Insert(SpanRow { + span_id: "live".into(), + ..Default::default() + })]) + .await + .unwrap(); + let weak = { + let rows = emitted.lock(); + let SpanOp::Insert(row) = &rows[0] else { + panic!("expected insert") + }; + Arc::downgrade(row.origin.as_ref().unwrap()) + }; + emitted.lock().clear(); + assert!( + weak.upgrade().is_none(), + "neither origin cache may own the snapshot" + ); + assert!(sink.span_origins.get("live").is_none()); + sink.set_capture_versions(Some("new-plugin"), Some("new-source")); + let merge = SpanOp::Merge(SpanRow { + span_id: "live".into(), + ..Default::default() + }); + sink.emit(std::slice::from_ref(&merge)).await.unwrap(); + let index = sink + .ledger + .as_ref() + .unwrap() + .path + .with_extension("origins.sqlite"); + drop(sink); + std::fs::remove_file(index).unwrap(); + let mut recovered = LedgerSink::new( + Box::new(RecordingSink { + emitted: emitted.clone(), + }), + Some(temp.path()), + "codex", + "weak", + Some(&config), + "new-bt", + ) + .await + .unwrap(); + recovered.emit(&[merge]).await.unwrap(); + let rows = emitted.lock(); + assert_eq!(rows.len(), 2); + for op in rows.iter() { + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = op; + let origin = row.origin.as_ref().unwrap(); + assert_eq!(origin.bt_version, "birth-bt"); + assert_eq!(origin.plugin_version.as_deref(), Some("birth-plugin")); + assert_eq!(origin.source_version.as_deref(), Some("birth-source")); + } + } + + #[tokio::test] + async fn failed_origin_index_write_retries_before_sdk_delivery() { + let temp = tempfile::tempdir().unwrap(); + let config = config("project"); + let emitted = Arc::new(Mutex::new(Vec::new())); + let mut sink = LedgerSink::new( + Box::new(RecordingSink { + emitted: emitted.clone(), + }), + Some(temp.path()), + "codex", + "index-failure", + Some(&config), + "birth-bt", + ) + .await + .unwrap(); + let path = sink + .ledger + .as_ref() + .unwrap() + .path + .with_extension("origins.sqlite"); + let db = rusqlite::Connection::open(path).unwrap(); + db.execute_batch("CREATE TRIGGER fail_origin BEFORE INSERT ON origins BEGIN SELECT RAISE(FAIL,'injected'); END;").unwrap(); + let op = SpanOp::Insert(SpanRow { + span_id: "live".into(), + ..Default::default() + }); + assert!(sink.emit(std::slice::from_ref(&op)).await.is_err()); + assert!(emitted.lock().is_empty()); + db.execute_batch("DROP TRIGGER fail_origin").unwrap(); + sink.bt_version = "new-bt".into(); + sink.emit(&[op]).await.unwrap(); + let rows = emitted.lock(); + assert_eq!(rows.len(), 1); + let SpanOp::Insert(row) = &rows[0] else { + panic!("expected insert") + }; + assert_eq!(row.origin.as_ref().unwrap().bt_version, "birth-bt"); + } + fn terminal(span_id: &str) -> SpanOp { SpanOp::Insert(SpanRow { span_id: span_id.into(), @@ -527,6 +762,114 @@ mod tests { }) } + #[tokio::test] + async fn missing_index_rebuilds_completion_and_late_merge_suppression() { + let temp = tempfile::tempdir().unwrap(); + let config = config("project"); + let mut sink = LedgerSink::new( + Box::new(RecordingSink::default()), + Some(temp.path()), + "codex", + "missing-completion-index", + Some(&config), + "bt", + ) + .await + .unwrap(); + sink.emit(&[terminal("done"), late_merge("done", "usage")]) + .await + .unwrap(); + sink.flush().await.unwrap(); + let path = sink + .ledger + .as_ref() + .unwrap() + .path + .with_extension("origins.sqlite"); + drop(sink); + std::fs::remove_file(path).unwrap(); + let mut recovered = LedgerSink::new( + Box::new(RecordingSink::default()), + Some(temp.path()), + "codex", + "missing-completion-index", + Some(&config), + "bt", + ) + .await + .unwrap(); + assert_eq!( + recovered + .emit(&[ + terminal("done"), + partial_merge("done"), + late_merge("done", "usage"), + ]) + .await + .unwrap(), + 0 + ); + assert_eq!( + recovered + .emit(&[late_merge("done", "new-usage")]) + .await + .unwrap(), + 1 + ); + recovered.flush().await.unwrap(); + assert_eq!( + recovered + .emit(&[late_merge("done", "new-usage")]) + .await + .unwrap(), + 0 + ); + } + + #[tokio::test] + async fn completion_index_failure_retries_acknowledgement_without_redelivery() { + let temp = tempfile::tempdir().unwrap(); + let config = config("project"); + let emitted = Arc::new(Mutex::new(Vec::new())); + let mut sink = LedgerSink::new( + Box::new(RecordingSink { + emitted: emitted.clone(), + }), + Some(temp.path()), + "codex", + "completion-index-failure", + Some(&config), + "bt", + ) + .await + .unwrap(); + sink.emit(&[terminal("accepted")]).await.unwrap(); + let path = sink + .ledger + .as_ref() + .unwrap() + .path + .with_extension("origins.sqlite"); + let db = rusqlite::Connection::open(path).unwrap(); + db.execute_batch("CREATE TRIGGER fail_completion BEFORE INSERT ON completed BEGIN SELECT RAISE(FAIL,'injected'); END;").unwrap(); + let error = sink.flush().await.unwrap_err(); + assert!(error.downcast_ref::().is_some()); + assert!(sink.ledger.as_ref().unwrap().pending.contains("accepted")); + assert_eq!( + sink.emit(&[terminal("accepted")]) + .await + .unwrap_err() + .to_string(), + "injected" + ); + assert_eq!(emitted.lock().len(), 1); + db.execute_batch("DROP TRIGGER fail_completion").unwrap(); + sink.flush().await.unwrap(); + assert!(sink.ledger.as_ref().unwrap().pending.is_empty()); + assert_eq!(sink.emit(&[terminal("accepted")]).await.unwrap(), 0); + assert_eq!(emitted.lock().len(), 1); + } + #[tokio::test] async fn post_acceptance_ledger_failures_retain_pending_completion() { for phase in ["lock", "refresh", "append"] { @@ -583,7 +926,7 @@ mod tests { let (restored, _) = DeliveryLedger::load(tmp.path(), "debug", "test", &config) .await .unwrap(); - assert!(restored.known.contains("one"), "{phase}"); + assert!(completed(&restored, "one").await, "{phase}"); } } @@ -697,7 +1040,7 @@ mod tests { }), unexpected => panic!("unexpected span: {unexpected}"), }; - assert_eq!(row.origin, expected, "{}", row.span_id); + assert_eq!(row.origin.as_deref(), expected.as_ref(), "{}", row.span_id); } } @@ -754,25 +1097,28 @@ mod tests { ) .await .unwrap(); - assert_eq!(recovered.span_origins.len(), 256); - assert_eq!( - recovered.span_origins["span-0"].plugin_version.as_deref(), - Some("v1") - ); - assert_eq!( - recovered.span_origins["span-1"].plugin_version.as_deref(), - Some("v2") - ); + assert!(recovered.span_origins.get("span-0").is_none()); + let stored = recovered + .ledger + .as_ref() + .unwrap() + .origins + .find((0..256).map(|i| format!("span-{i}")).collect()) + .await + .unwrap(); + assert_eq!(stored.len(), 256); + assert_eq!(stored["span-0"].plugin_version.as_deref(), Some("v1")); + assert_eq!(stored["span-1"].plugin_version.as_deref(), Some("v2")); assert_eq!(recovered.emit(&[terminal("span-0")]).await.unwrap(), 0); // Simulate upgrading a JSON checkpoint with versions and completion IDs. - let (ledger, origins) = DeliveryLedger::load(temp.path(), "codex", "session-1", &config) + let (_ledger, origins) = DeliveryLedger::load(temp.path(), "codex", "session-1", &config) .await .unwrap(); tokio::fs::write( path.with_extension("json"), serde_json::to_vec(&LedgerFile { - completed_span_ids: ledger.known, - late_merge_ids: ledger.known_late_merges, + completed_span_ids: (0..256).map(|i| format!("span-{i}")).collect(), + late_merge_ids: HashSet::new(), span_origins: origins, }) .unwrap(), @@ -783,7 +1129,16 @@ mod tests { let (legacy, origins) = DeliveryLedger::load(temp.path(), "codex", "session-1", &config) .await .unwrap(); - assert_eq!(legacy.known.len(), 256); + assert_eq!( + legacy + .origins + .completed((0..256).map(|i| format!("span-{i}")).collect(), vec![]) + .await + .unwrap() + .0 + .len(), + 256 + ); assert_eq!(origins.len(), 256); assert_eq!(origins["span-0"].plugin_version.as_deref(), Some("v1")); } @@ -904,12 +1259,18 @@ mod tests { assert_eq!(origins["shared"].plugin_version.as_deref(), Some("v1")); assert_eq!(origins["shared"].bt_version, "bt1"); assert_eq!(origins["shared"].source_version.as_deref(), Some("native1")); - assert!(ledger.known.contains("parent-only")); - assert!(ledger.known_late_merges.contains("parent-only\u{1f}usage")); + assert!(completed(&ledger, "parent-only").await); + assert!(ledger + .origins + .completed(vec![], vec!["parent-only\u{1f}usage".into()]) + .await + .unwrap() + .1 + .contains("parent-only\u{1f}usage")); for i in 0..32 { for (writer, version) in [("parent", "v1"), ("child", "v2")] { let span = format!("{writer}-{i}"); - assert!(ledger.known.contains(&span), "lost completion: {span}"); + assert!(completed(&ledger, &span).await, "lost completion: {span}"); assert_eq!(origins[&span].plugin_version.as_deref(), Some(version)); } } diff --git a/bt-daemon/src/delivery_ledger/origins.rs b/bt-daemon/src/delivery_ledger/origins.rs new file mode 100644 index 00000000..acb38b39 --- /dev/null +++ b/bt-daemon/src/delivery_ledger/origins.rs @@ -0,0 +1,265 @@ +//! Disk index for birth metadata and completed IDs. The append-only ledger remains authoritative. +use super::LedgerFile; +use crate::translate::OriginSnapshot; +use parking_lot::Mutex; +use rusqlite::{Connection, OptionalExtension}; +use std::io::BufRead; +use std::{ + collections::{HashMap, HashSet}, + path::PathBuf, + sync::Arc, +}; + +pub(super) struct OriginIndex(Arc>); +impl OriginIndex { + pub async fn open(path: PathBuf) -> anyhow::Result { + tokio::task::spawn_blocking(move || { + let db = Connection::open(path)?; + db.busy_timeout(std::time::Duration::from_secs(5))?; + // This index contains only small metadata, never SpanOp payloads. + // Bound SQLite's own cache too; the ledger rebuilds a missing index. + db.execute_batch( + "PRAGMA journal_mode=WAL; PRAGMA synchronous=NORMAL; PRAGMA cache_size=-64; + CREATE TABLE IF NOT EXISTS origins(span_id TEXT PRIMARY KEY, value TEXT NOT NULL); + CREATE TABLE IF NOT EXISTS completed(kind INTEGER NOT NULL, id TEXT NOT NULL, + PRIMARY KEY(kind,id)) WITHOUT ROWID", + )?; + Ok(Self(Arc::new(Mutex::new(db)))) + }) + .await? + } + + // Rebuild membership under the same cross-process ledger lock used by writers. + // The checkpoint and append-only log remain authoritative after a crash. + pub async fn rebuild_completed( + &self, + spans: HashSet, + late_merges: HashSet, + journal: PathBuf, + ) -> anyhow::Result<()> { + let db = self.0.clone(); + tokio::task::spawn_blocking(move || { + let mut db = db.lock(); + let tx = db.transaction()?; + tx.execute("DELETE FROM completed", [])?; + { + let mut insert = + tx.prepare_cached("INSERT OR IGNORE INTO completed(kind,id) VALUES (?,?)")?; + let mut remember = + |spans: HashSet, late_merges: HashSet| -> anyhow::Result<()> { + for (kind, ids) in [(0, spans), (1, late_merges)] { + for id in ids { + insert.execute(rusqlite::params![kind, id])?; + } + } + Ok(()) + }; + remember(spans, late_merges)?; + match std::fs::File::open(journal) { + Ok(file) => { + let mut reader = std::io::BufReader::new(file); + let mut line = Vec::new(); + loop { + line.clear(); + if reader.read_until(b'\n', &mut line)? == 0 + || line.last() != Some(&b'\n') + { + break; + } + let update: LedgerFile = serde_json::from_slice(&line)?; + remember(update.completed_span_ids, update.late_merge_ids)?; + } + } + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => return Err(error.into()), + } + } + tx.commit()?; + Ok(()) + }) + .await? + } + + pub async fn remember_completed( + &self, + spans: HashSet, + late_merges: HashSet, + ) -> anyhow::Result<()> { + if spans.is_empty() && late_merges.is_empty() { + return Ok(()); + } + let db = self.0.clone(); + tokio::task::spawn_blocking(move || { + let mut db = db.lock(); + let tx = db.transaction()?; + { + let mut insert = + tx.prepare_cached("INSERT OR IGNORE INTO completed(kind,id) VALUES (?,?)")?; + for (kind, ids) in [(0, spans), (1, late_merges)] { + for id in ids { + insert.execute(rusqlite::params![kind, id])?; + } + } + } + tx.commit()?; + Ok(()) + }) + .await? + } + + // Return only membership requested by the current operation, never the + // session's complete history. No SpanOp payload is stored or read here. + pub async fn completed( + &self, + spans: Vec, + late_merges: Vec, + ) -> anyhow::Result<(HashSet, HashSet)> { + if spans.is_empty() && late_merges.is_empty() { + return Ok(Default::default()); + } + let db = self.0.clone(); + tokio::task::spawn_blocking(move || { + let db = db.lock(); + let mut query = db.prepare_cached("SELECT 1 FROM completed WHERE kind=? AND id=?")?; + let mut found = [HashSet::new(), HashSet::new()]; + for (kind, ids) in [(0, spans), (1, late_merges)] { + for id in ids { + if query.exists(rusqlite::params![kind, &id])? { + found[kind].insert(id); + } + } + } + let [spans, late_merges] = found; + Ok((spans, late_merges)) + }) + .await? + } + + pub async fn remember( + &self, + origins: HashMap>, + ) -> anyhow::Result<()> { + self.write(origins, false).await + } + + pub async fn rebuild( + &self, + origins: HashMap>, + ) -> anyhow::Result<()> { + self.write(origins, true).await + } + + async fn write( + &self, + origins: HashMap>, + replace: bool, + ) -> anyhow::Result<()> { + if origins.is_empty() && !replace { + return Ok(()); + } + let db = self.0.clone(); + tokio::task::spawn_blocking(move || { + let mut db = db.lock(); + let tx = db.transaction()?; + if replace { + tx.execute("DELETE FROM origins", [])?; + } + for (span, origin) in origins { + tx.execute( + "INSERT OR IGNORE INTO origins(span_id,value) VALUES (?,?)", + rusqlite::params![span, serde_json::to_string(&origin)?], + )?; + } + tx.commit()?; + Ok(()) + }) + .await? + } + + pub async fn find( + &self, + spans: Vec, + ) -> anyhow::Result>> { + let db = self.0.clone(); + tokio::task::spawn_blocking(move || { + let db = db.lock(); + let mut query = db.prepare_cached("SELECT value FROM origins WHERE span_id=?")?; + let mut origins = HashMap::new(); + for span in spans { + if let Some(value) = query + .query_row([&span], |r| r.get::<_, String>(0)) + .optional()? + { + origins.insert(span, serde_json::from_str(&value)?); + } + } + Ok(origins) + }) + .await? + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn membership_queries_return_only_requested_ids_and_keep_namespaces_separate() { + let temp = tempfile::tempdir().unwrap(); + let index = OriginIndex::open(temp.path().join("index.sqlite")) + .await + .unwrap(); + index + .remember_completed( + (0..50_000).map(|i| format!("span-{i}")).collect(), + HashSet::from(["late-only".into()]), + ) + .await + .unwrap(); + let (spans, merges) = index + .completed( + vec!["span-49999".into(), "late-only".into(), "missing".into()], + vec!["span-49999".into(), "late-only".into()], + ) + .await + .unwrap(); + assert_eq!(spans, HashSet::from(["span-49999".into()])); + assert_eq!(merges, HashSet::from(["late-only".into()])); + } + + #[tokio::test] + async fn failed_rebuild_preserves_membership_visible_to_existing_writers() { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("index.sqlite"); + let index = OriginIndex::open(path.clone()).await.unwrap(); + index + .remember_completed(HashSet::from(["accepted".into()]), HashSet::new()) + .await + .unwrap(); + let peer = OriginIndex::open(path).await.unwrap(); + let journal = temp.path().join("ledger.ndjson"); + std::fs::write(&journal, b"{invalid}\n").unwrap(); + assert!(index + .rebuild_completed(HashSet::new(), HashSet::new(), journal.clone()) + .await + .is_err()); + assert!(peer + .completed(vec!["accepted".into()], vec![]) + .await + .unwrap() + .0 + .contains("accepted")); + std::fs::write(&journal, b"{\"completed_span_ids\":[\"new\"]}\n{partial").unwrap(); + index + .rebuild_completed(HashSet::new(), HashSet::new(), journal) + .await + .unwrap(); + assert_eq!( + peer.completed(vec!["accepted".into(), "new".into()], vec![]) + .await + .unwrap() + .0, + HashSet::from(["new".into()]) + ); + } +} diff --git a/bt-daemon/src/dispatch.rs b/bt-daemon/src/dispatch.rs index 9522bc95..946cbddf 100644 --- a/bt-daemon/src/dispatch.rs +++ b/bt-daemon/src/dispatch.rs @@ -12,7 +12,7 @@ use crate::sink::SinkFactory; use crate::wire::Envelope; use parking_lot::Mutex; use std::path::PathBuf; -use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::atomic::AtomicU64; use std::sync::Arc; use std::time::Instant; use tokio::sync::{mpsc, oneshot}; @@ -159,7 +159,7 @@ impl Session { }); } SessionMsg::Shutdown(target, reply) => { - let last = producer.routes.fetch_sub(1, Ordering::Relaxed) == 1; + let last = producer.begin_route_shutdown(); let target = match target { Some(tail) => Some(tail), None if last => producer @@ -177,9 +177,7 @@ impl Session { } producer.unobserve(observer_key.clone()).await; route.shutdown().await; - if producer.routes.load(Ordering::Relaxed) == 0 { - producer.shutdown().await; - } + producer.finish_route_shutdown().await; let _ = reply.send(()); break; } diff --git a/bt-daemon/src/pipeline/consumer.rs b/bt-daemon/src/pipeline/consumer.rs index 6876c240..64810ec0 100644 --- a/bt-daemon/src/pipeline/consumer.rs +++ b/bt-daemon/src/pipeline/consumer.rs @@ -9,7 +9,6 @@ use crate::{ delivery_ledger::{CommitFailure, LedgerSink}, dispatch::Counters, sink::{Sink, SinkFactory}, - translate::SpanOp, wire::{RecoveryInfo, SessionConfig, SessionRoute}, AuthProvider, AuthResolveReason, }; @@ -151,7 +150,9 @@ impl ConsumerOptions { Work::Resubmit => { // Dropping unresolved obligations asks their source to replay. // Recreate the consumed SDK batch and its in-memory ledger too. - *sink = None; + if let Some(previous) = sink.take() { + previous.shutdown().await; + } pending.clear(); return Ok(false); } @@ -189,7 +190,7 @@ impl ConsumerOptions { impl Consumer { pub fn spawn( - mut input: Box>>, + mut input: Box>>, options: ConsumerOptions, ) -> Arc { let config = options.config.clone(); @@ -308,7 +309,7 @@ impl Consumer { })?; routing.configure(¤t); routing - .emit(ops) + .emit_shared(ops) .await .map_err(|error| options.routing(&error, ¤t, attempts)) } @@ -340,6 +341,9 @@ impl Consumer { Err(failure) => held = Some(failure), } } + if let Some(sink) = sink { + sink.shutdown().await; + } }); *handle.task.lock() = Some(task); handle @@ -606,7 +610,8 @@ mod tests { root_span_id: "one".into(), end_ms: Some(1), ..Default::default() - })], + }) + .into()], Acknowledged(acknowledged.clone()), )) .await diff --git a/bt-daemon/src/pipeline/journal.rs b/bt-daemon/src/pipeline/journal.rs index 16e03446..9d8067f0 100644 --- a/bt-daemon/src/pipeline/journal.rs +++ b/bt-daemon/src/pipeline/journal.rs @@ -1,11 +1,12 @@ //! Concrete durable channel: one ordered SpanOp log and a head per route. +use super::live::{Multiplexer, Receiver, Record, SharedOp, BYTE_BUDGET, WINDOW}; use super::{ delivery::InputPort, recovery::{Boundary, Failure, Recovery, RecoveryOutcome}, Completion, Delivery, Stage, }; use crate::{ - translate::SpanOp, + translate::{OriginSnapshot, SpanOp, SpanRow, SpanType}, wire::{RecoveryInfo, SessionConfig, SessionRoute}, }; use parking_lot::Mutex; @@ -32,10 +33,16 @@ impl std::fmt::Display for BindingChanged { } impl std::error::Error for BindingChanged {} +type ReplayWindow = Vec<(u64, std::sync::Weak, usize)>; + #[derive(Clone)] pub(crate) struct SpanJournal { connection: Arc>>, changed: watch::Sender, + live: Arc, + replay: Arc>, + #[cfg(test)] + payload_reads: Arc, } /// Explicit codec preserves fields intentionally omitted from wire/plugin JSON. @@ -51,11 +58,19 @@ struct StoredOp { impl StoredOp { fn encode(op: &SpanOp) -> anyhow::Result> { let (SpanOp::Insert(row) | SpanOp::Merge(row)) = op; - Ok(serde_json::to_vec(&Self { + #[derive(Serialize)] + struct Borrowed<'a> { + version: u32, + op: &'a SpanOp, + late_merge_key: &'a Option, + turn_span_id: &'a Option, + is_turn: bool, + } + Ok(serde_json::to_vec(&Borrowed { version: 1, - op: op.clone(), - late_merge_key: row.late_merge_key.clone(), - turn_span_id: row.turn_span_id.clone(), + op, + late_merge_key: &row.late_merge_key, + turn_span_id: &row.turn_span_id, is_turn: row.is_turn, })?) } @@ -74,6 +89,84 @@ impl StoredOp { } } +/// Metadata needed to restore correlation and provenance without retaining +/// conversation input/output after all delivery obligations are complete. +pub(super) fn metadata_only(op: &SpanOp) -> SpanOp { + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = op; + let row = SpanRow { + span_id: row.span_id.clone(), + root_span_id: row.root_span_id.clone(), + parent_span_ids: row.parent_span_ids.clone(), + name: row.name.clone(), + span_type: row.span_type, + start_ms: row.start_ms, + end_ms: row.end_ms, + metadata: row.metadata.clone(), + origin: row.origin.clone(), + is_turn: row.is_turn, + ..Default::default() + }; + match op { + SpanOp::Insert(_) => SpanOp::Insert(row), + SpanOp::Merge(_) => SpanOp::Merge(row), + } +} +fn save_metadata(db: &Connection, op: &SpanOp, observation: bool) -> anyhow::Result<()> { + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = op; + if let Some(origin) = &row.origin { + db.execute( + "INSERT OR IGNORE INTO origins(span_id,value) VALUES (?,?)", + params![row.span_id, serde_json::to_string(origin)?], + )?; + } + if !observation { + return Ok(()); + } + if matches!(op, SpanOp::Insert(_)) + && (row.span_type == SpanType::Tool + || (row.span_type == SpanType::Task && row.span_id == row.root_span_id)) + { + let mut birth = metadata_only(op); + let (SpanOp::Insert(birth_row) | SpanOp::Merge(birth_row)) = &mut birth; + birth_row.end_ms = None; + db.execute( + "INSERT OR REPLACE INTO span_state(span_id,kind,value,end_ms) VALUES (?,?,?,?)", + params![ + row.span_id, + if row.span_type == SpanType::Tool { + "tool" + } else { + "root" + }, + StoredOp::encode(&birth)?, + row.end_ms + ], + )?; + } else { + db.execute( + "UPDATE span_state SET end_ms=? WHERE span_id=?", + params![row.end_ms, row.span_id], + )?; + } + db.execute( + "DELETE FROM span_state WHERE span_id=? AND kind='tool' AND end_ms IS NOT NULL", + [&row.span_id], + )?; + Ok(()) +} +fn prune(db: &Connection) -> anyhow::Result<()> { + let through: Option = db.query_row("SELECT min(position) FROM heads", [], |r| r.get(0))?; + if let Some(through) = through { + db.execute("DELETE FROM ops WHERE position<=?", [through])?; + db.execute( + "UPDATE publication SET floor=max(floor,?) WHERE id=1", + [through], + )?; + db.execute("DELETE FROM receipts WHERE EXISTS(SELECT 1 FROM batches WHERE batches.input=receipts.input AND committed=1) AND NOT EXISTS(SELECT 1 FROM ops WHERE ops.input=receipts.input)", [])?; + } + Ok(()) +} + impl SpanJournal { pub async fn open(path: &Path) -> anyhow::Result { let path = path.to_owned(); @@ -83,7 +176,7 @@ impl SpanJournal { let db = Connection::open(path)?; db.busy_timeout(std::time::Duration::from_secs(5))?; let version: u32 = db.query_row("PRAGMA user_version", [], |row| row.get(0))?; - anyhow::ensure!(version <= 5, "unsupported SpanOp journal schema {version}"); + anyhow::ensure!(version <= 6, "unsupported SpanOp journal schema {version}"); db.execute_batch("PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL;")?; if version < 3 { db.execute_batch( @@ -115,7 +208,27 @@ impl SpanJournal { if version < 5 { db.execute_batch("BEGIN IMMEDIATE; ALTER TABLE heads ADD COLUMN settings TEXT; ALTER TABLE heads ADD COLUMN settings_through INTEGER NOT NULL DEFAULT 0; PRAGMA user_version=5; COMMIT")?; } - let tail = db.query_row("SELECT coalesce(max(position),0) FROM ops JOIN batches USING(input) WHERE committed=1", [], |row| row.get(0))?; + if version < 6 { + db.execute_batch("BEGIN IMMEDIATE; + DELETE FROM ops WHERE input IN (SELECT input FROM batches WHERE committed=0); + CREATE TABLE receipts (input INTEGER NOT NULL, ordinal INTEGER NOT NULL, PRIMARY KEY(input,ordinal)); + CREATE TABLE publication (id INTEGER PRIMARY KEY CHECK(id=1), tail INTEGER NOT NULL, floor INTEGER NOT NULL); + INSERT INTO publication VALUES(1,(SELECT coalesce(max(position),0) FROM ops),0); + CREATE TABLE origins (span_id TEXT PRIMARY KEY, value TEXT NOT NULL); + CREATE TABLE span_state (span_id TEXT PRIMARY KEY, kind TEXT NOT NULL, value BLOB NOT NULL, end_ms INTEGER); + ")?; + // One bounded upgrade pass extracts small recovery metadata. + let mut after = 0; + loop { + let record: Option<(u64,Vec,bool)> = db.query_row("SELECT position,value,observation FROM ops JOIN batches USING(input) WHERE position>? ORDER BY position LIMIT 1", [after], |r| Ok((r.get(0)?,r.get(1)?,r.get(2)?))).optional()?; + let Some((position,value,observation)) = record else { break; }; + save_metadata(&db, &StoredOp::decode(&value)?, observation)?; + after = position; + } + prune(&db)?; + db.execute_batch("PRAGMA user_version=6; COMMIT")?; + } + let tail = db.query_row("SELECT tail FROM publication WHERE id=1", [], |row| row.get(0))?; Ok((db, tail)) }) .await??; @@ -124,6 +237,10 @@ impl SpanJournal { Ok(Self { connection: Arc::new(StorageMutex::new(Some(connection))), changed, + live: Arc::new(Multiplexer::default()), + replay: Arc::new(tokio::sync::Mutex::new(Vec::new())), + #[cfg(test)] + payload_reads: Arc::default(), }) } @@ -172,7 +289,7 @@ impl SpanJournal { .await } - /// Restart abandons only invisible chunks; a committed identity is never rewritten. + /// Restart abandons only invisible receipts; a committed identity is never rewritten. pub async fn begin(&self, input: u64, observation: bool) -> anyhow::Result { self.with_db(move |db| { let tx = db.transaction()?; @@ -186,7 +303,14 @@ impl SpanJournal { if done == Some(true) { return Ok(false); } - tx.execute("DELETE FROM ops WHERE input=?", [input])?; + let published: bool = tx.query_row( + "SELECT EXISTS(SELECT 1 FROM receipts WHERE input=?)", + [input], + |r| r.get(0), + )?; + if !published { + tx.execute("DELETE FROM ops WHERE input=?", [input])?; + } tx.execute( "INSERT OR REPLACE INTO batches(input,committed,observation) VALUES (?,0,?)", params![input, observation], @@ -197,6 +321,146 @@ impl SpanJournal { .await } + /// Journal available operations and hand the original instances to routes. + /// Per-input operation ordinals survive a crash halfway through translation, + /// so rebuilding translator state does not publish an accepted prefix twice. + pub async fn append_live( + &self, + input: u64, + ordinal: &mut u64, + ops: &mut Vec, + ) -> anyhow::Result<()> { + while !ops.is_empty() { + let started = std::time::Instant::now(); + let mut encoded = Vec::new(); + let mut bytes = 0; + for op in ops.iter() { + let value = StoredOp::encode(op)?; + if !encoded.is_empty() && bytes + value.len() > BYTE_BUDGET { + break; + } + bytes += value.len(); + encoded.push(value); + if bytes >= BYTE_BUDGET || started.elapsed() >= WINDOW { + break; + } + } + let sizes = encoded.iter().map(Vec::len).collect::>(); + let metadata = ops + .iter() + .take(encoded.len()) + .map(metadata_only) + .collect::>(); + let first = *ordinal; + let positions = self + .with_db(move |db| { + let tx = db.transaction()?; + let mut positions = Vec::new(); + let (committed, observation): (bool, bool) = tx.query_row( + "SELECT committed,observation FROM batches WHERE input=?", + [input], + |r| Ok((r.get(0)?, r.get(1)?)), + )?; + for (index, value) in encoded.into_iter().enumerate() { + let ordinal = first + index as u64; + let exists = tx.query_row( + "SELECT EXISTS(SELECT 1 FROM receipts WHERE input=? AND ordinal=?)", + params![input, ordinal], + |r| r.get::<_, bool>(0), + )?; + if exists { + positions.push(None); + continue; + } + anyhow::ensure!(!committed, "cannot change a completed translation input"); + save_metadata(&tx, &metadata[index], observation)?; + tx.execute( + "INSERT INTO ops(input,value) VALUES (?,?)", + params![input, value], + )?; + let position = tx.last_insert_rowid() as u64; + positions.push(Some(position)); + tx.execute("UPDATE publication SET tail=? WHERE id=1", [position])?; + tx.execute( + "INSERT INTO receipts(input,ordinal) VALUES (?,?)", + params![input, ordinal], + )?; + } + tx.commit()?; + Ok(positions) + }) + .await?; + let count = positions.len(); + *ordinal += count as u64; + for ((position, bytes), op) in positions.into_iter().zip(sizes).zip(ops.drain(..count)) + { + if let Some(position) = position { + self.live + .publish(Record { + position, + bytes, + op: SharedOp::new(op), + }) + .await; + self.changed.send_replace(position); + } + } + } + Ok(()) + } + + /// One serialized, weakly held recovery window is shared by all routes. + /// No decoded payload remains alive solely because of the replay cache. + async fn recover_after(&self, after: u64) -> anyhow::Result> { + let mut cache = self.replay.lock().await; + if cache + .first() + .is_some_and(|(position, _, _)| *position <= after + 1) + && cache + .last() + .is_some_and(|(position, _, _)| *position > after) + { + let shared = cache + .iter() + .filter(|(position, _, _)| *position > after) + .map(|(position, op, bytes)| { + Some(Record { + position: *position, + op: SharedOp::from_arc(op.upgrade()?), + bytes: *bytes, + }) + }) + .collect::>>(); + if let Some(shared) = shared { + return Ok(shared); + } + } + let records = self.with_db(move |db| { + let mut statement = db.prepare("SELECT position,value FROM ops JOIN batches USING(input) WHERE (committed=1 OR EXISTS(SELECT 1 FROM receipts WHERE receipts.input=ops.input)) AND position>? ORDER BY position")?; + let mut rows = statement.query([after])?; + let started = std::time::Instant::now(); + let mut records = Vec::new(); + let mut bytes = 0; + while let Some(row) = rows.next()? { + let value = row.get_ref(1)?.as_blob()?; + if !records.is_empty() && bytes + value.len() > BYTE_BUDGET { break; } + bytes += value.len(); + records.push(Record { position:row.get(0)?, op:SharedOp::new(StoredOp::decode(value)?), bytes:value.len() }); + if bytes >= BYTE_BUDGET || started.elapsed() >= WINDOW { break; } + } + Ok(records) + }).await?; + #[cfg(test)] + self.payload_reads + .fetch_add(records.len() as u64, Ordering::Relaxed); + *cache = records + .iter() + .map(|record| (record.position, record.op.downgrade(), record.bytes)) + .collect(); + Ok(records) + } + + #[cfg(test)] pub async fn append(&self, input: u64, ops: &[SpanOp]) -> anyhow::Result<()> { let encoded = ops .iter() @@ -211,6 +475,7 @@ impl SpanJournal { )?; anyhow::ensure!(!done, "cannot change a committed translation input"); for value in encoded { + save_metadata(&tx, &StoredOp::decode(&value)?, true)?; tx.execute( "INSERT INTO ops(input,value) VALUES (?,?)", params![input, value], @@ -309,6 +574,11 @@ impl SpanJournal { [input], |r| r.get(0), )?; + let tail: u64 = tail; + tx.execute("UPDATE publication SET tail=max(tail,?) WHERE id=1", [tail])?; + let tail = + tx.query_row("SELECT tail FROM publication WHERE id=1", [], |r| r.get(0))?; + prune(&tx)?; tx.commit()?; Ok(tail) }) @@ -322,11 +592,30 @@ impl SpanJournal { route: &SessionRoute, legacy_through: u64, ) -> anyhow::Result { + let live = self.live.subscribe(); let route = serde_json::to_string(route)?; let key = route.clone(); let position = self .with_db(move |db| { let tx = db.transaction()?; + // Auth can register a named-profile obligation before learning + // its stable ID. Move that unresolved placeholder to the resolved + // identity instead of leaving a second head that can never run. + let candidate: SessionRoute = serde_json::from_str(&key)?; + if candidate.auth.profile_id.is_some() { + let aliases = { + let mut query = tx.prepare("SELECT route FROM heads WHERE binding IS NULL AND route<>?")?; + let rows = query.query_map([&key], |r|r.get::<_,String>(0))?; + rows.collect::,_>>()? + }; + for alias in aliases { + let previous: SessionRoute = serde_json::from_str(&alias)?; + if previous.auth.profile_id.is_none() && previous.same_route(&candidate) { + tx.execute("UPDATE OR IGNORE heads SET route=? WHERE route=?", params![key,alias])?; + tx.execute("DELETE FROM heads WHERE route=?",[alias])?; + } + } + } // Seed only a new head, using released event checkpoints where provable. let legacy: u64 = tx.query_row( "SELECT coalesce(max(position),0) FROM ops JOIN batches USING(input) @@ -335,7 +624,7 @@ impl SpanJournal { |r| r.get(0), )?; tx.execute( - "INSERT OR IGNORE INTO heads(route,position) VALUES (?,?)", + "INSERT OR IGNORE INTO heads(route,position) VALUES (?,max(?,(SELECT floor FROM publication WHERE id=1)))", params![key, legacy], )?; let position = @@ -354,18 +643,22 @@ impl SpanJournal { completed: position, changes: self.changed.subscribe(), progress, + drain: watch::channel(position).0, abandoned: Arc::new(AtomicBool::new(false)), settings: None, version: String::new(), restored: false, + live, + queued: None, error: Arc::new(Mutex::new(None)), recovery: Arc::new(Mutex::new(None)), }) } + #[cfg(test)] pub async fn read_after(&self, after: u64, limit: usize) -> anyhow::Result> { self.with_db(move |db| { - let mut statement = db.prepare("SELECT position,value FROM ops JOIN batches USING(input) WHERE committed=1 AND observation=1 AND position>? ORDER BY position LIMIT ?")?; + let mut statement = db.prepare("SELECT position,value FROM ops JOIN batches USING(input) WHERE (committed=1 OR EXISTS(SELECT 1 FROM receipts WHERE receipts.input=ops.input)) AND observation=1 AND position>? ORDER BY position LIMIT ?")?; let rows = statement.query_map(params![after,limit as u64], |r| Ok((r.get::<_,u64>(0)?,r.get::<_,Vec>(1)?)))?; rows.map(|row| { let (position,bytes)=row?; Ok((position,StoredOp::decode(&bytes)?)) }).collect() }).await @@ -373,30 +666,115 @@ impl SpanJournal { pub async fn tail(&self) -> anyhow::Result { self.with_db(|db| { - Ok(db.query_row( - "SELECT coalesce(max(position),0) FROM ops JOIN batches USING(input) - WHERE committed=1", - [], - |r| r.get(0), - )?) + Ok(db.query_row("SELECT tail FROM publication WHERE id=1", [], |r| r.get(0))?) }) .await } + pub async fn origins_for( + &self, + spans: Vec, + ) -> anyhow::Result>> { + if spans.is_empty() { + return Ok(BTreeMap::new()); + } + self.with_db(move |db| { + let mut query = db.prepare_cached("SELECT value FROM origins WHERE span_id=?")?; + let mut origins = BTreeMap::new(); + for span in spans { + if let Some(value) = query + .query_row([&span], |r| r.get::<_, String>(0)) + .optional()? + { + origins.insert(span, serde_json::from_str(&value)?); + } + } + Ok(origins) + }) + .await + } + + #[cfg(test)] + pub async fn origins(&self) -> anyhow::Result> { + self.with_db(|db| { + let mut query = db.prepare("SELECT span_id,value FROM origins")?; + let rows = + query.query_map([], |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)))?; + rows.map(|row| { + let (id, value) = row?; + Ok((id, serde_json::from_str(&value)?)) + }) + .collect() + }) + .await + } + + pub async fn observations(&self) -> anyhow::Result> { + self.with_db(|db| { + let mut query = db.prepare("SELECT value,end_ms FROM span_state ORDER BY rowid")?; + let rows = query.query_map([], |r| { + Ok((r.get::<_, Vec>(0)?, r.get::<_, Option>(1)?)) + })?; + let mut result = Vec::new(); + for row in rows { + let (bytes, end) = row?; + let op = StoredOp::decode(&bytes)?; + let (SpanOp::Insert(birth) | SpanOp::Merge(birth)) = &op; + if let Some(end_ms) = end { + let mut terminal = birth.clone(); + terminal.end_ms = Some(end_ms); + result.push(op); + result.push(SpanOp::Merge(terminal)); + } else { + result.push(op); + } + } + Ok(result) + }) + .await + } + + pub async fn register(&self, route: &SessionRoute) -> anyhow::Result<()> { + let route = serde_json::to_string(route)?; + self.with_db(move |db| { + let candidate: SessionRoute = serde_json::from_str(&route)?; + let existing = { + let mut query = db.prepare("SELECT route FROM heads")?; + let rows = query.query_map([], |r|r.get::<_,String>(0))?; + let mut existing = false; + for row in rows { if serde_json::from_str::(&row?)?.same_route(&candidate) { existing = true; break; } } + existing + }; + if !existing { db.execute("INSERT OR IGNORE INTO heads(route,position) VALUES (?,(SELECT floor FROM publication WHERE id=1))",[route])?; } + Ok(()) + }).await + } + /// A completed subscription can remain dormant until its source has new work. pub async fn route_complete(&self, route: &SessionRoute, input: u64) -> anyhow::Result { let route = route.clone(); self.with_db(move |db| { - let committed: u64 = db.query_row("SELECT coalesce(max(input),0) FROM batches WHERE committed=1",[],|r|r.get(0))?; - if committed < input {return Ok(false);} - let tail: u64 = db.query_row("SELECT coalesce(max(position),0) FROM ops JOIN batches USING(input) WHERE committed=1",[],|r|r.get(0))?; - let mut query = db.prepare("SELECT route FROM heads WHERE failure IS NULL AND position>=?")?; - for stored in query.query_map([tail],|r|r.get::<_,String>(0))? { + let committed: u64 = db.query_row( + "SELECT coalesce(max(input),0) FROM batches WHERE committed=1", + [], + |r| r.get(0), + )?; + if committed < input { + return Ok(false); + } + let tail: u64 = + db.query_row("SELECT tail FROM publication WHERE id=1", [], |r| r.get(0))?; + let mut query = + db.prepare("SELECT route FROM heads WHERE failure IS NULL AND position>=?")?; + for stored in query.query_map([tail], |r| r.get::<_, String>(0))? { let stored: SessionRoute = serde_json::from_str(&stored?)?; - if stored.same_route(&route) {return Ok(true);} + if stored.same_route(&route) { + return Ok(true); + } } Ok(false) - }).await + }) + .await } /// Tokens can change; a durable head's backend and organization cannot. @@ -522,10 +900,11 @@ impl SpanJournal { }) .await } - /// Whole-partition retention only; an empty subscriber set is not completion. + /// Remaining metadata can expire once input and every route are complete. + /// An empty subscriber set is not completion. pub async fn collectible(&self, input: u64) -> anyhow::Result { self.with_db(move |db| Ok(db.query_row("SELECT EXISTS(SELECT 1 FROM heads) - AND NOT EXISTS(SELECT 1 FROM heads WHERE failure IS NOT NULL OR position < (SELECT coalesce(max(position),0) FROM ops JOIN batches USING(input) WHERE committed=1)) + AND NOT EXISTS(SELECT 1 FROM heads WHERE failure IS NOT NULL OR position < (SELECT tail FROM publication WHERE id=1)) AND NOT EXISTS(SELECT 1 FROM recovery WHERE failure IS NOT NULL) AND NOT EXISTS(SELECT 1 FROM batches WHERE committed=0) AND (SELECT coalesce(max(input),0) FROM batches WHERE committed=1)>=?",[input],|r|r.get(0))?)).await @@ -592,10 +971,13 @@ pub(crate) struct Cursor { completed: u64, changes: watch::Receiver, progress: watch::Sender, + drain: watch::Sender, abandoned: Arc, settings: Option>, version: String, restored: bool, + live: Receiver, + queued: Option, error: Arc>>, recovery: Arc>>, } @@ -606,6 +988,7 @@ pub(crate) struct Cursor { pub(crate) struct Progress { completed: watch::Receiver, published: watch::Receiver, + drain: watch::Sender, } #[cfg(test)] impl Default for Progress { @@ -613,6 +996,7 @@ impl Default for Progress { Self { completed: watch::channel(0).1, published: watch::channel(0).1, + drain: watch::channel(0).0, } } } @@ -625,6 +1009,16 @@ impl Progress { } pub async fn through(&self, target: u64, error: &Mutex>) -> Result<(), String> { + // Flush callers already know the publication boundary they need. Wake + // collection and keep draining promptly until that boundary is reported. + self.drain.send_if_modified(|requested| { + if target > *requested { + *requested = target; + true + } else { + false + } + }); let mut completed = self.completed.clone(); loop { if *completed.borrow_and_update() >= target { @@ -642,7 +1036,7 @@ impl Progress { } pub(crate) struct Subscription { - pub input: Box>>, + pub input: Box>>, pub progress: Progress, } @@ -657,17 +1051,19 @@ impl Cursor { let progress = Progress { completed: self.progress.subscribe(), published: self.changes.clone(), + drain: self.drain.clone(), }; self.settings = Some(settings); self.version = version; self.error = error; self.recovery = recovery; Subscription { - input: Box::new(self), + input: Box::new(SharedCursor(self)), progress, } } + #[cfg(test)] pub async fn next_batch( &mut self, limit: usize, @@ -679,7 +1075,7 @@ impl Cursor { .with_db(move |db| { let mut statement = db.prepare( "SELECT position,value FROM ops JOIN batches USING(input) - WHERE committed=1 AND position>? ORDER BY position LIMIT ?", + WHERE (committed=1 OR EXISTS(SELECT 1 FROM receipts WHERE receipts.input=ops.input)) AND position>? ORDER BY position LIMIT ?", )?; let rows = statement.query_map(params![after, limit as u64], |r| { Ok((r.get::<_, u64>(0)?, r.get::<_, Vec>(1)?)) @@ -699,6 +1095,77 @@ impl Cursor { ops, HeadCompletion { journal: self.journal.clone(), + live: self.live.control(), + 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(), + }, + ))) + } + + async fn next_shared(&mut self) -> anyhow::Result>>> { + let published = *self.changes.borrow(); + if self.read >= published { + self.live.resume(); + return Ok(None); + } + let after = self.read; + let first = self.queued.take().or_else(|| self.live.try_recv()); + let mut records = Vec::new(); + if let Some(first) = first { + if first.position == after + 1 { + records.push(first); + } + } + if records.is_empty() { + // Restart, abandoned delivery, a late subscription, or a bounded + // live-queue overrun. Healthy live delivery never enters this path. + records = self.journal.recover_after(after).await?; + self.live.clear(); + } else { + let deadline = tokio::time::Instant::now() + WINDOW; + let mut drain = self.drain.subscribe(); + let mut bytes = records[0].bytes; + while bytes < BYTE_BUDGET { + let next = if *drain.borrow_and_update() > after { + match self.live.try_recv() { + Some(record) => record, + None => break, + } + } else { + tokio::select! { + next = tokio::time::timeout_at(deadline, self.live.recv()) => match next { + Ok(Some(record)) => record, + _ => break, + }, + _ = drain.changed() => continue, + } + }; + if next.position != records.last().map_or(after, |record| record.position) + 1 + || bytes + next.bytes > BYTE_BUDGET + { + self.queued = Some(next); + break; + } + bytes += next.bytes; + records.push(next); + } + } + let Some(last) = records.last() else { + return Ok(None); + }; + let through = last.position; + self.read = through; + Ok(Some(Delivery::new( + records.into_iter().map(|record| record.op).collect(), + HeadCompletion { + journal: self.journal.clone(), + live: self.live.control(), route: self.route.clone(), expected: after, through, @@ -729,6 +1196,8 @@ impl Cursor { }) .await?; self.read = self.completed; + self.queued = None; + self.live.clear(); self.progress.send_replace(self.completed); Ok(()) } @@ -739,24 +1208,28 @@ impl Cursor { } } +struct SharedCursor(Cursor); + #[async_trait::async_trait] -impl InputPort for Cursor { - type Item = Vec; +impl InputPort for SharedCursor { + type Item = Vec; async fn receive(&mut self) -> anyhow::Result>> { - let result = self.receive_managed().await; + let result = self.0.receive_managed().await; match result { Ok(delivery) => Ok(Some(delivery)), Err(error) => { + self.0.live.control().pause(); 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); + let failure = Failure::storage(&error, &self.0.version, 0).at(self.0.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 + .0 .journal - .set_failure(&self.route, Some(serde_json::to_string(&failure)?)) + .set_failure(&self.0.route, Some(serde_json::to_string(&failure)?)) .await { tracing::warn!(%error, "input recovery evidence could not be persisted"); @@ -768,7 +1241,7 @@ impl InputPort for Cursor { } impl Cursor { - async fn receive_managed(&mut self) -> anyhow::Result>> { + async fn receive_managed(&mut self) -> anyhow::Result>> { let mut acknowledgements = self.progress.subscribe(); loop { if self.abandoned.load(Ordering::Acquire) { @@ -826,7 +1299,7 @@ impl Cursor { return Err(error); } } - if let Some(delivery) = self.next_batch(128).await? { + if let Some(delivery) = self.next_shared().await? { if self.abandoned.load(Ordering::Acquire) { drop(delivery); continue; @@ -842,8 +1315,20 @@ impl Cursor { } } +#[cfg(test)] +#[async_trait::async_trait] +impl InputPort for Cursor { + type Item = Vec; + async fn receive(&mut self) -> anyhow::Result>> { + Ok(Some(self.receive_managed().await?.map(|ops| { + ops.into_iter().map(|op| op.as_ref().clone()).collect() + }))) + } +} + struct HeadCompletion { journal: SpanJournal, + live: super::live::Control, route: String, expected: u64, through: u64, @@ -856,6 +1341,7 @@ struct HeadCompletion { impl Drop for HeadCompletion { fn drop(&mut self) { if !self.acknowledged { + self.live.pause(); self.abandoned.store(true, Ordering::Release); self.progress.send_modify(|_| {}); } @@ -880,6 +1366,7 @@ impl Completion for HeadCompletion { "UPDATE heads SET position=?, failure=NULL WHERE route=?", params![through, route], )?; + prune(&tx)?; tx.commit()?; Ok(through) }) @@ -893,6 +1380,7 @@ impl Completion for HeadCompletion { } async fn failed(&mut self, failure: &mut Failure) -> anyhow::Result<()> { + self.live.pause(); failure.position = self.expected + 1; failure.through = Some(self.through); *self.error.lock() = Some(failure.message()); @@ -915,11 +1403,14 @@ mod tests { late_merge_key: Some("late".into()), turn_span_id: Some("turn".into()), is_turn: true, - origin: Some(OriginSnapshot { - plugin_version: Some("capture".into()), - source_version: None, - bt_version: "test".into(), - }), + origin: Some( + OriginSnapshot { + plugin_version: Some("capture".into()), + source_version: None, + bt_version: "test".into(), + } + .into(), + ), ..SpanRow::default() }) } @@ -930,6 +1421,339 @@ mod tests { } } + #[tokio::test] + async fn schema_five_upgrade_reclaims_acknowledged_payloads_without_losing_provenance() { + let tmp = tempfile::tempdir().unwrap(); + let path = tmp.path().join("upgrade.sqlite"); + let journal = SpanJournal::open(&path).await.unwrap(); + journal.register(&route("old")).await.unwrap(); + journal.begin(10, true).await.unwrap(); + journal.append(10, &[op("old")]).await.unwrap(); + journal.commit(10).await.unwrap(); + journal.close().await.unwrap(); + let db = Connection::open(&path).unwrap(); + db.execute_batch("UPDATE heads SET position=1; DROP TABLE receipts; DROP TABLE publication; DROP TABLE origins; DROP TABLE span_state; PRAGMA user_version=5").unwrap(); + drop(db); + let upgraded = SpanJournal::open(&path).await.unwrap(); + assert_eq!(upgraded.tail().await.unwrap(), 1); + assert_eq!(upgraded.origins().await.unwrap()["old"].bt_version, "test"); + assert_eq!( + upgraded + .with_db(|db| Ok( + db.query_row("SELECT count(*) FROM ops", [], |r| r.get::<_, u64>(0))? + )) + .await + .unwrap(), + 0 + ); + assert!(upgraded.collectible(10).await.unwrap()); + } + + #[tokio::test] + async fn resolved_profile_completes_the_pre_authentication_placeholder() { + let tmp = tempfile::tempdir().unwrap(); + let journal = SpanJournal::open(&tmp.path().join("auth.sqlite")) + .await + .unwrap(); + let mut named = route("profile"); + named.auth.profile = Some("work".into()); + journal.register(&named).await.unwrap(); + journal.begin(10, true).await.unwrap(); + journal + .append_live(10, &mut 0, &mut vec![op("one")]) + .await + .unwrap(); + journal.commit(10).await.unwrap(); + let mut resolved = named.clone(); + resolved.auth.profile_id = Some("stable-id".into()); + let mut cursor = journal.subscribe(&resolved, 0).await.unwrap(); + assert_eq!(journal.routes().await.unwrap().len(), 1); + cursor + .next_shared() + .await + .unwrap() + .unwrap() + .into_completed() + .await + .ok() + .unwrap(); + assert_eq!( + journal + .with_db(|db| Ok( + db.query_row("SELECT count(*) FROM ops", [], |r| r.get::<_, u64>(0))? + )) + .await + .unwrap(), + 0 + ); + } + + #[tokio::test] + async fn recovery_routes_at_different_heads_share_the_same_decode_window() { + let tmp = tempfile::tempdir().unwrap(); + let journal = SpanJournal::open(&tmp.path().join("shared-recovery.sqlite")) + .await + .unwrap(); + journal.begin(10, true).await.unwrap(); + journal + .append_live(10, &mut 0, &mut vec![op("one"), op("two"), op("three")]) + .await + .unwrap(); + let mut a = journal.subscribe(&route("a"), 0).await.unwrap(); + let mut b = journal.subscribe(&route("b"), 0).await.unwrap(); + b.read = 1; + let first = a.next_shared().await.unwrap().unwrap(); + let second = b.next_shared().await.unwrap().unwrap(); + assert_eq!(first.value().len(), 3); + assert_eq!(second.value().len(), 2); + assert!(std::ptr::eq( + first.value()[1].as_ref(), + second.value()[0].as_ref() + )); + assert_eq!(journal.payload_reads.load(Ordering::Relaxed), 3); + } + + #[tokio::test] + async fn all_route_acknowledgements_reclaim_payloads_and_preserve_watermarks() { + let tmp = tempfile::tempdir().unwrap(); + let path = tmp.path().join("cleanup.sqlite"); + let journal = SpanJournal::open(&path).await.unwrap(); + let mut a = journal.subscribe(&route("a"), 0).await.unwrap(); + let mut b = journal.subscribe(&route("b"), 0).await.unwrap(); + journal.begin(10, true).await.unwrap(); + journal + .append_live(10, &mut 0, &mut vec![op("one")]) + .await + .unwrap(); + journal.commit(10).await.unwrap(); + let first = a.next_shared().await.unwrap().unwrap(); + let second = b.next_shared().await.unwrap().unwrap(); + first.into_completed().await.ok().unwrap(); + assert_eq!( + journal + .with_db(|db| Ok( + db.query_row("SELECT count(*) FROM ops", [], |r| r.get::<_, u64>(0))? + )) + .await + .unwrap(), + 1 + ); + second.into_completed().await.ok().unwrap(); + assert_eq!( + journal + .with_db(|db| Ok( + db.query_row("SELECT count(*) FROM ops", [], |r| r.get::<_, u64>(0))? + )) + .await + .unwrap(), + 0 + ); + assert_eq!(journal.tail().await.unwrap(), 1); + assert_eq!(journal.origins().await.unwrap()["one"].bt_version, "test"); + journal.close().await.unwrap(); + let restored = SpanJournal::open(&path).await.unwrap(); + assert_eq!(restored.tail().await.unwrap(), 1); + assert_eq!(restored.translated_through().await.unwrap(), 10); + let mut late = restored.subscribe(&route("new"), 0).await.unwrap(); + assert!(late.next_shared().await.unwrap().is_none()); + restored.begin(20, true).await.unwrap(); + restored + .append_live(20, &mut 0, &mut vec![op("two")]) + .await + .unwrap(); + assert_eq!(restored.tail().await.unwrap(), 2); + assert!(late.next_shared().await.unwrap().is_some()); + } + + #[tokio::test] + async fn explicit_flush_interrupts_live_batch_collection() { + let tmp = tempfile::tempdir().unwrap(); + let journal = SpanJournal::open(&tmp.path().join("flush.sqlite")) + .await + .unwrap(); + let mut cursor = journal.subscribe(&route("a"), 0).await.unwrap(); + let progress = Progress { + completed: cursor.progress.subscribe(), + published: cursor.changes.clone(), + drain: cursor.drain.clone(), + }; + journal.begin(10, true).await.unwrap(); + journal + .append_live(10, &mut 0, &mut vec![op("one")]) + .await + .unwrap(); + let mut receiving = Box::pin(cursor.next_shared()); + assert!( + tokio::time::timeout(std::time::Duration::from_millis(20), &mut receiving) + .await + .is_err(), + "a sparse batch should wait for more operations" + ); + let error = Mutex::new(None); + tokio::time::timeout(std::time::Duration::from_secs(1), async { + let ((), result) = tokio::join!( + async { + receiving + .await + .unwrap() + .unwrap() + .into_completed() + .await + .ok() + .unwrap(); + }, + progress.through(1, &error) + ); + result.unwrap(); + }) + .await + .expect("explicit flush must bypass the 30-second window"); + assert_eq!(journal.payload_reads.load(Ordering::Relaxed), 0); + } + + #[tokio::test] + async fn live_routes_share_original_payload_without_sqlite_reads() { + let tmp = tempfile::tempdir().unwrap(); + let journal = SpanJournal::open(&tmp.path().join("live.sqlite")) + .await + .unwrap(); + let mut first = journal.subscribe(&route("first"), 0).await.unwrap(); + let mut second = journal.subscribe(&route("second"), 0).await.unwrap(); + let payload = "x".repeat(BYTE_BUDGET * 2); + let pointer = payload.as_ptr(); + let mut ops = vec![SpanOp::Insert(SpanRow { + input: Some(serde_json::Value::String(payload)), + ..Default::default() + })]; + journal.begin(10, true).await.unwrap(); + journal.append_live(10, &mut 0, &mut ops).await.unwrap(); + assert!(ops.is_empty()); + // Live delivery is possible before the native event has finished. + let a = first.next_shared().await.unwrap().unwrap(); + let b = second.next_shared().await.unwrap().unwrap(); + assert!(std::ptr::eq(a.value()[0].as_ref(), b.value()[0].as_ref())); + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = a.value()[0].as_ref(); + assert_eq!( + row.input.as_ref().unwrap().as_str().unwrap().as_ptr(), + pointer + ); + assert_eq!(journal.payload_reads.load(Ordering::Relaxed), 0); + let weak = a.value()[0].downgrade(); + drop(a.into_completed().await.ok().unwrap()); + drop(b.into_completed().await.ok().unwrap()); + assert!(weak.upgrade().is_none()); + } + + #[tokio::test] + async fn recovery_decodes_once_for_concurrent_routes_and_keeps_only_weak_references() { + let tmp = tempfile::tempdir().unwrap(); + let journal = SpanJournal::open(&tmp.path().join("recovery.sqlite")) + .await + .unwrap(); + journal.begin(10, true).await.unwrap(); + journal + .append_live(10, &mut 0, &mut vec![op("recover")]) + .await + .unwrap(); + let mut first = journal.subscribe(&route("first"), 0).await.unwrap(); + let mut second = journal.subscribe(&route("second"), 0).await.unwrap(); + let a = first.next_shared().await.unwrap().unwrap(); + let b = second.next_shared().await.unwrap().unwrap(); + assert!(std::ptr::eq(a.value()[0].as_ref(), b.value()[0].as_ref())); + assert_eq!(journal.payload_reads.load(Ordering::Relaxed), 1); + let weak = a.value()[0].downgrade(); + drop(a); + drop(b); + assert!(weak.upgrade().is_none()); + } + + #[tokio::test] + async fn partial_event_restart_preserves_published_operations_and_route_heads() { + let tmp = tempfile::tempdir().unwrap(); + let path = tmp.path().join("partial.sqlite"); + let journal = SpanJournal::open(&path).await.unwrap(); + let mut first = journal.subscribe(&route("first"), 0).await.unwrap(); + journal.begin(10, true).await.unwrap(); + journal + .append_live(10, &mut 0, &mut vec![op("one")]) + .await + .unwrap(); + first + .next_shared() + .await + .unwrap() + .unwrap() + .into_completed() + .await + .ok() + .unwrap(); + assert_eq!(journal.translated_through().await.unwrap(), 0); + journal.close().await.unwrap(); + let journal = SpanJournal::open(&path).await.unwrap(); + journal.begin(10, true).await.unwrap(); + let mut ordinal = 0; + journal + .append_live(10, &mut ordinal, &mut vec![op("one"), op("two")]) + .await + .unwrap(); + journal.commit(10).await.unwrap(); + let mut restored = journal.subscribe(&route("first"), 0).await.unwrap(); + let delivery = restored.next_shared().await.unwrap().unwrap(); + assert_eq!(delivery.value().len(), 1); + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = delivery.value()[0].as_ref(); + assert_eq!(row.name, "two"); + assert_eq!(journal.tail().await.unwrap(), 2); + } + + #[tokio::test] + async fn paused_route_recovers_without_blocking_a_healthy_live_route() { + let tmp = tempfile::tempdir().unwrap(); + let journal = SpanJournal::open(&tmp.path().join("lag.sqlite")) + .await + .unwrap(); + let mut fast = journal.subscribe(&route("fast"), 0).await.unwrap(); + let mut slow = journal.subscribe(&route("slow"), 0).await.unwrap(); + slow.live.control().pause(); + journal.begin(10, true).await.unwrap(); + let mut ordinal = 0; + for index in 0..3 { + journal + .append_live( + 10, + &mut ordinal, + &mut vec![SpanOp::Insert(SpanRow { + name: index.to_string(), + input: Some(serde_json::json!("x".repeat(BYTE_BUDGET))), + ..Default::default() + })], + ) + .await + .unwrap(); + fast.next_shared() + .await + .unwrap() + .unwrap() + .into_completed() + .await + .ok() + .unwrap(); + } + assert_eq!(journal.payload_reads.load(Ordering::Relaxed), 0); + journal.commit(10).await.unwrap(); + for _ in 0..3 { + let delivery = slow.next_shared().await.unwrap().unwrap(); + assert_eq!( + delivery.value().len(), + 1, + "an oversized operation travels alone" + ); + delivery.into_completed().await.ok().unwrap(); + } + slow.rewind().await.unwrap(); + assert_eq!(slow.completed_position(), 3); + assert_eq!(journal.payload_reads.load(Ordering::Relaxed), 3); + } + #[tokio::test] async fn root_candidates_commit_and_release_atomically() { let tmp = tempfile::tempdir().unwrap(); @@ -1050,7 +1874,7 @@ mod tests { } #[tokio::test] - async fn invisible_chunks_empty_commit_and_restart_preserve_the_boundary() { + async fn legacy_uncommitted_output_empty_commit_and_restart_preserve_the_boundary() { let tmp = tempfile::tempdir().unwrap(); let path = tmp.path().join("spans.sqlite"); let journal = SpanJournal::open(&path).await.unwrap(); diff --git a/bt-daemon/src/pipeline/live.rs b/bt-daemon/src/pipeline/live.rs new file mode 100644 index 00000000..a9b23e42 --- /dev/null +++ b/bt-daemon/src/pipeline/live.rs @@ -0,0 +1,117 @@ +//! Bounded, ephemeral fan-out. SQLite is the recovery path, never the live queue. +use crate::translate::CowSpanOp; +use parking_lot::Mutex; +use std::sync::{Arc, Weak}; +use tokio::sync::{mpsc, watch, OwnedSemaphorePermit, Semaphore}; + +pub(crate) type SharedOp = CowSpanOp; +pub(crate) const BYTE_BUDGET: usize = 256 * 1024; +pub(crate) const WINDOW: std::time::Duration = std::time::Duration::from_secs(30); + +pub(crate) struct Record { + pub position: u64, + pub op: SharedOp, + pub bytes: usize, +} +impl Clone for Record { + fn clone(&self) -> Self { + Self { + position: self.position, + op: self.op.clone(), + bytes: self.bytes, + } + } +} +struct Queued { + record: Record, + _permit: OwnedSemaphorePermit, +} +struct Route { + tx: mpsc::Sender, + budget: Arc, + paused: watch::Sender, +} +#[derive(Default)] +pub(crate) struct Multiplexer { + routes: Mutex>>, +} +pub(crate) struct Receiver { + route: Arc, + rx: mpsc::Receiver, +} +impl Multiplexer { + pub fn subscribe(&self) -> Receiver { + let (tx, rx) = mpsc::channel(32); + let route = Arc::new(Route { + tx, + budget: Arc::new(Semaphore::new(BYTE_BUDGET)), + paused: watch::channel(false).0, + }); + self.routes.lock().push(Arc::downgrade(&route)); + Receiver { route, rx } + } + pub async fn publish(&self, record: Record) { + let routes = { + let mut routes = self.routes.lock(); + let active = routes.iter().filter_map(Weak::upgrade).collect::>(); + routes.retain(|route| route.strong_count() > 0); + active + }; + for route in routes { + let mut paused = route.paused.subscribe(); + if *paused.borrow() { + continue; + } + // Healthy routes apply byte-bounded backpressure. Failed/abandoned + // routes explicitly pause their live feed and recover from disk. + // An oversized operation travels alone. + let permits = record.bytes.clamp(1, BYTE_BUDGET) as u32; + let permit = tokio::select! { + permit=route.budget.clone().acquire_many_owned(permits)=>match permit { Ok(permit)=>permit, Err(_)=>continue }, + _=paused.changed()=>continue, + }; + if *paused.borrow() { + continue; + } + tokio::select! { + _=route.tx.send(Queued { record:record.clone(), _permit:permit })=>{}, + _=paused.changed()=>{}, + } + } + } +} +#[derive(Clone)] +pub(crate) struct Control(Weak); +impl Control { + pub fn pause(&self) { + if let Some(route) = self.0.upgrade() { + route.paused.send_replace(true); + } + } +} +impl Receiver { + pub fn control(&self) -> Control { + Control(Arc::downgrade(&self.route)) + } + pub fn resume(&self) { + self.route.paused.send_replace(false); + } + + pub fn try_recv(&mut self) -> Option { + self.rx.try_recv().ok().map(|queued| queued.record) + } + pub async fn recv(&mut self) -> Option { + self.rx.recv().await.map(|queued| queued.record) + } + pub fn clear(&mut self) { + while self.rx.try_recv().is_ok() {} + } +} +impl Drop for Receiver { + fn drop(&mut self) { + // Only weak route references live in the multiplexer; idle journals do + // not retain payloads or abandoned route channels. + self.route.paused.send_replace(true); + self.route.budget.close(); + } +} diff --git a/bt-daemon/src/pipeline/mod.rs b/bt-daemon/src/pipeline/mod.rs index d2a9a7af..7a830105 100644 --- a/bt-daemon/src/pipeline/mod.rs +++ b/bt-daemon/src/pipeline/mod.rs @@ -1,6 +1,7 @@ //! Typed stage contracts and completion-preserving channel messages. pub(crate) mod delivery; +pub(crate) mod live; pub(crate) mod plugins; pub(crate) mod prepare; pub(crate) mod roots; diff --git a/bt-daemon/src/pipeline/prepare.rs b/bt-daemon/src/pipeline/prepare.rs index a29e4f7e..4148bfc0 100644 --- a/bt-daemon/src/pipeline/prepare.rs +++ b/bt-daemon/src/pipeline/prepare.rs @@ -18,6 +18,35 @@ impl Prepare { } } + pub fn apply_shared(&mut self, ops: &mut [super::live::SharedOp], config: &SessionConfig) { + let (parent, root) = config.attached_span_ids(); + for op in ops { + let insert = matches!(op.as_ref(), SpanOp::Insert(_)); + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = op.as_ref(); + let is_root = self.roots.contains(&row.span_id) + || (row.span_type == SpanType::Task + && row.parent_span_ids.is_empty() + && row.span_id == row.root_span_id); + let subagent = row.metadata.as_ref().is_some_and(|metadata| { + metadata.get("is_subagent").and_then(|v| v.as_bool()) == Some(true) + || (self.source == "codex" && metadata.get("agent_id").is_some()) + }); + if is_root { + self.roots.insert(row.span_id.clone()); + } + let needs_overlay = parent.is_some() + || root.is_some() + || (insert + && (is_root || subagent) + && (!config.tags.is_empty() + || config.additional_metadata.is_some() + || (self.source == "codex" && config.project_name().is_some()))); + if needs_overlay { + self.apply(std::slice::from_mut(op.make_mut()), config); + } + } + } + pub fn apply(&mut self, ops: &mut [SpanOp], config: &SessionConfig) { let (parent, root) = config.attached_span_ids(); let root = root.or_else(|| parent.clone()); diff --git a/bt-daemon/src/pipeline/roots.rs b/bt-daemon/src/pipeline/roots.rs index a0c9f41e..3de5929b 100644 --- a/bt-daemon/src/pipeline/roots.rs +++ b/bt-daemon/src/pipeline/roots.rs @@ -18,17 +18,13 @@ impl Roots { } } -#[async_trait::async_trait] -impl Stage for Roots { - type Input = Vec; - type Outcome = Vec; - - async fn process(&mut self, ops: &Vec) -> Vec { +impl Roots { + pub fn process_owned(&mut self, ops: Vec) -> Vec { let mut ready = Vec::new(); for op in ops { - let (SpanOp::Insert(row) | SpanOp::Merge(row)) = op; + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = &op; if let Some(root) = self.pending.get_mut(&row.span_id) { - root.push(op.clone()); + root.push(op); continue; } for parent in &row.parent_span_ids { @@ -44,19 +40,28 @@ impl Stage for Roots { } } } - if matches!(op, SpanOp::Insert(_)) + if matches!(&op, SpanOp::Insert(_)) && row.metadata.as_ref().and_then(|v| v.get("_bt_defer_root")) == Some(&serde_json::Value::Bool(true)) { - self.pending.insert(row.span_id.clone(), vec![op.clone()]); + self.pending.insert(row.span_id.clone(), vec![op]); } else { - ready.push(op.clone()); + ready.push(op); } } ready } } +#[async_trait::async_trait] +impl Stage for Roots { + type Input = Vec; + type Outcome = Vec; + async fn process(&mut self, ops: &Vec) -> Vec { + self.process_owned(ops.clone()) + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/bt-daemon/src/pipeline/routing.rs b/bt-daemon/src/pipeline/routing.rs index 6c8ea0d9..48a84c71 100644 --- a/bt-daemon/src/pipeline/routing.rs +++ b/bt-daemon/src/pipeline/routing.rs @@ -33,6 +33,7 @@ impl Completion for Forward { pub(crate) struct RoutingSink { tx: mpsc::Sender>, + task: tokio::task::JoinHandle<()>, config: Option, pending: bool, permalink: Option, @@ -47,7 +48,7 @@ impl RoutingSink { last_error: Arc>>, ) -> Self { let (tx, mut rx) = mpsc::channel::>(16); - tokio::spawn(async move { + let task = tokio::spawn(async move { let mut plugins = Plugins::new(&source, &session_id); let mut prepare = Prepare::new(&source); let mut submission = Submission::new(sink); @@ -56,14 +57,20 @@ impl RoutingSink { let (result, state) = match &request.operation { Operation::Emit(ops) => { let mut ops = ops.clone(); - prepare.apply(&mut ops, &request.config); + prepare.apply_shared(&mut ops, &request.config); plugins.configure(&request.config.span_plugins); let mut prepared = Vec::with_capacity(ops.len()); let mut failure = None; let mut report = false; for op in &ops { - match plugins.process(op).await { - PluginOutcome::Prepared(op) => prepared.push(*op), + if request.config.span_plugins.is_empty() { + prepared.push(op.clone()); + continue; + } + match plugins.process(op.as_ref()).await { + PluginOutcome::Prepared(op) => { + prepared.push(super::live::SharedOp::new(*op)) + } PluginOutcome::Filtered => {} PluginOutcome::Failed(error) => { report |= error.failure.newly_seen; @@ -171,12 +178,27 @@ impl RoutingSink { }); Self { tx, + task, config: None, pending: false, permalink: None, } } + pub async fn shutdown(self) { + let Self { tx, mut task, .. } = self; + drop(tx); + // Join the sink owner so completed shutdown also releases ledger files. + // A stalled backend cannot keep daemon shutdown waiting indefinitely. + if tokio::time::timeout(std::time::Duration::from_secs(2), &mut task) + .await + .is_err() + { + task.abort(); + let _ = task.await; + } + } + async fn request(&mut self, operation: Operation) -> anyhow::Result { let config = self .config @@ -208,6 +230,15 @@ impl Sink for RoutingSink { self.config = Some(config.clone()); } async fn emit(&mut self, ops: &[SpanOp]) -> anyhow::Result { + self.emit_shared( + &ops.iter() + .cloned() + .map(super::live::SharedOp::new) + .collect::>(), + ) + .await + } + async fn emit_shared(&mut self, ops: &[super::live::SharedOp]) -> anyhow::Result { self.request(Operation::Emit(ops.to_vec())).await } async fn flush(&mut self) -> anyhow::Result<()> { @@ -242,6 +273,36 @@ mod tests { } } + #[tokio::test] + async fn shutdown_releases_the_owned_sink_before_returning() { + struct Owned(std::sync::Arc); + impl Drop for Owned { + fn drop(&mut self) { + self.0.store(true, Ordering::SeqCst); + } + } + #[async_trait::async_trait] + impl Sink for Owned { + async fn emit(&mut self, _: &[SpanOp]) -> anyhow::Result { + Ok(0) + } + async fn flush(&mut self) -> anyhow::Result<()> { + Ok(()) + } + } + let dropped = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let tmp = tempfile::tempdir().unwrap(); + let sink = RoutingSink::spawn( + Box::new(Owned(dropped.clone())), + "debug".into(), + "owned".into(), + tmp.path().into(), + Arc::new(Mutex::new(None)), + ); + sink.shutdown().await; + assert!(dropped.load(Ordering::SeqCst)); + } + #[tokio::test] async fn partial_plugin_failure_preserves_count_and_reports_only_once() { let tmp = tempfile::tempdir().unwrap(); diff --git a/bt-daemon/src/pipeline/submission.rs b/bt-daemon/src/pipeline/submission.rs index 5252163f..634cb26f 100644 --- a/bt-daemon/src/pipeline/submission.rs +++ b/bt-daemon/src/pipeline/submission.rs @@ -1,9 +1,9 @@ //! The route-local sink adapter is one stage in the routing container. use super::{recovery::RetryCondition, Stage}; -use crate::{sink::Sink, translate::SpanOp, wire::SessionConfig}; +use crate::{sink::Sink, wire::SessionConfig}; pub(crate) enum Operation { - Emit(Vec), + Emit(Vec), Flush, } @@ -104,7 +104,7 @@ impl Stage for Submission { let result = if ops.is_empty() { Ok(0) } else { - self.sink.emit(ops).await.map_err(Failure::classify) + self.sink.emit_shared(ops).await.map_err(Failure::classify) }; Outcome::Emitted { result, diff --git a/bt-daemon/src/pipeline/translation.rs b/bt-daemon/src/pipeline/translation.rs index 042bc6d9..779667b9 100644 --- a/bt-daemon/src/pipeline/translation.rs +++ b/bt-daemon/src/pipeline/translation.rs @@ -8,7 +8,7 @@ use super::{ }; use crate::{ journal::{ControlKind, JournalReader, JournalRecord, JournalWriter, PipelineControl}, - translate::{AgentTranslator, OriginSnapshot, Registry, SessionCtx, SpanOp}, + translate::{AgentTranslator, OriginSnapshot, Registry, SessionCtx, SpanOp, WeakOrigins}, wire::{SessionConfig, SessionRoute}, }; use parking_lot::Mutex; @@ -16,7 +16,7 @@ use std::{ collections::{HashMap, HashSet}, path::PathBuf, sync::{ - atomic::{AtomicU64, AtomicUsize, Ordering}, + atomic::{AtomicU64, Ordering}, Arc, }, }; @@ -49,13 +49,19 @@ impl Progress { } } +#[derive(Default)] +struct RouteLifetimes { + active: usize, + draining: usize, +} + pub(crate) struct Translation { tx: mpsc::Sender, task: tokio::sync::Mutex>>, journal: SpanJournal, pub error: Arc>>, pub recovery: Arc>>, - pub routes: AtomicUsize, + routes: Mutex, pub progress: Arc, } @@ -102,7 +108,7 @@ impl Translation { translator: None, reader: None, observers: Vec::new(), - origins: HashMap::new(), + origins: WeakOrigins::default(), roots: Roots::default(), capture: None, pending_snapshots: HashSet::new(), @@ -120,7 +126,7 @@ impl Translation { journal, error, recovery, - routes: AtomicUsize::new(0), + routes: Mutex::new(RouteLifetimes::default()), progress, })) } @@ -135,12 +141,30 @@ impl Translation { } } pub async fn observe(&self, observer: Observation) -> anyhow::Result<()> { - self.routes.fetch_add(1, Ordering::Relaxed); + self.routes.lock().active += 1; self.tx .send(Command::Observe(Box::new(observer))) .await .map_err(|_| anyhow::anyhow!("translation actor is gone")) } + pub fn begin_route_shutdown(&self) -> bool { + let mut routes = self.routes.lock(); + routes.active -= 1; + routes.draining += 1; + routes.active == 0 + } + pub async fn finish_route_shutdown(&self) { + let finished = { + let mut routes = self.routes.lock(); + routes.draining -= 1; + routes.active == 0 && routes.draining == 0 + }; + // The last route to finish owns closing storage. Merely starting + // shutdown on every route does not mean their acknowledgements are saved. + if finished { + self.shutdown().await; + } + } pub async fn unobserve(&self, key: String) { let _ = self.tx.send(Command::Unobserve(key)).await; } @@ -201,6 +225,7 @@ enum State { } struct Publication { input: u64, + ordinal: u64, observation: bool, capture: Option, ops: Option>, @@ -218,7 +243,7 @@ struct Producer { translator: Option>, reader: Option, observers: Vec, - origins: HashMap, + origins: WeakOrigins, roots: Roots, capture: Option, pending_snapshots: HashSet, @@ -442,24 +467,7 @@ impl Producer { for op in ops { let (SpanOp::Insert(row) | SpanOp::Merge(row)) = op; if let Some(origin) = &row.origin { - self.origins - .entry(row.span_id.clone()) - .or_insert_with(|| origin.clone()); - } - } - } - // Load creation provenance so an upgrade cannot restamp old spans. - let mut after = 0; - loop { - let rows = self.journal.read_after(after, 128).await?; - if rows.is_empty() { - break; - } - for (position, op) in rows { - after = position; - let (SpanOp::Insert(row) | SpanOp::Merge(row)) = op; - if let Some(origin) = row.origin { - self.origins.entry(row.span_id).or_insert(origin); + self.origins.insert(row.span_id.clone(), origin); } } } @@ -511,6 +519,9 @@ impl Producer { let (first, capture) = match entry.record { JournalRecord::Event(event) => { let mut env = crate::journal::envelope_from_redacted(event); + if let Some(route) = &env.route { + self.journal.register(route).await?; + } if crate::translate::canonical_source_name(&env.source) != self.options.source { self.position = input; continue; @@ -580,6 +591,7 @@ impl Producer { }; self.publication = Some(Publication { input, + ordinal: 0, observation, capture, ops: Some(first), @@ -617,19 +629,39 @@ impl Producer { while !pending.complete { if let Some(ops) = pending.ops.as_mut() { if !pending.prepared { + let missing = ops + .iter() + .filter_map(|op| { + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = op; + self.origins + .get(&row.span_id) + .is_none() + .then(|| row.span_id.clone()) + }) + .collect(); + let mut saved = self.journal.origins_for(missing).await?; for op in ops.iter_mut() { let insert = matches!(op, SpanOp::Insert(_)); let (SpanOp::Insert(row) | SpanOp::Merge(row)) = op; - if insert { - self.origins.entry(row.span_id.clone()).or_insert_with(|| { - pending.capture.clone().unwrap_or_else(|| OriginSnapshot { - plugin_version: None, - source_version: None, - bt_version: self.options.version.clone(), + let origin = self + .origins + .get(&row.span_id) + .or_else(|| saved.remove(&row.span_id)) + .or_else(|| { + insert.then(|| { + Arc::new(pending.capture.clone().unwrap_or_else(|| { + OriginSnapshot { + plugin_version: None, + source_version: None, + bt_version: self.options.version.clone(), + } + })) }) }); + if let Some(origin) = &origin { + self.origins.insert(row.span_id.clone(), origin); } - row.origin = self.origins.get(&row.span_id).cloned(); + row.origin = origin; } self.phase = Boundary::Journal; if pending.observation { @@ -644,12 +676,14 @@ impl Producer { } } if publish { - *ops = self.roots.process(ops).await; + *ops = self.roots.process_owned(std::mem::take(ops)); } pending.prepared = true; } if publish { - self.journal.append(pending.input, ops).await?; + self.journal + .append_live(pending.input, &mut pending.ordinal, ops) + .await?; } pending.ops = None; } @@ -699,16 +733,8 @@ impl Producer { ) .await?; } - let mut after = 0; - loop { - let rows = self.journal.read_after(after, 128).await?; - let Some(&(through, _)) = rows.last() else { - break; - }; - after = through; - let ops = rows.into_iter().map(|(_, op)| op).collect::>(); - Self::observe(&self.options, &mut self.pending_snapshots, &ops, observer).await?; - } + let ops = self.journal.observations().await?; + Self::observe(&self.options, &mut self.pending_snapshots, &ops, observer).await?; Ok(()) } async fn observe( @@ -717,7 +743,10 @@ impl Producer { ops: &[SpanOp], observer: &Observation, ) -> anyhow::Result<()> { - let mut ops = ops.to_vec(); + let mut ops = ops + .iter() + .map(super::journal::metadata_only) + .collect::>(); Prepare::new(&options.source).apply(&mut ops, &observer.config); if options .correlation @@ -745,6 +774,35 @@ mod tests { use super::*; use crate::{pipeline::delivery::InputPort, translate::SpanRow}; + #[tokio::test] + async fn shared_storage_stays_open_until_the_last_route_finishes_draining() { + let tmp = tempfile::tempdir().unwrap(); + let event_path = tmp.path().join("events.ndjson"); + let translation = Translation::spawn(TranslationOptions { + source: "debug".into(), + session_id: "draining".into(), + namespace: "draining".into(), + version: "test".into(), + data_dir: tmp.path().into(), + events: Arc::new(tokio::sync::Mutex::new( + JournalWriter::open_path(&event_path).await.unwrap(), + )), + event_path, + registry: Arc::new(Registry::default_agents()), + correlation: Arc::new(crate::correlation::CorrelationRegistry::default()), + correlation_changed: Arc::new(Notify::new()), + }) + .await + .unwrap(); + translation.routes.lock().active = 2; + assert!(!translation.begin_route_shutdown()); + assert!(translation.begin_route_shutdown()); + translation.finish_route_shutdown().await; + assert_eq!(translation.journal.tail().await.unwrap(), 0); + translation.finish_route_shutdown().await; + assert!(translation.journal.tail().await.is_err()); + } + #[tokio::test] async fn recreated_observer_restores_a_staged_session_root() { let tmp = tempfile::tempdir().unwrap(); @@ -921,7 +979,7 @@ mod tests { translator: Some(Box::new(CountedTranslation(drains.clone()))), reader: None, observers: Vec::new(), - origins: HashMap::new(), + origins: WeakOrigins::default(), roots: Roots::default(), capture: None, pending_snapshots: HashSet::new(), @@ -932,6 +990,7 @@ mod tests { phase: Boundary::Journal, publication: Some(Publication { input: 10, + ordinal: 0, observation: true, capture: None, ops: Some(vec![ @@ -957,7 +1016,7 @@ mod tests { db.execute_batch("DROP TRIGGER fail_append; CREATE TRIGGER fail_commit BEFORE UPDATE OF committed ON batches BEGIN SELECT RAISE(FAIL,'commit'); END;").unwrap(); assert!(producer.publish().await.is_err()); assert_eq!(drains.load(Ordering::Relaxed), 1); - assert_eq!(journal.tail().await.unwrap(), 0); + assert_eq!(journal.tail().await.unwrap(), 1); db.execute_batch("DROP TRIGGER fail_commit").unwrap(); producer.publish().await.unwrap(); assert_eq!(drains.load(Ordering::Relaxed), 1); diff --git a/bt-daemon/src/server/maintenance.rs b/bt-daemon/src/server/maintenance.rs index bf914b6a..61e9d45c 100644 --- a/bt-daemon/src/server/maintenance.rs +++ b/bt-daemon/src/server/maintenance.rs @@ -338,11 +338,17 @@ pub(super) async fn drain_all(daemon: &Arc) { .iter() .map(|(key, session)| (key.clone(), session.clone())) .collect(); + let mut draining = tokio::task::JoinSet::new(); for (key, session) in sessions { let target = targets .get(&crate::ids::session_namespace(&key.source, &key.session_id)) .copied(); - session.shutdown_through(target).await; + draining.spawn(async move { session.shutdown_through(target).await }); + } + while let Some(result) = draining.join_next().await { + if let Err(error) = result { + tracing::warn!(%error, "route shutdown failed"); + } } let sources: Vec<_> = daemon.translations.lock().values().cloned().collect(); for source in sources { diff --git a/bt-daemon/src/sink/braintrust.rs b/bt-daemon/src/sink/braintrust.rs index 998047bf..445cbbfd 100644 --- a/bt-daemon/src/sink/braintrust.rs +++ b/bt-daemon/src/sink/braintrust.rs @@ -739,11 +739,14 @@ mod tests { let row = SpanRow { span_id: "root".into(), root_span_id: "root".into(), - origin: Some(OriginSnapshot { - plugin_version: None, - bt_version: "birth-bt".into(), - source_version: None, - }), + origin: Some( + OriginSnapshot { + plugin_version: None, + bt_version: "birth-bt".into(), + source_version: None, + } + .into(), + ), ..Default::default() }; let handle = client diff --git a/bt-daemon/src/sink/mod.rs b/bt-daemon/src/sink/mod.rs index 220e494c..61b7f695 100644 --- a/bt-daemon/src/sink/mod.rs +++ b/bt-daemon/src/sink/mod.rs @@ -38,6 +38,16 @@ pub trait Sink: Send { /// Emit span ops. Returns the number of rows written, for status counters. async fn emit(&mut self, ops: &[SpanOp]) -> anyhow::Result; + /// Shared canonical operations remain shared until a route changes them. + /// The default adapter borrows one operation without copying its payload. + async fn emit_shared(&mut self, ops: &[crate::translate::CowSpanOp]) -> anyhow::Result { + let mut emitted = 0; + for op in ops { + emitted += self.emit(std::slice::from_ref(op.as_ref())).await?; + } + Ok(emitted) + } + /// Deliver everything buffered (bounded by the caller's flush timeout). async fn flush(&mut self) -> anyhow::Result<()>; diff --git a/bt-daemon/src/translate/mod.rs b/bt-daemon/src/translate/mod.rs index 412b2717..d3c87f9a 100644 --- a/bt-daemon/src/translate/mod.rs +++ b/bt-daemon/src/translate/mod.rs @@ -7,6 +7,9 @@ //! whole pipeline be exercised with a debug sink and makes translators unit- //! testable without any network. +mod shared; +pub use shared::CowSpanOp; + mod antigravity; mod claude; mod codex; @@ -19,7 +22,9 @@ mod opencode; mod pi; pub(crate) use claude::successful_subagent_handback_message; pub(crate) use pi::request_config as pi_request_config; +mod origins; mod recent; +pub(crate) use origins::WeakOrigins; mod tool; mod turn_lineage; pub use turn_lineage::TURN_SPAN_ID_KEY; @@ -103,9 +108,10 @@ pub struct SpanRow { /// Translator-owned marker for turn creation, not other task spans. #[serde(skip)] pub is_turn: bool, - /// Delivery-owned creation snapshot, present on each coding-agent span. + /// Creation snapshot shared by live operations; origin caches hold only + /// weak references so completed spans do not keep their snapshots resident. #[serde(default, skip_serializing_if = "Option::is_none")] - pub origin: Option, + pub origin: Option>, } /// A span operation. `Insert` creates (or replaces) a row; `Merge` updates an diff --git a/bt-daemon/src/translate/origins.rs b/bt-daemon/src/translate/origins.rs new file mode 100644 index 00000000..a7ad840c --- /dev/null +++ b/bt-daemon/src/translate/origins.rs @@ -0,0 +1,57 @@ +//! Provenance is owned by live operations, never by the lookup cache. +use super::OriginSnapshot; +use std::{ + collections::HashMap, + sync::{Arc, Weak}, +}; + +#[derive(Default)] +pub(crate) struct WeakOrigins(HashMap>); + +impl WeakOrigins { + pub fn get(&mut self, span: &str) -> Option> { + let origin = self.0.get(span).and_then(Weak::upgrade); + if origin.is_none() { + self.0.remove(span); + } + origin + } + + pub fn insert(&mut self, span: String, origin: &Arc) { + // Weak values alone still leak keys. Sweep dead entries as new work + // arrives, keeping the index proportional to currently live owners. + self.0.retain(|_, origin| origin.strong_count() > 0); + self.0.insert(span, Arc::downgrade(origin)); + } + + pub fn remove(&mut self, span: &str) { + self.0.remove(span); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn cache_releases_origins_and_dead_keys_but_preserves_live_operations() { + let mut cache = WeakOrigins::default(); + let live = Arc::new(OriginSnapshot { + plugin_version: None, + bt_version: "original".into(), + source_version: None, + }); + cache.insert("live".into(), &live); + for i in 0..10_000 { + let origin = Arc::new((*live).clone()); + let weak = Arc::downgrade(&origin); + cache.insert(i.to_string(), &origin); + drop(origin); + assert!(weak.upgrade().is_none()); + } + assert!(cache.0.len() <= 2); + assert!(Arc::ptr_eq(&cache.get("live").unwrap(), &live)); + drop(live); + assert!(cache.get("live").is_none()); + } +} diff --git a/bt-daemon/src/translate/recent.rs b/bt-daemon/src/translate/recent.rs index a6a3e567..557280be 100644 --- a/bt-daemon/src/translate/recent.rs +++ b/bt-daemon/src/translate/recent.rs @@ -9,7 +9,7 @@ use std::borrow::Borrow; use std::collections::{HashMap, HashSet, VecDeque}; use std::hash::Hash; -pub(super) const RECENT_ID_CAPACITY: usize = 4_096; +pub(super) const RECENT_ID_CAPACITY: usize = 1_024; pub(super) struct RecentSet { values: HashSet, diff --git a/bt-daemon/src/translate/shared.rs b/bt-daemon/src/translate/shared.rs new file mode 100644 index 00000000..8573facf --- /dev/null +++ b/bt-daemon/src/translate/shared.rs @@ -0,0 +1,72 @@ +//! Owned copy-on-write span operations for asynchronous route fan-out. +use super::SpanOp; +use std::sync::{Arc, Weak}; + +/// A route's independently mutable reference to a canonical span operation. +/// Clones share immutable data; mutation copies only when another owner exists. +#[derive(Debug, Clone)] +pub struct CowSpanOp(Arc); +impl CowSpanOp { + pub fn new(op: SpanOp) -> Self { + Self(Arc::new(op)) + } + pub fn make_mut(&mut self) -> &mut SpanOp { + Arc::make_mut(&mut self.0) + } + pub(crate) fn downgrade(&self) -> Weak { + Arc::downgrade(&self.0) + } + pub(crate) fn from_arc(op: Arc) -> Self { + Self(op) + } +} +impl From for CowSpanOp { + fn from(op: SpanOp) -> Self { + Self::new(op) + } +} +impl AsRef for CowSpanOp { + fn as_ref(&self) -> &SpanOp { + &self.0 + } +} +impl std::ops::Deref for CowSpanOp { + type Target = SpanOp; + fn deref(&self) -> &SpanOp { + &self.0 + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::translate::SpanRow; + #[tokio::test] + async fn concurrent_routes_share_until_one_mutates() { + let canonical = CowSpanOp::new(SpanOp::Insert(SpanRow { + input: Some(serde_json::json!({"history":"original"})), + ..Default::default() + })); + let mut first = canonical.clone(); + let second = canonical.clone(); + assert!(Arc::ptr_eq(&first.0, &second.0)); + let (first, second) = tokio::join!( + async move { + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = first.make_mut(); + row.input = Some(serde_json::json!({"history":"changed"})); + first + }, + async move { second } + ); + assert!(!Arc::ptr_eq(&first.0, &second.0)); + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = second.as_ref(); + assert_eq!(row.input.as_ref().unwrap()["history"], "original"); + assert!(Arc::ptr_eq(&canonical.0, &second.0)); + let weak = first.downgrade(); + drop(first); + assert!( + weak.upgrade().is_none(), + "no route cache pins a finished payload" + ); + } +} diff --git a/bt-daemon/tests/braintrust_sink.rs b/bt-daemon/tests/braintrust_sink.rs index 98c24903..99962754 100644 --- a/bt-daemon/tests/braintrust_sink.rs +++ b/bt-daemon/tests/braintrust_sink.rs @@ -63,11 +63,14 @@ fn row( late_merge_key: None, turn_span_id: None, tags: None, - origin: Some(OriginSnapshot { - plugin_version: None, - bt_version: "test".into(), - source_version: None, - }), + origin: Some( + OriginSnapshot { + plugin_version: None, + bt_version: "test".into(), + source_version: None, + } + .into(), + ), is_turn: false, } } @@ -423,7 +426,7 @@ async fn span_updates_preserve_name_linkage_and_origin() { 1, None, ); - initial.origin = Some(origin.clone()); + initial.origin = Some(origin.clone().into()); let mut sink = factory.create(&span_id, source).unwrap(); sink.configure(&session_config(&base)); sink.emit(&[SpanOp::Insert(initial)]).await.unwrap(); @@ -434,7 +437,7 @@ async fn span_updates_preserve_name_linkage_and_origin() { root_span_id: "trace-root".into(), parent_span_ids: vec!["turn-parent".into()], span_type, - origin: Some(origin), + origin: Some(origin.into()), ..Default::default() }; for (status, end_ms) in [("open", None), ("closed", Some(2)), ("late", None)] { @@ -615,11 +618,14 @@ async fn exported_parent_preserves_object_root_and_propagated_event() { 1, Some(2), ); - root.origin = Some(OriginSnapshot { - plugin_version: Some("attached-plugin".into()), - bt_version: "attached-bt".into(), - source_version: Some("attached-codex".into()), - }); + root.origin = Some( + OriginSnapshot { + plugin_version: Some("attached-plugin".into()), + bt_version: "attached-bt".into(), + source_version: Some("attached-codex".into()), + } + .into(), + ); sink.emit(&[SpanOp::Insert(root)]).await.unwrap(); sink.flush().await.unwrap(); diff --git a/bt-daemon/tests/cursor_pipeline.rs b/bt-daemon/tests/cursor_pipeline.rs index 8e45b9ec..a2e7c432 100644 --- a/bt-daemon/tests/cursor_pipeline.rs +++ b/bt-daemon/tests/cursor_pipeline.rs @@ -22,6 +22,24 @@ use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex}; use std::time::Duration; +/// Register a dormant route before any output exists, so completed peers cannot +/// reclaim the recovery payloads this fixture deliberately leaves unresolved. +fn seed_dormant_route(data: &Path, source: &str, session: &str, route: &SessionRoute) { + let journal = bt_daemon::source_journal_path(data, source, session); + let stem = journal.file_stem().unwrap().to_str().unwrap(); + let id = stem.rsplit("--").next().unwrap(); + std::fs::create_dir_all(data.join("pipeline")).unwrap(); + let db = + rusqlite::Connection::open(data.join("pipeline").join(format!("{id}.sqlite"))).unwrap(); + // The daemon upgrades this preparatory schema on first use. + db.execute_batch("CREATE TABLE heads(route TEXT PRIMARY KEY,position INTEGER NOT NULL DEFAULT 0,failure TEXT)").unwrap(); + db.execute( + "INSERT INTO heads(route,position) VALUES (?,0)", + [serde_json::to_string(route).unwrap()], + ) + .unwrap(); +} + /// Smoke-test Cursor's portable direct command through the native Windows /// shell. This checks executable lookup, stdin forwarding, and its JSON /// response without relying on a POSIX-only launcher. @@ -31,6 +49,7 @@ use std::time::Duration; reason = "Exercises a foreground command exactly as Cursor invokes Windows hooks." )] #[test] + fn cursor_direct_hook_command_receives_native_stdin_on_windows() { use std::io::Write; use std::process::{Command, Stdio}; @@ -548,6 +567,15 @@ async fn cursor_mirror_generations_and_capture_bounds_survive_deleted_native_rec let socket = endpoint(tmp.path()); let native = tmp.path().join("native.jsonl"); let daemon = start(&data, &socket).await; + let recovery_route = SessionRoute { + destination: Some(TraceDestination::ProjectLogs { + project_id: None, + project_name: Some("cursor-tests".into()), + }), + additional_metadata: Some(json!({"test_route":"recovery"})), + ..Default::default() + }; + seed_dormant_route(&data, "cursor", "cursor-recovery", &recovery_route); send( &socket, "sessionStart", @@ -684,7 +712,7 @@ async fn cursor_mirror_generations_and_capture_bounds_survive_deleted_native_rec .unwrap() .contains("cursor-test-secret")); - // Simulate unresolved sink delivery while retaining both durable source stores. + // Recover the dormant route as well as the primary route from the retained payloads. span_journal(&data) .execute("UPDATE heads SET position=0,failure=NULL", []) .unwrap(); diff --git a/bt-daemon/tests/distributed_tracing.rs b/bt-daemon/tests/distributed_tracing.rs index 6a666a5d..32abad49 100644 --- a/bt-daemon/tests/distributed_tracing.rs +++ b/bt-daemon/tests/distributed_tracing.rs @@ -820,6 +820,7 @@ async fn indistinguishable_concurrent_tools_attach_to_the_known_session() { } flush("ambiguous-child", &socket).await; + flush("ambiguous-parent", &socket).await; assert_attached_to_session(recording.as_ref(), "ambiguous-parent", "ambiguous-child"); shutdown_daemon(&socket).await.unwrap(); @@ -1411,6 +1412,7 @@ async fn completed_tools_are_not_candidates_for_later_children() { ) .await; flush(&child_session, &socket).await; + flush(&parent_session, &socket).await; assert_attached_to_session(recording.as_ref(), &parent_session, &child_session); } @@ -1514,6 +1516,7 @@ async fn child_output_does_not_guess_between_concurrent_tools() { .await; flush("output-child", &socket).await; + flush("output-parent", &socket).await; assert_attached_to_session(recording.as_ref(), "output-parent", "output-child"); shutdown_daemon(&socket).await.unwrap(); @@ -1837,6 +1840,7 @@ async fn open_tool_prevents_idle_retirement_during_a_long_child_spawn() { ) .await; flush("idle-child", &socket).await; + flush("idle-parent", &socket).await; assert_pair_linked( recording.as_ref(), "idle-parent", @@ -1906,6 +1910,7 @@ async fn retired_parent_still_parents_a_child_from_its_next_tool() { flush("thinking-child", &socket).await; // The revived parent pipeline replays its root without redelivering it, // so only the new tool row is in its latest recording. + flush("thinking-parent", &socket).await; let parent_rows = inserted_rows(&recording.session("thinking-parent")); let tool = parent_rows .iter() @@ -2015,6 +2020,7 @@ async fn concurrent_ambiguous_children_do_not_deadlock_or_cross_link() { } for (_, session, _) in &child_cases { flush(session, &socket).await; + flush("deadlock-parent", &socket).await; assert_attached_to_session(recording.as_ref(), "deadlock-parent", session); } @@ -2090,6 +2096,7 @@ async fn one_tool_can_fan_out_to_concurrent_children_of_every_agent() { ) .await; flush(session, &socket).await; + flush(parent_session, &socket).await; assert_pair_linked( recording.as_ref(), parent_session, @@ -2521,6 +2528,7 @@ async fn standalone_parentage_is_not_reconsidered_on_later_session_start() { .await; flush("late-tool-child", &socket).await; // A later tool must not move a child already attached to its session. + flush("late-tool-parent", &socket).await; assert_attached_to_session(recording.as_ref(), "late-tool-parent", "late-tool-child"); shutdown_daemon(&socket).await.unwrap(); @@ -2582,6 +2590,7 @@ async fn unconfirmed_child_is_visible_and_delivered_without_more_events() { assert_standalone(recording.as_ref(), "bounded-child"); // No parent completion event has been sent: optional correlation must // not wait for the spawning tool to finish before delivering the root. + flush("bounded-parent", &socket).await; let parent = recording.session("bounded-parent"); let parent_rows = inserted_rows(&parent); let tool = parent_rows @@ -2724,6 +2733,7 @@ async fn explicit_parent_and_child_route_settings_win_over_ancestry() { assert_eq!(root.parent_span_ids, vec!["external-parent".to_owned()]); assert_eq!(root.root_span_id, "external-root"); } else { + flush("bounded-parent", &socket).await; assert_pair_linked( recording.as_ref(), "bounded-parent", @@ -2768,13 +2778,9 @@ async fn one_native_parent_session_selects_the_childs_delivery_route() { } let route = events[0].route.clone().unwrap(); forward_all(&mut events, &socket, &host).await; - run_status(StatusArgs { - socket: Some(socket.clone()), - session_id: Some(parent_session.into()), - }) - .await - .unwrap() - .unwrap(); + // Status only settles translation; wait for asynchronous route delivery + // before looking up the sink created for this route. + flush(parent_session, &socket).await; // Keep both sinks: the fixture's session lookup retains only the latest. parents.push((route, recording.session(parent_session))); } @@ -3128,7 +3134,7 @@ async fn pending_child( } async fn wait_for_root(recording: &RecordingSinkFactory, session: &str) { - tokio::time::timeout(Duration::from_secs(10), async { + tokio::time::timeout(Duration::from_secs(45), async { loop { let record = recording.sessions.lock().unwrap().get(session).cloned(); if record.is_some_and(|record| { @@ -3148,7 +3154,7 @@ async fn wait_for_root(recording: &RecordingSinkFactory, session: &str) { } }) .await - .unwrap_or_else(|_| panic!("root for {session} was not delivered within 10 seconds")); + .unwrap_or_else(|_| panic!("root for {session} was not delivered within 45 seconds")); } fn assert_standalone(recording: &RecordingSinkFactory, session: &str) { diff --git a/bt-daemon/tests/pipeline.rs b/bt-daemon/tests/pipeline.rs index 314cfd56..58f1bd5c 100644 --- a/bt-daemon/tests/pipeline.rs +++ b/bt-daemon/tests/pipeline.rs @@ -184,6 +184,24 @@ impl Sink for RouteRecordingSink { } } +/// Register a dormant route before any output exists, so completed peers cannot +/// reclaim the recovery payloads this fixture deliberately leaves unresolved. +fn seed_dormant_route(data: &Path, source: &str, session: &str, route: &SessionRoute) { + let journal = bt_daemon::source_journal_path(data, source, session); + let stem = journal.file_stem().unwrap().to_str().unwrap(); + let id = stem.rsplit("--").next().unwrap(); + std::fs::create_dir_all(data.join("pipeline")).unwrap(); + let db = + rusqlite::Connection::open(data.join("pipeline").join(format!("{id}.sqlite"))).unwrap(); + // The daemon upgrades this preparatory schema on first use. + db.execute_batch("CREATE TABLE heads(route TEXT PRIMARY KEY,position INTEGER NOT NULL DEFAULT 0,failure TEXT)").unwrap(); + db.execute( + "INSERT INTO heads(route,position) VALUES (?,0)", + [serde_json::to_string(route).unwrap()], + ) + .unwrap(); +} + fn dummy_host() -> HostInfo { // The daemon is started in-process, so the client never spawns; serve_argv // is unused but must be non-empty. @@ -385,9 +403,13 @@ fn pipeline_database(data_dir: &Path) -> rusqlite::Connection { rusqlite::Connection::open(path).unwrap() } fn head_and_tail(data_dir: &Path) -> (u64, u64) { - pipeline_database(data_dir).query_row( - "SELECT position,(SELECT coalesce(max(position),0) FROM ops JOIN batches USING(input) WHERE committed=1) FROM heads", - [],|row| Ok((row.get(0)?,row.get(1)?))).unwrap() + pipeline_database(data_dir) + .query_row( + "SELECT position,(SELECT tail FROM publication WHERE id=1) FROM heads", + [], + |row| Ok((row.get(0)?, row.get(1)?)), + ) + .unwrap() } async fn wait_for(endpoint: &Path) { @@ -1046,7 +1068,7 @@ async fn a_route_backfills_observations_captured_by_another_route_once() { } #[tokio::test] -async fn a_new_destination_receives_history_once_then_uses_its_own_checkpoint() { +async fn a_new_destination_starts_at_the_reclaimed_prefix_and_uses_its_own_checkpoint() { let tmp = tempfile::tempdir().unwrap(); let data_dir = tmp.path().join("data"); let socket = test_endpoint(tmp.path()); @@ -1074,8 +1096,8 @@ async fn a_new_destination_receives_history_once_then_uses_its_own_checkpoint() first.await.unwrap(); // A managed run or newly configured global route can select a second - // destination for the same native session. It gets the complete source - // history even though that history was originally captured via work. + // destination for the same native session. Fully reported payloads have + // been reclaimed, so it starts at the retained prefix and receives new work. let provider = Arc::new(TestAuthProvider { calls: Mutex::new(Vec::new()), fail: false, @@ -1106,8 +1128,8 @@ async fn a_new_destination_receives_history_once_then_uses_its_own_checkpoint() delivered[0] .emitted .load(std::sync::atomic::Ordering::Relaxed), - 3, - "personal receives work's root and SessionStart plus its own Stop" + 1, + "personal receives its Stop; fulfilled historical payloads were reclaimed" ); assert_eq!( delivered[0].org.lock().unwrap().clone().as_deref(), @@ -1187,7 +1209,15 @@ async fn auth_resolution_failure_is_reported_without_exposing_credentials() { source_journal_path(&data_dir, "debug", "login-needed").exists(), "capture must remain durable even when background auth fails" ); - assert_eq!(provider.calls.lock().unwrap().len(), 1); + { + let calls = provider.calls.lock().unwrap(); + // Background recovery may retry before status returns on a busy runner. + assert!(!calls.is_empty()); + assert!(calls.iter().all(|(selection, _)| { + selection.profile.as_deref() == Some("missing") + && selection.org_name.as_deref() == Some("missing-org") + })); + } shutdown(&socket).await; handle.await.unwrap(); @@ -1521,6 +1551,12 @@ async fn opencode_lifecycle_flushes_without_an_explicit_client_flush() { // Observe the work flush separately, so it cannot satisfy the terminal // event's assertion below. + assert!( + flush_session(&session_id, &socket, 5_000) + .await + .unwrap() + .flushed + ); tokio::time::timeout(Duration::from_secs(5), async { loop { if flushes @@ -2999,7 +3035,7 @@ async fn invoke_grok_hook( #[cfg(all(feature = "cli", unix))] async fn wait_for_grok_translation(socket: &Path, session_id: &str) { let mut last_sessions = Vec::new(); - for _ in 0..200 { + for _ in 0..4500 { let status = run_status(StatusArgs { socket: Some(socket.to_path_buf()), session_id: Some(session_id.to_string()), @@ -3633,6 +3669,9 @@ async fn startup_restores_a_durable_head_absent_from_captured_routes() { }); let (data_dir, socket, handle, _tmp) = start_routed_daemon(provider.clone()).await; let env = envelope("durable-subscription", "SessionStart", 1); + let mut dormant = env.route.clone().unwrap(); + dormant.tags = vec!["dormant".into()]; + seed_dormant_route(&data_dir, &env.source, &env.session_id, &dormant); forward_envelope(&env, &socket, &dummy_host(), false) .await .unwrap(); @@ -3644,21 +3683,6 @@ async fn startup_restores_a_durable_head_absent_from_captured_routes() { ); shutdown(&socket).await; handle.await.unwrap(); - let path = std::fs::read_dir(data_dir.join("pipeline")) - .unwrap() - .filter_map(Result::ok) - .map(|entry| entry.path()) - .find(|path| path.extension().is_some_and(|ext| ext == "sqlite")) - .unwrap(); - let mut dormant = env.route.unwrap(); - dormant.tags = vec!["dormant".into()]; - let db = rusqlite::Connection::open(&path).unwrap(); - db.execute( - "INSERT INTO heads(route,position) VALUES(?,0)", - [serde_json::to_string(&dormant).unwrap()], - ) - .unwrap(); - drop(db); let recording = Arc::new(RouteRecordingSinkFactory::default()); let handle = start_daemon_at_with( data_dir.clone(), @@ -3805,6 +3829,18 @@ async fn clean_shutdown_delivers_the_shared_retirement_tail_to_every_route() { HashSet::from(["left".into(), "right".into()]) ); let (head, tail) = head_and_tail(&data_dir); + let heads = pipeline_database(&data_dir) + .prepare("SELECT position FROM heads") + .unwrap() + .query_map([], |r| r.get::<_, u64>(0)) + .unwrap() + .collect::, _>>() + .unwrap(); + assert_eq!(heads.len(), 2); + assert!( + heads.iter().all(|position| *position == tail), + "route heads: {heads:?}; tail: {tail}" + ); assert_eq!( head, tail, "every route durably completed the retirement output" @@ -3845,7 +3881,7 @@ async fn initial_span_storage_failure_reports_journal_and_recovers_without_nativ .await .unwrap(); std::fs::remove_file(&storage).unwrap(); - tokio::time::timeout(Duration::from_secs(5), async { + tokio::time::timeout(Duration::from_secs(45), async { loop { if std::fs::read_to_string(data.join("spans/storage-open.ndjson")) .is_ok_and(|rows| rows.lines().count() >= 2) @@ -3945,7 +3981,7 @@ async fn changed_settings_resume_the_original_rejected_route_through_ingress() { project_id: Some("test-project".into()), project_name: None, }); - tokio::time::timeout(Duration::from_secs(5), async { + tokio::time::timeout(Duration::from_secs(45), async { forward_envelope(&event, &socket, &dummy_host(), false) .await .unwrap(); @@ -3981,7 +4017,7 @@ async fn changed_settings_resume_the_original_rejected_route_through_ingress() { forward_envelope(&event, &socket, &dummy_host(), false) .await .unwrap(); - tokio::time::timeout(Duration::from_secs(5), async { + tokio::time::timeout(Duration::from_secs(45), async { loop { let status = run_status(StatusArgs { socket: Some(socket.clone()), @@ -4051,7 +4087,7 @@ async fn changed_plugin_chain_resumes_the_original_held_route() { forward_envelope(&event, &socket, &dummy_host(), false) .await .unwrap(); - tokio::time::timeout(Duration::from_secs(5), async { + tokio::time::timeout(Duration::from_secs(45), async { loop { let status = run_status(StatusArgs { socket: Some(socket.clone()), @@ -4078,7 +4114,7 @@ async fn changed_plugin_chain_resumes_the_original_held_route() { forward_envelope(&event, &socket, &dummy_host(), false) .await .unwrap(); - tokio::time::timeout(Duration::from_secs(5), async { + tokio::time::timeout(Duration::from_secs(45), async { loop { let status = run_status(StatusArgs { socket: Some(socket.clone()), @@ -4205,7 +4241,7 @@ async fn storage_failure_after_retirement_is_reported_and_recovers_without_nativ .any(|session| session.queued > 0 && session.last_error.is_some())); std::fs::remove_file(data.join("pipeline")).unwrap(); std::fs::rename(data.join("saved-pipeline"), data.join("pipeline")).unwrap(); - tokio::time::timeout(Duration::from_secs(5), async { + tokio::time::timeout(Duration::from_secs(45), async { loop { if std::fs::read_to_string(data.join("spans/retired-failure.ndjson")) .is_ok_and(|rows| rows.contains("\"name\":\"Stop\""))