diff --git a/bt-daemon/src/journal.rs b/bt-daemon/src/journal.rs index c301022b..809a3316 100644 --- a/bt-daemon/src/journal.rs +++ b/bt-daemon/src/journal.rs @@ -333,6 +333,26 @@ pub struct JournalReader { } impl JournalReader { + pub(crate) fn position(&self) -> u64 { + self.position + } + + /// Rebuild decoder context before continuing at a saved record boundary. + pub(crate) async fn resume(path: &Path, position: u64, through: u64) -> anyhow::Result { + anyhow::ensure!(position <= through, "invalid recovery range"); + let mut reader = Self::open(path, through) + .await? + .ok_or_else(|| anyhow::anyhow!("recovery journal is missing"))?; + while reader.position < position { + anyhow::ensure!( + reader.next_record().await?.is_some(), + "recovery journal ended before its boundary" + ); + } + anyhow::ensure!(reader.position == position, "invalid recovery boundary"); + Ok(reader) + } + /// Open a journal for streaming, reading at most `through` bytes. /// /// That bound is what keeps replay from consuming the very event that @@ -385,49 +405,19 @@ impl JournalReader { } self.position += read as u64; self.line_no += 1; - let line = std::str::from_utf8(&bytes) - .map_err(|error| { - anyhow::anyhow!("journal {}:{}: {error}", self.path.display(), self.line_no) - })? - .trim(); - if line.is_empty() { - continue; + let path = self.path.clone(); + let line_no = self.line_no; + let position = self.position; + let mut pi_context = std::mem::take(&mut self.pi_context); + let (pi_context, record) = tokio::task::spawn_blocking(move || { + let record = decode_record(&bytes, &path, line_no, position, &mut pi_context); + (pi_context, record) + }) + .await?; + self.pi_context = pi_context; + if let Some(record) = record? { + return Ok(Some(record)); } - let value: serde_json::Value = serde_json::from_str(line).map_err(|error| { - anyhow::anyhow!("journal {}:{}: {error}", self.path.display(), self.line_no) - })?; - let record = if value - .get("_bt_record_type") - .and_then(serde_json::Value::as_str) - == Some("delivery_checkpoint") - { - let checkpoint: DeliveryCheckpointRecord = - serde_json::from_value(value).map_err(|error| { - anyhow::anyhow!("journal {}:{}: {error}", self.path.display(), self.line_no) - })?; - JournalRecord::DeliveryCheckpoint { - route: checkpoint.route, - through: checkpoint.through, - } - } else if value - .get("_bt_record_type") - .and_then(serde_json::Value::as_str) - == Some("pipeline_control") - { - JournalRecord::Control(serde_json::from_value(value)?) - } else { - let mut event = serde_json::from_value(value).map_err(|error| { - anyhow::anyhow!("journal {}:{}: {error}", self.path.display(), self.line_no) - })?; - expand_pi_payload(&mut event, &mut self.pi_context).map_err(|error| { - anyhow::anyhow!("journal {}:{}: {error}", self.path.display(), self.line_no) - })?; - JournalRecord::Event(event) - }; - return Ok(Some(JournalRecordEntry { - record, - through: self.position, - })); } } @@ -461,6 +451,51 @@ impl JournalReader { } } +fn decode_record( + bytes: &[u8], + path: &Path, + line_no: usize, + position: u64, + pi_context: &mut Vec, +) -> anyhow::Result> { + let line = std::str::from_utf8(bytes) + .map_err(|error| anyhow::anyhow!("journal {}:{}: {error}", path.display(), line_no))? + .trim(); + if line.is_empty() { + return Ok(None); + } + let value: serde_json::Value = serde_json::from_str(line) + .map_err(|error| anyhow::anyhow!("journal {}:{}: {error}", path.display(), line_no))?; + let record = if value + .get("_bt_record_type") + .and_then(serde_json::Value::as_str) + == Some("delivery_checkpoint") + { + let checkpoint: DeliveryCheckpointRecord = serde_json::from_value(value) + .map_err(|error| anyhow::anyhow!("journal {}:{}: {error}", path.display(), line_no))?; + JournalRecord::DeliveryCheckpoint { + route: checkpoint.route, + through: checkpoint.through, + } + } else if value + .get("_bt_record_type") + .and_then(serde_json::Value::as_str) + == Some("pipeline_control") + { + JournalRecord::Control(serde_json::from_value(value)?) + } else { + let mut event = serde_json::from_value(value) + .map_err(|error| anyhow::anyhow!("journal {}:{}: {error}", path.display(), line_no))?; + expand_pi_payload(&mut event, pi_context) + .map_err(|error| anyhow::anyhow!("journal {}:{}: {error}", path.display(), line_no))?; + JournalRecord::Event(event) + }; + Ok(Some(JournalRecordEntry { + record, + through: position, + })) +} + const PI_MESSAGES_DELTA: &str = "_bt_messages_delta"; fn compact_pi_payload(event: &mut RedactedEnvelope, previous: &mut Vec) { diff --git a/bt-daemon/src/pipeline/input.rs b/bt-daemon/src/pipeline/input.rs new file mode 100644 index 00000000..63d5586c --- /dev/null +++ b/bt-daemon/src/pipeline/input.rs @@ -0,0 +1,121 @@ +//! Stage input shared by direct capture and recovery. Storage supplies only +//! byte positions; original live payloads arrive from capture. +pub use crate::journal::PipelineControl; +use crate::wire::Envelope; +use parking_lot::Mutex; +use std::{collections::VecDeque, sync::Arc}; + +const RECORD_BUDGET: usize = 64; +const BYTE_BUDGET: u64 = 8 * 1024 * 1024; + +/// Durable byte range returned by the journal append. +#[derive(Clone)] +pub(crate) struct SourcePosition { + pub before: u64, + pub through: u64, +} + +pub(crate) enum SourceEvent { + Event(Box), + Control(PipelineControl), + LegacyDeliveryCheckpoint, +} + +pub(crate) struct SourceRecord { + pub event: SourceEvent, + pub position: SourcePosition, +} + +impl SourceRecord { + fn bytes(&self) -> u64 { + self.position.through.saturating_sub(self.position.before) + } +} + +#[derive(Default)] +struct Buffer { + records: VecDeque, + bytes: u64, +} + +#[derive(Clone, Default)] +pub(crate) struct SourceInput(Arc>); + +impl SourceInput { + pub(crate) fn push_event(&self, event: Envelope, position: SourcePosition) { + self.push(SourceRecord { + event: SourceEvent::Event(Box::new(event)), + position, + }); + } + pub(crate) fn is_empty(&self) -> bool { + self.0.lock().records.is_empty() + } + pub(crate) fn push(&self, record: SourceRecord) { + let mut buffer = self.0.lock(); + let bytes = record.bytes(); + // An oversized event travels alone, just as it does on the SpanOp feed. + while !buffer.records.is_empty() + && (buffer.records.len() >= RECORD_BUDGET + || buffer.bytes.saturating_add(bytes) > BYTE_BUDGET) + { + if let Some(evicted) = buffer.records.pop_front() { + buffer.bytes -= evicted.bytes(); + } + } + buffer.bytes = buffer.bytes.saturating_add(bytes); + buffer.records.push_back(record); + } + + pub(crate) fn take(&self, position: u64, through: u64) -> Option { + let mut buffer = self.0.lock(); + // Recovery may already have consumed records whose live copies remained. + while buffer + .records + .front() + .is_some_and(|record| record.position.through <= position) + { + if let Some(consumed) = buffer.records.pop_front() { + buffer.bytes -= consumed.bytes(); + } + } + let first = buffer.records.front()?; + if first.position.before != position || first.position.through > through { + return None; + } + let record = buffer.records.pop_front()?; + buffer.bytes -= record.bytes(); + Some(record) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn oversized_events_flow_alone_and_eviction_exposes_a_recovery_gap() { + let input = SourceInput::default(); + let event: Envelope = serde_json::from_value(json!({"source":"debug","session_id":"direct","event":"native","ts_ms":1,"payload":{"history":"x".repeat(9 * 1024 * 1024)}})).unwrap(); + let stride = 9 * 1024 * 1024; + let position = |before| SourcePosition { + before, + through: before + stride, + }; + input.push_event(event.clone(), position(0)); + assert!(input.take(0, 0).is_none()); + let SourceEvent::Event(actual) = input.take(0, stride).unwrap().event else { + panic!("expected original event"); + }; + assert_eq!(actual.payload, event.payload); + input.push_event(event.clone(), position(stride)); + input.push_event(event, position(2 * stride)); + assert!( + input.take(stride, 3 * stride).is_none(), + "eviction requires recovery of the gap" + ); + assert!(input.take(2 * stride, 3 * stride).is_some()); + assert!(input.is_empty()); + } +} diff --git a/bt-daemon/src/pipeline/journal.rs b/bt-daemon/src/pipeline/journal.rs index 0bbff985..c854ed26 100644 --- a/bt-daemon/src/pipeline/journal.rs +++ b/bt-daemon/src/pipeline/journal.rs @@ -1,4 +1,8 @@ //! Concrete durable channel: one ordered SpanOp log and a head per route. +//! +//! Tracking mutations acquire SQLite write access before reading. Deferred +//! read-to-write upgrades can fail without waiting when another connection +//! writes concurrently, even with a busy timeout. use super::live::{Multiplexer, Receiver, Record, SharedOp, BYTE_BUDGET, WINDOW}; use super::{ delivery::InputPort, @@ -180,13 +184,27 @@ impl SpanJournal { std::fs::create_dir_all(parent)?; let db = Connection::open(path)?; db.busy_timeout(std::time::Duration::from_secs(5))?; + // Concurrent WAL transitions can return BUSY without invoking + // SQLite's busy handler. Retry initialization within the same + // bound, before any schema transaction has started. + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + loop { + match db.execute_batch("PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL; BEGIN IMMEDIATE;") { + Ok(()) => break, + Err(error) if error.sqlite_error_code() == Some(rusqlite::ErrorCode::DatabaseBusy) + && std::time::Instant::now() < deadline => { + std::thread::sleep(std::time::Duration::from_millis(10)); + } + Err(error) => return Err(error.into()), + } + } + // Read the version under the same write transaction as every + // upgrade, so another opener cannot run a stale migration plan. let version: u32 = db.query_row("PRAGMA user_version", [], |row| row.get(0))?; 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( - "BEGIN IMMEDIATE; - CREATE TABLE IF NOT EXISTS batches ( + "CREATE TABLE IF NOT EXISTS batches ( input INTEGER PRIMARY KEY, committed INTEGER NOT NULL DEFAULT 0, observation INTEGER NOT NULL DEFAULT 1); CREATE TABLE IF NOT EXISTS ops ( @@ -205,16 +223,16 @@ impl SpanJournal { )?; } db.execute_batch("ALTER TABLE heads ADD COLUMN binding TEXT")?; - db.execute_batch("PRAGMA user_version=3; COMMIT")?; + db.execute_batch("PRAGMA user_version=3")?; } if version < 4 { - db.execute_batch("BEGIN IMMEDIATE; CREATE TABLE root_candidates (span_id TEXT PRIMARY KEY, value BLOB NOT NULL); PRAGMA user_version=4; COMMIT")?; + db.execute_batch("CREATE TABLE root_candidates (span_id TEXT PRIMARY KEY, value BLOB NOT NULL); PRAGMA user_version=4")?; } 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")?; + db.execute_batch("ALTER TABLE heads ADD COLUMN settings TEXT; ALTER TABLE heads ADD COLUMN settings_through INTEGER NOT NULL DEFAULT 0; PRAGMA user_version=5")?; } if version < 6 { - db.execute_batch("BEGIN IMMEDIATE; + db.execute_batch(" 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); @@ -231,9 +249,10 @@ impl SpanJournal { after = position; } prune(&db)?; - db.execute_batch("PRAGMA user_version=6; COMMIT")?; + db.execute_batch("PRAGMA user_version=6")?; } let tail = db.query_row("SELECT tail FROM publication WHERE id=1", [], |row| row.get(0))?; + db.execute_batch("COMMIT")?; Ok((db, tail)) }) .await??; @@ -294,10 +313,9 @@ impl SpanJournal { .await } - /// 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()?; + let tx = db.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?; let done: Option = tx .query_row( "SELECT committed FROM batches WHERE input=?", @@ -359,7 +377,8 @@ impl SpanJournal { let first = *ordinal; let positions = self .with_db(move |db| { - let tx = db.transaction()?; + let tx = + db.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?; let mut positions = Vec::new(); let (committed, observation): (bool, bool) = tx.query_row( "SELECT committed,observation FROM batches WHERE input=?", @@ -472,7 +491,7 @@ impl SpanJournal { .map(StoredOp::encode) .collect::>>()?; self.with_db(move |db| { - let tx = db.transaction()?; + let tx = db.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?; let done: bool = tx.query_row( "SELECT committed FROM batches WHERE input=?", [input], @@ -545,7 +564,7 @@ impl SpanJournal { ) -> anyhow::Result { let tail = self .with_db(move |db| { - let tx = db.transaction()?; + let tx = db.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?; let previous: u64 = tx.query_row( "SELECT coalesce(max(input),0) FROM batches WHERE committed=1", [], @@ -602,7 +621,7 @@ impl SpanJournal { let key = route.clone(); let position = self .with_db(move |db| { - let tx = db.transaction()?; + let tx = db.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?; // 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. @@ -819,7 +838,7 @@ impl SpanJournal { ) -> anyhow::Result> { let route = route.clone(); self.with_db(move |db| { - let tx = db.transaction()?; + let tx = db.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?; let updates = { let mut query = tx.prepare( "SELECT route,failure,settings FROM heads WHERE failure IS NOT NULL AND settings_through, } +#[derive(Clone)] pub(crate) struct TranslationOptions { pub source: String, pub session_id: String, @@ -73,6 +75,7 @@ pub(crate) struct TranslationOptions { pub data_dir: PathBuf, pub events: Arc>, pub event_path: PathBuf, + pub input: SourceInput, pub registry: Arc, pub correlation: Arc, pub correlation_changed: Arc, @@ -80,45 +83,12 @@ pub(crate) struct TranslationOptions { impl Translation { pub async fn spawn(options: TranslationOptions) -> anyhow::Result> { - let path = options.data_dir.join("pipeline").join(format!( - "{}.sqlite", - crate::ids::session_storage_id(&options.source, &options.session_id) - )); - let journal = SpanJournal::open(&path).await?; - let committed = journal.translated_through().await?; - let progress = Arc::new(Progress { - requested: AtomicU64::new(committed), - committed: AtomicU64::new(committed), - }); - let recovery = Arc::new(Mutex::new(None)); - let error = Arc::new(Mutex::new(None)); + let actor = Producer::open(options).await?; + let journal = actor.journal.clone(); + let progress = actor.progress.clone(); + let error = actor.error.clone(); + let recovery = actor.recovery.clone(); let (tx, rx) = mpsc::channel(64); - let state = match journal.source_failure().await? { - Some(raw) => State::Recovering(serde_json::from_str(&raw).unwrap_or_else(|_| { - Failure::revision(Boundary::Journal, committed + 1, raw, &options.version) - })), - None => State::Processing, - }; - let actor = Producer { - options, - progress: progress.clone(), - journal: journal.clone(), - error: error.clone(), - recovery: recovery.clone(), - translator: None, - reader: None, - observers: Vec::new(), - origins: WeakOrigins::default(), - roots: Roots::default(), - capture: None, - pending_snapshots: HashSet::new(), - bounds: HashMap::new(), - position: 0, - committed, - state, - publication: None, - phase: Boundary::Journal, - }; let task = tokio::spawn(actor.run(rx)); Ok(Arc::new(Self { tx, @@ -256,6 +226,72 @@ struct Producer { } impl Producer { + async fn open(options: TranslationOptions) -> anyhow::Result { + let path = options.data_dir.join("pipeline").join(format!( + "{}.sqlite", + crate::ids::session_storage_id(&options.source, &options.session_id) + )); + let journal = SpanJournal::open(&path).await?; + let committed = journal.translated_through().await?; + let progress = Arc::new(Progress { + requested: AtomicU64::new(committed), + committed: AtomicU64::new(committed), + }); + let recovery = Arc::new(Mutex::new(None)); + let error = Arc::new(Mutex::new(None)); + let state = match journal.source_failure().await? { + Some(raw) => State::Recovering(serde_json::from_str(&raw).unwrap_or_else(|_| { + Failure::revision(Boundary::Journal, committed + 1, raw, &options.version) + })), + None => State::Processing, + }; + Ok(Self { + options, + progress, + journal, + error, + recovery, + translator: None, + reader: None, + observers: Vec::new(), + origins: WeakOrigins::default(), + roots: Roots::default(), + capture: None, + pending_snapshots: HashSet::new(), + bounds: HashMap::new(), + position: 0, + committed, + state, + publication: None, + phase: Boundary::Journal, + }) + } + + /// Translators perform synchronous native transcript I/O and reduction. + /// Keep both replay and live reduction off the runtime serving hook IPC. + async fn translate( + &mut self, + operation: impl FnOnce(&mut dyn AgentTranslator, &SessionCtx) -> anyhow::Result + + Send + + 'static, + ) -> anyhow::Result { + let mut translator = self + .translator + .take() + .ok_or_else(|| anyhow::anyhow!("translator is missing"))?; + let ctx = SessionCtx { + session_id: self.options.session_id.clone(), + config: None, + }; + let (translator, result) = tokio::task::spawn_blocking(move || { + let result = operation(translator.as_mut(), &ctx); + (translator, result) + }) + .await?; + self.translator = Some(translator); + result + } + async fn run(mut self, mut rx: mpsc::Receiver) { let mut recorded = None; let mut record_retry = 0; @@ -397,6 +433,7 @@ impl Producer { } } } + fn failure(&self, error: anyhow::Error, attempt: u32) -> Failure { let position = self.committed.saturating_add(1); let mut failure = if matches!(self.phase, Boundary::Translation) { @@ -428,12 +465,15 @@ impl Producer { matches!(self.state, State::Processing), "translation is recovering" ); - self.append_control(kind).await?; + self.initialize().await?; self.advance(self.progress.requested.load(Ordering::Relaxed)) .await?; + if let Some(through) = self.append_control(kind).await? { + self.advance(through).await?; + } self.journal.tail().await } - async fn append_control(&mut self, kind: ControlKind) -> anyhow::Result<()> { + async fn append_control(&mut self, kind: ControlKind) -> anyhow::Result> { self.phase = Boundary::Journal; let mut bounds = HashMap::new(); for path in self.bounds.keys() { @@ -444,17 +484,27 @@ impl Producer { bounds.insert(path.clone(), through); } let control = PipelineControl { kind, bounds }; - let through = self - .options - .events - .lock() - .await - .append_control(&control) - .await?; + let through = { + let mut writer = self.options.events.lock().await; + // A checkpoint observes dispatched work. If capture is ahead, + // flush the published tail and let queued events arrive normally; + // chasing the file here can reorder parent/child processing. + if writer.position() > self.position { + return Ok(None); + } + let before = writer.position(); + let through = writer.append_control(&control).await?; + let position = SourcePosition { before, through }; + self.options.input.push(SourceRecord { + event: SourceEvent::Control(control), + position, + }); + through + }; self.progress .requested .fetch_max(through, Ordering::Relaxed); - Ok(()) + Ok(Some(through)) } async fn initialize(&mut self) -> anyhow::Result<()> { @@ -471,18 +521,63 @@ impl Producer { } } } + let registry = self.options.registry.clone(); + let source = self.options.source.clone(); + let namespace = self.options.namespace.clone(); self.translator = Some( - self.options - .registry - .create_checked_with_session_namespace( - &self.options.source, - &self.options.namespace, - )?, + tokio::task::spawn_blocking(move || { + registry.create_checked_with_session_namespace(&source, &namespace) + }) + .await??, ); - self.reader = JournalReader::open(&self.options.event_path, self.committed).await?; + self.position = 0; + self.reader = if self.committed > 0 { + JournalReader::open(&self.options.event_path, self.committed).await? + } else { + None + }; Ok(()) } + async fn next_input(&mut self, through: u64) -> anyhow::Result> { + if let Some(record) = self.options.input.take(self.position, through) { + // A future gap resumes from this logical position. The live path + // does not open, seek, read, or pass payloads through a journal reader. + self.reader = None; + return Ok(Some(record)); + } + if self.position >= through { + return Ok(None); + } + if self.reader.is_none() { + self.reader = Some( + JournalReader::resume(&self.options.event_path, self.position, through).await?, + ); + } + let reader = self + .reader + .as_mut() + .ok_or_else(|| anyhow::anyhow!("recovery reader is missing"))?; + reader.extend(through); + let Some(entry) = reader.next_record().await? else { + return Ok(None); + }; + let event = match entry.record { + JournalRecord::Event(event) => { + SourceEvent::Event(Box::new(crate::journal::envelope_from_redacted(event))) + } + JournalRecord::Control(control) => SourceEvent::Control(control), + JournalRecord::DeliveryCheckpoint { .. } => SourceEvent::LegacyDeliveryCheckpoint, + }; + Ok(Some(SourceRecord { + event, + position: SourcePosition { + before: self.position, + through: entry.through, + }, + })) + } + async fn advance(&mut self, through: u64) -> anyhow::Result<()> { anyhow::ensure!( matches!(self.state, State::Processing), @@ -497,28 +592,12 @@ impl Producer { if self.position >= through { return Ok(()); } - let reader = self - .reader - .as_mut() - .ok_or_else(|| anyhow::anyhow!("event journal is missing"))?; - reader.extend(through); self.phase = Boundary::Journal; - while let Some(entry) = self - .reader - .as_mut() - .ok_or_else(|| anyhow::anyhow!("event journal is missing"))? - .next_record() - .await? - { - let input = entry.through; - let observation = !matches!(&entry.record,JournalRecord::Control(control) if matches!(control.kind,ControlKind::Retire)); - let ctx = SessionCtx { - session_id: self.options.session_id.clone(), - config: None, - }; - let (first, capture) = match entry.record { - JournalRecord::Event(event) => { - let mut env = crate::journal::envelope_from_redacted(event); + while let Some(entry) = self.next_input(through).await? { + let input = entry.position.through; + let observation = !matches!(&entry.event,SourceEvent::Control(control) if matches!(control.kind,ControlKind::Retire)); + let (first, capture) = match entry.event { + SourceEvent::Event(mut env) => { if let Some(route) = &env.route { self.journal.register(route).await?; } @@ -552,14 +631,12 @@ impl Producer { self.capture = Some(capture.clone()); self.phase = Boundary::Translation; ( - self.translator - .as_mut() - .ok_or_else(|| anyhow::anyhow!("translator is missing"))? - .handle(&env, &ctx)?, + self.translate(move |translator, ctx| translator.handle(&env, ctx)) + .await?, Some(capture), ) } - JournalRecord::Control(control) => { + SourceEvent::Control(control) => { // Retirement exports defensive terminal updates but does not // close native state when rebuilding for later captured work. if input <= self.committed && matches!(control.kind, ControlKind::Retire) { @@ -567,18 +644,18 @@ impl Producer { continue; } self.phase = Boundary::Translation; - let translator = self - .translator - .as_mut() - .ok_or_else(|| anyhow::anyhow!("translator is missing"))?; - let ops = if !matches!(control.kind, ControlKind::Checkpoint) { - translator.finalize_at(&ctx, &control.bounds)? - } else { - translator.checkpoint_at(&ctx, &control.bounds)? - }; + let ops = self + .translate(move |translator, ctx| { + if !matches!(control.kind, ControlKind::Checkpoint) { + translator.finalize_at(ctx, &control.bounds) + } else { + translator.checkpoint_at(ctx, &control.bounds) + } + }) + .await?; (ops, self.capture.clone()) } - JournalRecord::DeliveryCheckpoint { .. } => { + SourceEvent::LegacyDeliveryCheckpoint => { if input > self.committed { self.journal.begin(input, observation).await?; self.journal.commit(input).await?; @@ -615,10 +692,6 @@ impl Producer { .ok_or_else(|| anyhow::anyhow!("pending publication is missing"))?; let result = async { let publish = pending.input > self.committed; - let ctx = SessionCtx { - session_id: self.options.session_id.clone(), - config: None, - }; if publish && !pending.begun { self.phase = Boundary::Journal; self.journal @@ -689,10 +762,8 @@ impl Producer { } self.phase = Boundary::Translation; pending.ops = self - .translator - .as_mut() - .ok_or_else(|| anyhow::anyhow!("translator is missing"))? - .drain_pending(&ctx)?; + .translate(|translator, ctx| translator.drain_pending(ctx)) + .await?; pending.complete = pending.ops.is_none(); pending.prepared = false; } @@ -770,30 +841,236 @@ impl Producer { } #[cfg(test)] -mod tests { +pub(crate) mod tests { use super::*; use crate::{pipeline::delivery::InputPort, translate::SpanRow}; + mod streams { + use crate as bt_daemon; + include!(concat!( + env!("CARGO_MANIFEST_DIR"), + "/tests/support/translation_streams.rs" + )); + } + + pub(crate) struct SourceFixture { + _dir: tempfile::TempDir, + pub(crate) options: TranslationOptions, + pub(crate) source: Arc, + } + impl SourceFixture { + pub(crate) async fn new(agent: &str) -> Self { + let dir = tempfile::tempdir().unwrap(); + let path = crate::journal::source_journal_path(dir.path(), agent, "checkpoint"); + let options = TranslationOptions { + source: agent.into(), + session_id: "checkpoint".into(), + namespace: crate::ids::session_namespace(agent, "checkpoint"), + version: "test".into(), + data_dir: dir.path().into(), + events: Arc::new(tokio::sync::Mutex::new( + JournalWriter::open_path(&path).await.unwrap(), + )), + event_path: path, + input: SourceInput::default(), + registry: Arc::new(Registry::default_agents()), + correlation: Arc::default(), + correlation_changed: Arc::default(), + }; + let source = Translation::spawn(options.clone()).await.unwrap(); + Self { + _dir: dir, + options, + source, + } + } + pub(crate) async fn write(&self, event: &crate::wire::Envelope, live: bool) -> u64 { + let mut writer = self.options.events.lock().await; + let before = writer.position(); + let through = writer.append(event).await.unwrap(); + let position = SourcePosition { before, through }; + if live { + self.options.input.push_event(event.clone(), position); + } + through + } + pub(crate) async fn process(&self, through: u64) { + self.source.advance(through).await.unwrap(); + self.source.barrier().await.unwrap(); + assert_eq!( + self.source.journal.translated_through().await.unwrap(), + through + ); + } + async fn reopen(&mut self) { + self.source.shutdown().await; + *self.options.events.lock().await = JournalWriter::open_path(&self.options.event_path) + .await + .unwrap(); + self.options.input = SourceInput::default(); + self.source = Translation::spawn(self.options.clone()).await.unwrap(); + } + fn db(&self) -> rusqlite::Connection { + rusqlite::Connection::open(self.options.data_dir.join("pipeline").join(format!( + "{}.sqlite", + crate::ids::session_storage_id(&self.options.source, &self.options.session_id) + ))) + .unwrap() + } + pub(crate) async fn poison(&self, range: std::ops::Range) { + let mut bytes = tokio::fs::read(&self.options.event_path).await.unwrap(); + for byte in &mut bytes[range.start as usize..range.end as usize] { + if *byte != b'\n' { + *byte = b'x'; + } + } + tokio::fs::write(&self.options.event_path, bytes) + .await + .unwrap(); + } + } + + #[tokio::test] + async fn every_agent_recovers_equivalent_spans_at_every_native_boundary() { + let mut covered = Vec::new(); + for (agent, mut events, _files) in streams::native_streams() { + covered.push(agent.clone()); + events.insert(0, events[0].clone()); + for event in &mut events { + event.payload["padding"] = serde_json::json!("é📚history".repeat(1024)); + } + let mut cases = vec![("journal", 0), ("live", 0), ("overflow", 2)]; + cases.extend((1..events.len()).map(|split| ("gap", split))); + cases.extend((1..=events.len()).map(|split| ("cold", split))); + let mut expected = None; + for (mode, split) in cases { + let mut f = SourceFixture::new(&agent).await; + let mut positions = vec![0]; + for (index, event) in events.iter().enumerate() { + let live = + mode == "live" || mode == "overflow" || (mode == "gap" && index != split); + positions.push(f.write(event, live).await); + } + let tail = *positions.last().unwrap(); + match mode { + "live" => { + f.process(positions[1]).await; + f.source.control(ControlKind::Checkpoint).await.unwrap(); + assert_eq!( + f.source.journal.translated_through().await.unwrap(), + positions[1] + ); + assert_eq!(f.options.events.lock().await.position(), tail); + f.poison(0..tail).await; + } + "overflow" => { + assert!(f.options.input.take(0, positions[1]).is_some()); + assert!(f.options.input.take(positions[1], positions[2]).is_some()); + f.poison(positions[2]..tail).await; + } + "gap" => { + // Rebuild decoder context after live input, then return + // to live input without rereading its retained tail. + f.poison(positions[split + 1]..tail).await; + } + _ => { + f.process(positions[split]).await; + f.source.shutdown().await; + f.reopen().await; + } + } + f.process(tail).await; + f.source.control(ControlKind::Finalize).await.unwrap(); + assert!(f.source.journal.translated_through().await.unwrap() > tail); + // StoredOp already includes the private correlation fields; + // compare its complete encoding rather than duplicate its codec. + let db = f.db(); + let mut query = db + .prepare("SELECT value FROM ops ORDER BY position") + .unwrap(); + let rows: Vec> = query + .query_map([], |row| row.get(0)) + .unwrap() + .map(Result::unwrap) + .collect(); + assert!(!rows.is_empty(), "{agent} {mode} produced no observations"); + if let Some(expected) = &expected { + assert_eq!(&rows, expected, "{agent} {mode} split {split}"); + } else { + expected = Some(rows); + } + f.source.shutdown().await; + } + } + covered.sort(); + assert_eq!(covered, Registry::default_agents().sources()); + } + + struct BlockedTranslator { + entered: Arc, + release: std::sync::mpsc::Receiver<()>, + } + impl BlockedTranslator { + fn block(&self) { + self.entered.notify_one(); + self.release + .recv_timeout(std::time::Duration::from_secs(5)) + .unwrap(); + } + } + impl AgentTranslator for BlockedTranslator { + fn handle( + &mut self, + _: &crate::wire::Envelope, + _: &SessionCtx, + ) -> anyhow::Result> { + self.block(); + Ok(Vec::new()) + } + } + + #[tokio::test] + async fn synchronous_replay_does_not_block_capture_or_chase_new_events() { + let f = SourceFixture::new("debug").await; + let event = streams::native_streams().pop().unwrap().1.remove(0); + let through = f.write(&event, false).await; + let mut producer = Producer::open(f.options.clone()).await.unwrap(); + let entered = Arc::new(Notify::new()); + let (release, receiver) = std::sync::mpsc::channel(); + producer.translator = Some(Box::new(BlockedTranslator { + entered: entered.clone(), + release: receiver, + })); + let started = std::time::Instant::now(); + let work = tokio::spawn(async move { + let result = producer.advance(through).await; + (producer, result) + }); + entered.notified().await; + assert!(started.elapsed() < std::time::Duration::from_secs(1)); + assert!( + tokio::time::timeout(std::time::Duration::from_secs(1), f.write(&event, true)) + .await + .unwrap() + > through + ); + assert!( + !work.is_finished(), + "capture must complete before replay is released" + ); + release.send(()).unwrap(); + let (producer, result) = work.await.unwrap(); + result.unwrap(); + assert_eq!( + producer.journal.translated_through().await.unwrap(), + through + ); + f.source.shutdown().await; + } #[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(); + let f = SourceFixture::new("debug").await; + let translation = &f.source; translation.routes.lock().active = 2; assert!(!translation.begin_route_shutdown()); assert!(translation.begin_route_shutdown()); @@ -805,25 +1082,9 @@ mod tests { #[tokio::test] async fn recreated_observer_restores_a_staged_session_root() { - let tmp = tempfile::tempdir().unwrap(); - let event_path = tmp.path().join("events.ndjson"); - let correlation = Arc::new(crate::correlation::CorrelationRegistry::default()); - let translation = Translation::spawn(TranslationOptions { - source: "debug".into(), - session_id: "staged".into(), - namespace: "staged".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: correlation.clone(), - correlation_changed: Arc::new(Notify::new()), - }) - .await - .unwrap(); + let f = SourceFixture::new("debug").await; + let translation = &f.source; + let correlation = &f.options.correlation; let mut roots = Roots::default(); roots .process(&vec![ @@ -913,6 +1174,7 @@ mod tests { JournalWriter::open_path(&event_path).await.unwrap(), )), event_path, + input: SourceInput::default(), registry: Arc::new(Registry::default_agents()), correlation: Arc::new(crate::correlation::CorrelationRegistry::default()), correlation_changed: Arc::new(Notify::new()), @@ -951,64 +1213,37 @@ mod tests { } #[tokio::test] async fn failed_append_and_commit_retain_the_built_chunk() { - let tmp = tempfile::tempdir().unwrap(); - let path = tmp.path().join("spans.sqlite"); - let journal = SpanJournal::open(&path).await.unwrap(); - let db = rusqlite::Connection::open(&path).unwrap(); - let event_path = tmp.path().join("events.ndjson"); + let f = SourceFixture::new("debug").await; + f.source.shutdown().await; + let mut producer = Producer::open(f.options.clone()).await.unwrap(); + let journal = producer.journal.clone(); + let path = f.options.data_dir.as_path().join("pipeline").join(format!( + "{}.sqlite", + crate::ids::session_storage_id("debug", "checkpoint") + )); + let db = rusqlite::Connection::open(path).unwrap(); let drains = Arc::new(AtomicU64::new(0)); - let mut producer = Producer { - options: TranslationOptions { - source: "debug".into(), - session_id: "test".into(), - namespace: "test".into(), - version: "test".into(), - data_dir: tmp.path().into(), - events: Arc::new(tokio::sync::Mutex::new( - JournalWriter::open_path(&event_path).await.unwrap(), - )), - event_path, - registry: Arc::new(Registry::default_agents()), - correlation: Arc::new(crate::correlation::CorrelationRegistry::default()), - correlation_changed: Arc::new(Notify::new()), - }, - progress: Arc::new(Progress::default()), - journal: journal.clone(), - error: Arc::new(Mutex::new(None)), - recovery: Arc::new(Mutex::new(None)), - translator: Some(Box::new(CountedTranslation(drains.clone()))), - reader: None, - observers: Vec::new(), - origins: WeakOrigins::default(), - roots: Roots::default(), + producer.translator = Some(Box::new(CountedTranslation(drains.clone()))); + producer.publication = Some(Publication { + input: 10, + ordinal: 0, + observation: true, capture: None, - pending_snapshots: HashSet::new(), - bounds: HashMap::new(), - position: 0, - committed: 0, - state: State::Processing, - phase: Boundary::Journal, - publication: Some(Publication { - input: 10, - ordinal: 0, - observation: true, - capture: None, - ops: Some(vec![ - SpanOp::Insert(SpanRow { - span_id: "unused-root".into(), - metadata: Some(serde_json::json!({"_bt_defer_root": true})), - ..Default::default() - }), - SpanOp::Insert(SpanRow { - span_id: "built".into(), - ..Default::default() - }), - ]), - begun: false, - prepared: false, - complete: false, - }), - }; + ops: Some(vec![ + SpanOp::Insert(SpanRow { + span_id: "unused-root".into(), + metadata: Some(serde_json::json!({"_bt_defer_root": true})), + ..Default::default() + }), + SpanOp::Insert(SpanRow { + span_id: "built".into(), + ..Default::default() + }), + ]), + begun: false, + prepared: false, + complete: false, + }); db.execute_batch("CREATE TRIGGER fail_append BEFORE INSERT ON ops BEGIN SELECT RAISE(FAIL,'append'); END;").unwrap(); assert!(producer.publish().await.is_err()); assert_eq!(drains.load(Ordering::Relaxed), 0); diff --git a/bt-daemon/src/server/ingress.rs b/bt-daemon/src/server/ingress.rs index d2ae453d..c74c2eb9 100644 --- a/bt-daemon/src/server/ingress.rs +++ b/bt-daemon/src/server/ingress.rs @@ -3,9 +3,148 @@ //! the journals as overflow when the queue is full. use super::*; +use std::collections::VecDeque; +/// Captures remain ordered and bounded. Evicted entries are +/// already durable; their source journal's high-water mark is replayed first. +#[derive(Default)] +pub(super) struct CaptureBuffer { + pub(super) events: VecDeque, + pub(super) evicted: HashMap, +} -pub(super) enum IngressMsg { +#[derive(Clone, Copy, Debug)] +pub(super) struct EvictedRange { + pub(super) from: u64, + pub(super) through: u64, +} + +impl CaptureBuffer { + fn push(&mut self, data_dir: &std::path::Path, event: PendingEvent) { + if self.events.len() == INGRESS_QUEUE_CAPACITY { + if let Some(oldest) = self.events.pop_front() { + let path = journal::source_journal_path( + data_dir, + &oldest.env.source, + &oldest.env.session_id, + ); + self.evicted + .entry(path) + .and_modify(|range| range.through = range.through.max(oldest.journal_through)) + .or_insert(EvictedRange { + from: oldest.replay_through, + through: oldest.journal_through, + }); + } + } + self.events.push_back(event); + } +} + +pub(super) enum CaptureWork { + Evicted(HashMap), Event(Box), + Finished, +} + +impl Daemon { + pub(super) fn next_capture_work(&self) -> CaptureWork { + let mut buffer = self.capture_buffer.lock(); + if !buffer.evicted.is_empty() { + return CaptureWork::Evicted(std::mem::take(&mut buffer.evicted)); + } + if let Some(event) = buffer.events.pop_front() { + return CaptureWork::Event(Box::new(event)); + } + CaptureWork::Finished + } +} + +pub(super) async fn drain_capture_buffer(daemon: &Arc) -> anyhow::Result<()> { + loop { + match daemon.next_capture_work() { + CaptureWork::Evicted(bounds) => replay_evicted(daemon, &bounds).await?, + CaptureWork::Event(event) => { + daemon + .source_input_for(&event.env.source, &event.env.session_id) + .push_event( + event.env.clone(), + crate::pipeline::input::SourcePosition { + before: event.replay_through, + through: event.journal_through, + }, + ); + dispatch_ingress_event(daemon, *event).await; + } + CaptureWork::Finished => return Ok(()), + } + } +} + +/// Read frozen evicted ranges in source order, merging by event timestamp as +/// startup recovery does. Rebuild decoder context before each range. +pub(super) async fn replay_evicted( + daemon: &Arc, + bounds: &HashMap, +) -> anyhow::Result<()> { + let mut ready = Vec::new(); + for (path, range) in bounds { + let mut reader = journal::JournalReader::resume(path, range.from, range.through).await?; + if let Some((timestamp, event)) = + next_evicted_event(daemon, &mut reader, range.through).await? + { + ready.push((timestamp, event, reader, range.through)); + } + } + while let Some(index) = ready + .iter() + .enumerate() + .min_by_key(|(_, item)| item.0) + .map(|(index, _)| index) + { + let (_, event, mut reader, through) = ready.swap_remove(index); + dispatch_ingress_event(daemon, event).await; + if let Some((timestamp, event)) = next_evicted_event(daemon, &mut reader, through).await? { + ready.push((timestamp, event, reader, through)); + } + } + Ok(()) +} + +async fn next_evicted_event( + daemon: &Daemon, + reader: &mut journal::JournalReader, + through: u64, +) -> anyhow::Result> { + while reader.position() < through { + let before = reader.position(); + let entry = reader + .next_record() + .await? + .ok_or_else(|| anyhow::anyhow!("replay range ends inside a record"))?; + anyhow::ensure!(entry.through <= through, "invalid replay end boundary"); + if let journal::JournalRecord::Event(redacted) = entry.record { + let mut env = journal::envelope_from_redacted(redacted); + let Some(source) = daemon.translators.canonical_source(&env.source) else { + continue; + }; + env.source = source.to_owned(); + return Ok(Some(( + env.ts_ms, + PendingEvent { + env, + replay_through: before, + journal_through: entry.through, + settings_through: None, + }, + ))); + } + } + anyhow::ensure!(reader.position() == through, "invalid replay end boundary"); + Ok(None) +} + +pub(super) enum IngressMsg { + Wake, Barrier(oneshot::Sender<()>), Stop(oneshot::Sender<()>), } @@ -14,34 +153,37 @@ pub(super) enum IngressMsg { /// journals become the overflow queue and the worker catches them up in place. pub(super) const INGRESS_QUEUE_CAPACITY: usize = 64; -pub(super) fn spawn_ingress_worker(daemon: Arc, mut rx: mpsc::Receiver) { +pub(super) fn spawn_ingress_worker( + daemon: Arc, + mut rx: mpsc::Receiver, + startup: Option>, +) { tokio::spawn(async move { - loop { - if daemon.ingress_overflow.swap(false, Ordering::AcqRel) { - recover_ingress_overflow(&daemon).await; - continue; + if let Some(startup) = startup { + if startup.await.is_err() { + tracing::error!("startup recovery failed; stopping ingress"); + daemon.trigger_shutdown(); + return; + } + } + while let Some(msg) = rx.recv().await { + // The same queue drains before barriers and after ordinary wakes. + // Capture never waits for its reader, actors, or destinations. + if let Err(error) = drain_capture_buffer(&daemon).await { + tracing::error!(%error, "capture recovery failed; stopping ingress"); + daemon.trigger_shutdown(); + return; } - let Some(msg) = rx.recv().await else { - break; - }; match msg { - IngressMsg::Stop(reply) => { + IngressMsg::Wake => {} + IngressMsg::Barrier(reply) => { + daemon.settle_session_actors().await; + let _ = retry_pending_sessions(&daemon).await; let _ = reply.send(()); - break; - } - IngressMsg::Event(event) => { - dispatch_ingress_event(&daemon, *event).await; } - IngressMsg::Barrier(reply) => { - loop { - daemon.settle_session_actors().await; - let _ = retry_pending_sessions(&daemon).await; - if !daemon.ingress_overflow.swap(false, Ordering::AcqRel) { - break; - } - recover_ingress_overflow(&daemon).await; - } + IngressMsg::Stop(reply) => { let _ = reply.send(()); + break; } } } @@ -49,19 +191,9 @@ pub(super) fn spawn_ingress_worker(daemon: Arc, mut rx: mpsc::Receiver, event: PendingEvent) { - let ingress_key = - event.env.route.as_ref().and_then(|route| { - DeliveryKey::new(&event.env.source, &event.env.session_id, route).ok() - }); - if ingress_key.as_ref().is_some_and(|key| { - daemon - .ingress_dispatched - .lock() - .get(key) - .is_some_and(|through| *through >= event.journal_through) - }) { - return; - } + // Startup discovery rebuilds these observations before this worker is + // released, so a historical event cannot overwrite a newer capture. + daemon.remember_claude_transcripts(&event.env); let journal_through = event.journal_through; // A child may arrive immediately after a parent's tool hook. Catch prior @@ -74,17 +206,8 @@ pub(super) async fn dispatch_ingress_event(daemon: &Arc, event: PendingE daemon.settle_session_actors().await; } let source_env = event.env.clone(); - match accept_event(daemon, event).await { - Ok(()) => { - if let Some(key) = ingress_key { - let mut dispatched = daemon.ingress_dispatched.lock(); - let through = dispatched.entry(key).or_default(); - *through = (*through).max(journal_through); - } - } - Err(error) => { - tracing::warn!(%error, "journaled ingress event could not be dispatched"); - } + if let Err(error) = accept_event(daemon, event).await { + tracing::warn!(%error, "journaled ingress event could not be dispatched"); } // Authentication and route discovery do not prevent canonical persistence. match daemon.translation_for(&source_env).await { @@ -99,92 +222,6 @@ pub(super) async fn dispatch_ingress_event(daemon: &Arc, event: PendingE } } -/// Drain events omitted from the bounded in-memory queue. The queue contains -/// only a latency fast path; journals remain the complete source of ingress. -pub(super) async fn recover_ingress_overflow(daemon: &Arc) { - let Ok(mut entries) = tokio::fs::read_dir(journal::journal_dir(&daemon.data_dir)).await else { - return; - }; - let mut recovered = 0usize; - while let Ok(Some(entry)) = entries.next_entry().await { - let path = entry.path(); - if path.extension().and_then(|ext| ext.to_str()) != Some("ndjson") { - continue; - } - let recorded_len = journal::JournalReader::recorded_len(&path).await; - let Ok(Some(mut checkpoints)) = journal::JournalReader::open(&path, recorded_len).await - else { - continue; - }; - let mut acknowledged_by_route: HashMap = HashMap::new(); - while let Ok(Some(entry)) = checkpoints.next_record().await { - if let journal::JournalRecord::DeliveryCheckpoint { route, through } = entry.record { - let key = serde_json::to_string(&route).unwrap_or_default(); - let acknowledged = acknowledged_by_route.entry(key).or_default(); - *acknowledged = (*acknowledged).max(through); - } - } - - let Ok(Some(mut reader)) = journal::JournalReader::open(&path, recorded_len).await else { - continue; - }; - let mut before = 0u64; - while let Ok(Some(entry)) = reader.next_record().await { - let through = entry.through; - let journal::JournalRecord::Event(redacted) = entry.record else { - before = through; - continue; - }; - let mut env = journal::envelope_from_redacted(redacted); - let Some(source) = daemon.translators.canonical_source(&env.source) else { - before = through; - continue; - }; - env.source = source.to_string(); - let Some(route) = env.route.as_ref() else { - before = through; - continue; - }; - let route_json = recovered_delivery_route(daemon, &env) - .await - .as_ref() - .and_then(|route| serde_json::to_string(route).ok()) - .unwrap_or_else(|| serde_json::to_string(route).unwrap_or_default()); - let key = match DeliveryKey::new(&env.source, &env.session_id, route) { - Ok(key) => key, - Err(_) => { - before = through; - continue; - } - }; - let dispatched = daemon - .ingress_dispatched - .lock() - .get(&key) - .copied() - .unwrap_or(0); - let acknowledged = acknowledged_by_route.get(&route_json).copied().unwrap_or(0); - if through > dispatched.max(acknowledged) { - dispatch_ingress_event( - daemon, - PendingEvent { - env, - replay_through: before, - journal_through: through, - settings_through: None, - }, - ) - .await; - recovered += 1; - } - before = through; - } - } - if recovered > 0 { - tracing::info!(recovered, "drained journal-backed ingress overflow"); - } -} - pub(super) fn spawn_pending_reconciler(daemon: Arc) { tokio::spawn(async move { let mut tick = tokio::time::interval(Duration::from_millis(100)); @@ -318,25 +355,28 @@ impl Daemon { /// Caller holds the source-session capture lock; frozen observations enter /// the same durable queue as hooks without reacquiring that lock. - pub(super) async fn journal_and_enqueue(&self, mut env: Envelope) -> Result { - let (replay_through, journal_through) = self - .append_to_journal(&mut env) + pub(super) async fn journal_and_enqueue(&self, env: Envelope) -> Result { + let writer = self + .journal_writer_for(&env.source, &env.session_id) .await .map_err(|error| format!("journal failed: {error}"))?; - self.remember_claude_transcripts(&env); - match self - .ingress_tx - .try_send(IngressMsg::Event(Box::new(PendingEvent { - env, - replay_through, - journal_through, - settings_through: None, - }))) { - Ok(()) => {} - Err(mpsc::error::TrySendError::Full(_)) => { - self.ingress_overflow.store(true, Ordering::Release); - tracing::debug!("ingress queue full; journal will be drained by daemon worker"); - } + let (replay_through, journal_through) = { + let mut writer = writer.lock().await; + let position = self + .append_to_journal(&env, &mut writer) + .await + .map_err(|error| format!("journal failed: {error}"))?; + (position.before, position.through) + }; + let event = PendingEvent { + env, + replay_through, + journal_through, + settings_through: None, + }; + self.capture_buffer.lock().push(&self.data_dir, event); + match self.ingress_tx.try_send(IngressMsg::Wake) { + Ok(()) | Err(mpsc::error::TrySendError::Full(_)) => {} Err(mpsc::error::TrySendError::Closed(_)) => { tracing::warn!("journaled event will be recovered after daemon restart"); } diff --git a/bt-daemon/src/server/maintenance.rs b/bt-daemon/src/server/maintenance.rs index 61e9d45c..442b7df3 100644 --- a/bt-daemon/src/server/maintenance.rs +++ b/bt-daemon/src/server/maintenance.rs @@ -70,10 +70,21 @@ async fn collect_partitions(daemon: &Daemon, age: Duration) { if tokio::fs::try_exists(&wal).await.unwrap_or(true) && !old(&wal, age).await { continue; } + // Scanning a large journal must not hold the capture lock. The length + // and route set are revalidated under that lock before deletion. + let Ok(partition) = crate::pipeline::journal::SpanJournal::open(&storage).await else { + continue; + }; + let Ok(routes) = partition.routes().await else { + continue; + }; + if !captured_routes_registered(&daemon.data_dir, &path, recorded, &routes).await { + continue; + } let lock = daemon.session_lock(&first.source, &first.session_id); let _guard = lock.lock().await; let namespace = crate::ids::session_namespace(&first.source, &first.session_id); - if daemon.journals.lock().contains_key(&namespace) + if daemon.journals.lock().contains_key(&path) || daemon.translations.lock().contains_key(&namespace) { continue; @@ -81,19 +92,24 @@ async fn collect_partitions(daemon: &Daemon, age: Duration) { if journal::JournalReader::recorded_len(&path).await != recorded { continue; } - let Ok(partition) = crate::pipeline::journal::SpanJournal::open(&storage).await else { + let Ok(current_routes) = partition.routes().await else { continue; }; - let Ok(routes) = partition.routes().await else { + let (Ok(current_routes), Ok(scanned_routes)) = ( + serde_json::to_value(¤t_routes), + serde_json::to_value(&routes), + ) else { continue; }; + if current_routes != scanned_routes { + continue; + } let pending = daemon.pending_sessions.lock().values().any(|state| { state.events.iter().any(|event| { event.env.source == first.source && event.env.session_id == first.session_id }) }); - if pending || !captured_routes_registered(&daemon.data_dir, &path, recorded, &routes).await - { + if pending { continue; } @@ -269,6 +285,9 @@ pub(super) fn spawn_session_reaper(daemon: Arc, idle_timeout: Duration) pub(super) fn spawn_gc(daemon: Arc) { tokio::spawn(async move { + // Retention may scan old transcripts. Start it after recovery releases + // live ingress, so it cannot delay resuming an active source session. + collect_garbage(&daemon).await; loop { tokio::select! { _ = daemon.shutdown.notified() => return, @@ -313,6 +332,9 @@ pub(super) async fn drain_all(daemon: &Arc) { } daemon.begin_quiesce(); let _capture_guard = daemon.capture_gate.write().await; + // Captures already acknowledged may still be waiting to register their + // transcript paths in ingress. Observe only after those have dispatched. + daemon.settle_ingress().await; daemon.observe_claude_transcripts(None, true).await; daemon.settle_ingress().await; if let Err(error) = resolve_pending_sessions(daemon, |_| true).await { @@ -357,6 +379,7 @@ pub(super) async fn drain_all(daemon: &Arc) { daemon.sessions.lock().clear(); daemon.translations.lock().clear(); daemon.journals.lock().clear(); + daemon.source_inputs.lock().clear(); let (reply, done) = oneshot::channel(); if daemon .ingress_tx @@ -406,6 +429,17 @@ mod tests { wire::Envelope, }; + #[allow(dead_code)] + mod distributed_fixture { + use crate as bt_daemon; + include!(concat!( + env!("CARGO_MANIFEST_DIR"), + "/tests/support/distributed.rs" + )); + } + + use super::super::recovery::tests::TestAuth; + fn event(session: &str, route: SessionRoute) -> Envelope { Envelope { source: "debug".into(), @@ -421,6 +455,295 @@ mod tests { source_version: None, } } + + fn paused_ingress(root: &std::path::Path) -> (Arc, oneshot::Sender<()>) { + let (ready, startup) = oneshot::channel(); + ( + Daemon::new_with_startup( + debug_serve_options("test", root), + root.into(), + Some(startup), + ), + ready, + ) + } + + #[tokio::test] + async fn concurrent_recovery_and_dispatch_share_one_source_producer() { + let f = crate::pipeline::translation::tests::SourceFixture::new("debug").await; + f.source.shutdown().await; + let (daemon, ready) = paused_ingress(&f.options.data_dir); + let start = Arc::new(tokio::sync::Barrier::new(8)); + let mut opening = tokio::task::JoinSet::new(); + for _ in 0..8 { + let daemon = daemon.clone(); + let start = start.clone(); + opening.spawn(async move { + start.wait().await; + daemon + .translation_for(&event("checkpoint", SessionRoute::default())) + .await + .unwrap() + }); + } + let first = opening.join_next().await.unwrap().unwrap(); + while let Some(source) = opening.join_next().await { + assert!( + Arc::ptr_eq(&first, &source.unwrap()), + "source initialization must have one owner" + ); + } + ready.send(()).unwrap(); + drain_all(&daemon).await; + } + + #[tokio::test] + async fn capture_is_durable_during_startup_and_recovery_stops_before_it() { + let tmp = tempfile::tempdir().unwrap(); + let mut env = event("resuming", SessionRoute::default()); + let path = journal::source_journal_path(tmp.path(), &env.source, &env.session_id); + let old_end = JournalWriter::open_path(&path) + .await + .unwrap() + .append(&env) + .await + .unwrap(); + let (daemon, ready) = paused_ingress(tmp.path()); + env.ts_ms = 2; + env.event = "next".into(); + assert!( + tokio::time::timeout(Duration::from_secs(1), daemon.capture_event(env)) + .await + .unwrap() + .unwrap() + ); + assert!(tokio::fs::read_to_string(&path) + .await + .unwrap() + .contains("next")); + assert!(daemon.translations.lock().is_empty()); + assert!( + daemon.source_inputs.lock().is_empty(), + "capture must retain live payloads only in the bounded ingress queue" + ); + let recovered = discover_unprocessed_journals(&daemon).await; + assert_eq!(recovered.len(), 1); + assert_eq!(recovered[0].journal_through, old_end); + for event in recovered { + dispatch_ingress_event(&daemon, event).await; + } + drain_capture_buffer(&daemon).await.unwrap(); + ready.send(()).unwrap(); + daemon.settle_ingress().await; + let source = daemon.translations.lock().values().next().unwrap().clone(); + assert!(source.barrier().await.unwrap() > 0); + let storage = tmp.path().join("pipeline").join(format!( + "{}.sqlite", + crate::ids::session_storage_id("debug", "resuming") + )); + let partition = crate::pipeline::journal::SpanJournal::open(&storage) + .await + .unwrap(); + assert_eq!( + partition.translated_through().await.unwrap(), + journal::JournalReader::recorded_len(&path).await + ); + drain_all(&daemon).await; + } + + #[tokio::test] + async fn repeated_overflow_preserves_each_session_and_event_order() { + let tmp = tempfile::tempdir().unwrap(); + let (daemon, ready) = paused_ingress(tmp.path()); + // Four journals interleave through two eviction ranges and a live tail. + let mut envs = Vec::new(); + for source in ["debug", "pi"] { + for session in ["first", "second"] { + let mut env = event(session, SessionRoute::default()); + env.source = source.into(); + env.event = "context".into(); + envs.push(env); + } + } + let mut first = None; + let total = 2 * INGRESS_QUEUE_CAPACITY + 12; + for index in 0..total { + let mut env = envs[index % envs.len()].clone(); + env.ts_ms = index as i64; + env.payload = serde_json::json!({"index":index,"event":{"messages":[{"role":"user","content":"history".repeat(1024)},{"role":"assistant","content":index}]}}); + daemon.capture_event(env).await.unwrap(); + if index + 1 == 2 * INGRESS_QUEUE_CAPACITY { + let CaptureWork::Evicted(bounds) = daemon.next_capture_work() else { + panic!("missing first eviction"); + }; + assert_eq!(bounds.len(), envs.len()); + first = Some(bounds); + } + } + { + let buffer = daemon.capture_buffer.lock(); + assert_eq!(buffer.events.len(), INGRESS_QUEUE_CAPACITY); + assert_eq!( + buffer.events.front().unwrap().env.payload["index"], + total - INGRESS_QUEUE_CAPACITY + ); + } + replay_evicted(&daemon, &first.unwrap()).await.unwrap(); + let CaptureWork::Evicted(second) = daemon.next_capture_work() else { + panic!("new captures must extend recovery"); + }; + replay_evicted(&daemon, &second).await.unwrap(); + drain_capture_buffer(&daemon).await.unwrap(); + ready.send(()).unwrap(); + daemon.settle_ingress().await; + for (slot, env) in envs.iter().enumerate() { + // These sources may have no authenticated delivery actor. Wait for + // translation itself before inspecting its durable head. + daemon + .translation_for(env) + .await + .unwrap() + .barrier() + .await + .unwrap(); + let path = journal::source_journal_path(tmp.path(), &env.source, &env.session_id); + let storage = tmp.path().join("pipeline").join(format!( + "{}.sqlite", + crate::ids::session_storage_id(&env.source, &env.session_id) + )); + let partition = crate::pipeline::journal::SpanJournal::open(&storage) + .await + .unwrap(); + assert_eq!( + partition.translated_through().await.unwrap(), + journal::JournalReader::recorded_len(&path).await + ); + let db = rusqlite::Connection::open(storage).unwrap(); + let count: usize = db + .query_row( + "SELECT count(*) FROM batches WHERE committed=1", + [], + |row| row.get(0), + ) + .unwrap(); + let expected: Vec<_> = (slot..total).step_by(envs.len()).collect(); + assert_eq!(count, expected.len()); + if env.source == "debug" { + let actual: Vec<_> = partition + .read_after(0, total) + .await + .unwrap() + .into_iter() + .filter_map(|(_, op)| { + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = op; + row.input?.get("index")?.as_u64() + }) + .collect(); + assert_eq!( + actual, + expected.into_iter().map(|i| i as u64).collect::>() + ); + } + } + drain_all(&daemon).await; + } + + #[tokio::test] + async fn shutdown_observes_transcript_registered_by_queued_capture() { + let tmp = tempfile::tempdir().unwrap(); + let transcript = tmp.path().join("transcript.jsonl"); + tokio::fs::write(&transcript, b"{}\n").await.unwrap(); + let (daemon, ready) = paused_ingress(tmp.path()); + let mut env = event("late-transcript", SessionRoute::default()); + env.source = "claude-code".into(); + env.event = "Stop".into(); + env.payload = serde_json::json!({"transcript_path": transcript}); + daemon.capture_event(env.clone()).await.unwrap(); + tokio::fs::write(&transcript, b"{}\n{}\n").await.unwrap(); + + let shutdown_daemon = daemon.clone(); + let shutdown = tokio::spawn(async move { drain_all(&shutdown_daemon).await }); + while !daemon.quiescing.load(Ordering::SeqCst) { + tokio::task::yield_now().await; + } + tokio::task::yield_now().await; + drain_capture_buffer(&daemon).await.unwrap(); + ready.send(()).unwrap(); + tokio::time::timeout(Duration::from_secs(10), shutdown) + .await + .unwrap() + .unwrap(); + let path = journal::source_journal_path(tmp.path(), &env.source, &env.session_id); + let content = tokio::fs::read_to_string(path).await.unwrap(); + assert!( + content.contains("TranscriptUpdate"), + "late transcript must be journaled before shutdown" + ); + } + + #[tokio::test] + async fn startup_overflow_preserves_parent_child_tool_order() { + use distributed_fixture::{AgentKind, DistributedFixtures, ProcessTree}; + let tmp = tempfile::tempdir().unwrap(); + let (ready, startup) = oneshot::channel(); + let mut opts = debug_serve_options("test", tmp.path()); + opts.auth_provider = Some(Arc::new(TestAuth)); + let daemon = Daemon::new_with_startup(opts, tmp.path().into(), Some(startup)); + let mut fixtures = DistributedFixtures::new(); + let parent = ProcessTree::root(1000); + let child = ProcessTree::child(1002, 1001, &parent); + let mut events = + fixtures.start_turn(AgentKind::OpenCode, "parent", &parent, "delegate", 100); + events.push(fixtures.open_tool( + AgentKind::OpenCode, + "parent", + &parent, + "tool", + "do work", + 102, + )); + let child_events = + fixtures.start_turn(AgentKind::OpenCode, "child", &child, "do work", 103); + let child_key = automatic_link_key(&child_events[0]); + events.extend(child_events); + events.push(fixtures.close_tool( + AgentKind::OpenCode, + "parent", + &parent, + "tool", + "do work", + "done", + 105, + )); + for env in events { + daemon.capture_event(env).await.unwrap(); + } + for i in 0..INGRESS_QUEUE_CAPACITY { + let mut filler = event("filler", SessionRoute::default()); + filler.ts_ms = i as i64; + daemon.capture_event(filler).await.unwrap(); + } + drain_capture_buffer(&daemon).await.unwrap(); + ready.send(()).unwrap(); + daemon.settle_ingress().await; + let route = daemon + .automatic_links + .lock() + .get(&child_key) + .cloned() + .expect("child must be linked"); + let Some(crate::wire::TraceDestination::ParentSpan { components }) = route.destination + else { + panic!("child was not attached to the parent tool"); + }; + let tool_id = crate::ids::span_id( + &crate::ids::session_namespace("opencode", "parent"), + "tool:parent:tool", + ); + assert_eq!(components.span_id.as_deref(), Some(tool_id.as_str())); + drain_all(&daemon).await; + } + #[tokio::test] async fn retention_requires_complete_inactive_partitions_and_nonempty_heads() { for scenario in [ diff --git a/bt-daemon/src/server/mod.rs b/bt-daemon/src/server/mod.rs index fba1ef20..2b25a34f 100644 --- a/bt-daemon/src/server/mod.rs +++ b/bt-daemon/src/server/mod.rs @@ -102,6 +102,8 @@ struct SessionAuthState { lease: AuthLease, } +type JournalSlot = Arc>>>; + pub struct Daemon { version: String, data_dir: PathBuf, @@ -115,12 +117,14 @@ pub struct Daemon { /// Serializes daemon-side routing and actor creation independently of /// capture, which must never inherit actor or sink backpressure. dispatch_locks: Mutex>>>, - journals: Mutex>>>, + journals: Mutex>, + source_inputs: Mutex>, claude_transcripts: Mutex>, managed_run_sessions: Mutex>>, auth_errors: Mutex>, sessions: Mutex>>, translations: Mutex>>, + translation_init: tokio::sync::Mutex<()>, correlation: Arc, automatic_links: Mutex>, standalone_links: Mutex>, @@ -129,8 +133,10 @@ pub struct Daemon { correlation_locks: Mutex>>>, correlation_changed: Arc, ingress_tx: mpsc::Sender, - ingress_overflow: AtomicBool, - ingress_dispatched: Mutex>, + /// Earliest live append during startup, keyed by source journal path. + /// Recovery must not absorb queued live events into its historical head. + startup_cutoffs: Mutex>>, + capture_buffer: Mutex, started: Instant, last_activity: Mutex, /// Prevents a shutdown drain from racing capture that has passed the @@ -143,8 +149,18 @@ pub struct Daemon { } impl Daemon { + #[cfg(test)] fn new(opts: ServeOptions, data_dir: PathBuf) -> Arc { + Self::new_with_startup(opts, data_dir, None) + } + + fn new_with_startup( + opts: ServeOptions, + data_dir: PathBuf, + startup: Option>, + ) -> Arc { let (ingress_tx, ingress_rx) = mpsc::channel(INGRESS_QUEUE_CAPACITY); + let startup_cutoffs = startup.as_ref().map(|_| HashMap::new()); let daemon = Arc::new(Daemon { version: opts.version, data_dir, @@ -156,11 +172,13 @@ impl Daemon { session_locks: Mutex::new(HashMap::new()), dispatch_locks: Mutex::new(HashMap::new()), journals: Mutex::new(HashMap::new()), + source_inputs: Mutex::new(HashMap::new()), claude_transcripts: Mutex::new(HashMap::new()), managed_run_sessions: Mutex::new(HashMap::new()), auth_errors: Mutex::new(HashMap::new()), sessions: Mutex::new(HashMap::new()), translations: Mutex::new(HashMap::new()), + translation_init: tokio::sync::Mutex::new(()), correlation: Arc::new(crate::correlation::CorrelationRegistry::default()), automatic_links: Mutex::new(HashMap::new()), standalone_links: Mutex::new(HashSet::new()), @@ -169,8 +187,8 @@ impl Daemon { correlation_locks: Mutex::new(HashMap::new()), correlation_changed: Arc::new(Notify::new()), ingress_tx, - ingress_overflow: AtomicBool::new(false), - ingress_dispatched: Mutex::new(HashMap::new()), + startup_cutoffs: Mutex::new(startup_cutoffs), + capture_buffer: Mutex::new(CaptureBuffer::default()), started: Instant::now(), last_activity: Mutex::new(Instant::now()), capture_gate: tokio::sync::RwLock::new(()), @@ -179,7 +197,7 @@ impl Daemon { shutting_down: AtomicBool::new(false), shutdown: Notify::new(), }); - spawn_ingress_worker(daemon.clone(), ingress_rx); + spawn_ingress_worker(daemon.clone(), ingress_rx, startup); daemon } @@ -228,6 +246,12 @@ pub async fn run(args: ServeArgs, opts: ServeOptions) -> anyhow::Result<()> { paths::ensure_private_dir(parent)?; } + #[cfg(unix)] + let Some(ownership) = transport::wait_for_endpoint_lock(&socket).await? + else { + tracing::info!("another daemon owns {}; exiting", socket.display()); + return Ok(()); + }; let listener = match transport::claim(&socket, || probe_alive(&socket)).await? { Some(l) => l, None => { @@ -240,25 +264,49 @@ pub async fn run(args: ServeArgs, opts: ServeOptions) -> anyhow::Result<()> { }; tracing::info!(socket = %socket.display(), "bt-daemon listening"); - let daemon = Daemon::new(opts, data_dir); - restore_active_parent_snapshots(&daemon.data_dir, &daemon.correlation).await; - restore_pending_sessions(&daemon).await; - collect_garbage(&daemon).await; - recover_unprocessed_journals(&daemon).await; - spawn_pending_reconciler(daemon.clone()); + let (ready, startup) = oneshot::channel(); + let daemon = Daemon::new_with_startup(opts, data_dir, Some(startup)); let idle_timeout = Duration::from_secs(args.idle_timeout_secs); - spawn_idle_watchdog(daemon.clone(), idle_timeout); - spawn_session_reaper( - daemon.clone(), - Duration::from_secs(args.session_idle_timeout_secs), - ); - spawn_gc(daemon.clone()); + let session_idle_timeout = Duration::from_secs(args.session_idle_timeout_secs); + let recovery_daemon = daemon.clone(); + let recovery = tokio::spawn(async move { + restore_active_parent_snapshots(&recovery_daemon.data_dir, &recovery_daemon.correlation) + .await; + restore_pending_sessions(&recovery_daemon).await; + let candidates = discover_unprocessed_journals(&recovery_daemon).await; + *recovery_daemon.startup_cutoffs.lock() = None; + for event in candidates { + dispatch_ingress_event(&recovery_daemon, event).await; + } + recovery_daemon.settle_session_actors().await; + let _ = retry_pending_sessions(&recovery_daemon).await; + if let Err(error) = drain_capture_buffer(&recovery_daemon).await { + tracing::error!(%error, "startup ingress recovery failed"); + recovery_daemon.trigger_shutdown(); + return; + } + let _ = ready.send(()); + if !recovery_daemon.shutting_down.load(Ordering::SeqCst) { + spawn_pending_reconciler(recovery_daemon.clone()); + spawn_idle_watchdog(recovery_daemon.clone(), idle_timeout); + spawn_session_reaper(recovery_daemon.clone(), session_idle_timeout); + spawn_gc(recovery_daemon); + } + }); let accept_result = accept_loop(daemon.clone(), listener).await; // Graceful drain regardless of why we stopped. + daemon.trigger_shutdown(); + if let Err(error) = recovery.await { + tracing::error!(%error, "startup recovery task failed"); + } drain_all(&daemon).await; transport::cleanup(&socket); + #[cfg(unix)] + if let Err(error) = fs2::FileExt::unlock(&ownership) { + tracing::warn!(%error, "daemon endpoint lock release failed"); + } accept_result } diff --git a/bt-daemon/src/server/recovery.rs b/bt-daemon/src/server/recovery.rs index 3a453bac..1166fa8e 100644 --- a/bt-daemon/src/server/recovery.rs +++ b/bt-daemon/src/server/recovery.rs @@ -2,7 +2,7 @@ use super::*; -pub(super) async fn recover_unprocessed_journals(daemon: &Arc) { +pub(super) async fn discover_unprocessed_journals(daemon: &Arc) -> Vec { let pending: Vec = daemon .pending_sessions .lock() @@ -10,7 +10,7 @@ pub(super) async fn recover_unprocessed_journals(daemon: &Arc) { .flat_map(|state| state.events.iter().cloned()) .collect(); let Ok(mut entries) = tokio::fs::read_dir(journal::journal_dir(&daemon.data_dir)).await else { - return; + return Vec::new(); }; let mut candidates = Vec::new(); while let Ok(Some(entry)) = entries.next_entry().await { @@ -18,7 +18,16 @@ pub(super) async fn recover_unprocessed_journals(daemon: &Arc) { if path.extension().and_then(|ext| ext.to_str()) != Some("ndjson") { continue; } - let recorded_len = journal::JournalReader::recorded_len(&path).await; + let recorded_len = daemon.journal_snapshot(&path).await; + let recorded_len = daemon + .startup_cutoffs + .lock() + .as_ref() + .and_then(|cutoffs| cutoffs.get(&path).copied()) + .map_or(recorded_len, |cutoff| recorded_len.min(cutoff)); + if let Some(cutoffs) = daemon.startup_cutoffs.lock().as_mut() { + cutoffs.entry(path.clone()).or_insert(recorded_len); + } let Ok(Some(mut reader)) = journal::JournalReader::open(&path, recorded_len).await else { continue; }; @@ -191,31 +200,30 @@ pub(super) async fn recover_unprocessed_journals(daemon: &Arc) { } candidates.sort_by_key(|event| event.env.ts_ms); let recovered = candidates.len(); - for event in candidates { - let _ = daemon - .ingress_tx - .send(IngressMsg::Event(Box::new(event))) - .await; - } - // Reconciliation stays in the daemon worker and follows every recovered - // event in queue order. Dropping the receiver is intentional: startup and - // hook capture do not wait for translation or reporting. - let (reply, _ignored) = oneshot::channel(); - let _ = daemon.ingress_tx.send(IngressMsg::Barrier(reply)).await; if recovered > 0 { tracing::info!( recovered, "queued unprocessed journal sessions for recovery" ); } + candidates +} + +#[cfg(test)] +pub(super) async fn recover_unprocessed_journals(daemon: &Arc) { + for event in discover_unprocessed_journals(daemon).await { + dispatch_ingress_event(daemon, event).await; + } + let (reply, _ignored) = oneshot::channel(); + let _ = daemon.ingress_tx.send(IngressMsg::Barrier(reply)).await; } #[cfg(test)] -mod tests { +pub(super) mod tests { use super::*; use crate::translate::{SpanOp, SpanRow}; - struct TestAuth; + pub(crate) struct TestAuth; #[async_trait::async_trait] impl AuthProvider for TestAuth { async fn resolve( diff --git a/bt-daemon/src/server/sessions.rs b/bt-daemon/src/server/sessions.rs index b1028f17..67e5000c 100644 --- a/bt-daemon/src/server/sessions.rs +++ b/bt-daemon/src/server/sessions.rs @@ -37,6 +37,16 @@ impl Daemon { &self, env: &Envelope, ) -> anyhow::Result> { + let storage_key = crate::ids::session_namespace(&env.source, &env.session_id); + if let Some(source) = self.translations.lock().get(&storage_key).cloned() { + return Ok(source); + } + // Initialization owns one reducer per source. This lock covers only + // opening its SQLite head and spawning the actor, never capture/replay. + let _init = self.translation_init.lock().await; + if let Some(source) = self.translations.lock().get(&storage_key).cloned() { + return Ok(source); + } let event_path = journal::ensure_source_journal(&self.data_dir, &env.source, &env.session_id).await?; let namespace = @@ -45,37 +55,29 @@ impl Daemon { { env.session_id.clone() } else { - crate::ids::session_namespace(&env.source, &env.session_id) + storage_key.clone() }; - let journal = self + let events = self .journal_writer_for(&env.source, &env.session_id) .await?; - let storage_key = crate::ids::session_namespace(&env.source, &env.session_id); - let existing = self.translations.lock().get(&storage_key).cloned(); - Ok(match existing { - Some(translation) => translation, - None => { - let translation = crate::pipeline::translation::Translation::spawn( - crate::pipeline::translation::TranslationOptions { - source: env.source.clone(), - session_id: env.session_id.clone(), - namespace: namespace.clone(), - version: self.version.clone(), - data_dir: self.data_dir.clone(), - events: journal.clone(), - event_path: event_path.clone(), - registry: self.translators.clone(), - correlation: self.correlation.clone(), - correlation_changed: self.correlation_changed.clone(), - }, - ) - .await?; - self.translations - .lock() - .insert(storage_key, translation.clone()); - translation - } - }) + let source = crate::pipeline::translation::Translation::spawn( + crate::pipeline::translation::TranslationOptions { + source: env.source.clone(), + session_id: env.session_id.clone(), + namespace, + version: self.version.clone(), + data_dir: self.data_dir.clone(), + events, + event_path, + input: self.source_input_for(&env.source, &env.session_id), + registry: self.translators.clone(), + correlation: self.correlation.clone(), + correlation_changed: self.correlation_changed.clone(), + }, + ) + .await?; + self.translations.lock().insert(storage_key, source.clone()); + Ok(source) } pub(super) async fn session_for( @@ -182,6 +184,7 @@ impl Daemon { /// journal file for the rest of the daemon's life; deterministic span ids /// mean a late event simply rebuilds it from the journal. pub(super) async fn retire_session(&self, key: &DeliveryKey) { + self.ingress_barrier().await; self.observe_claude_transcripts(Some(key), false).await; self.ingress_barrier().await; let lock = self.dispatch_lock(&key.source, &key.session_id); @@ -220,7 +223,19 @@ impl Daemon { self.claude_transcripts.lock().retain(|candidate, _| { candidate.source != key.source || candidate.session_id != key.session_id }); - self.journals.lock().remove(&storage_key); + self.journals.lock().remove(&journal::source_journal_path( + &self.data_dir, + &key.source, + &key.session_id, + )); + let mut inputs = self.source_inputs.lock(); + if inputs + .get(&storage_key) + .is_some_and(|input| input.is_empty()) + { + inputs.remove(&storage_key); + } + drop(inputs); self.translations.lock().remove(&storage_key); { let mut locks = self.session_locks.lock(); @@ -231,9 +246,6 @@ impl Daemon { } } self.dispatch_locks.lock().remove(&storage_key); - self.ingress_dispatched.lock().retain(|candidate, _| { - candidate.source != key.source || candidate.session_id != key.session_id - }); } tracing::info!(session_id = %key.session_id, "session retired"); } @@ -255,30 +267,66 @@ impl Daemon { source: &str, session_id: &str, ) -> anyhow::Result>> { - let storage_key = crate::ids::session_namespace(source, session_id); - if let Some(writer) = self.journals.lock().get(&storage_key).cloned() { - return Ok(writer); - } - let path = journal::ensure_source_journal(&self.data_dir, source, session_id).await?; - let writer = Arc::new(tokio::sync::Mutex::new( - JournalWriter::open_path(&path).await?, - )); - Ok(self + let path = journal::source_journal_path(&self.data_dir, source, session_id); + let slot = self .journals .lock() - .entry(storage_key) - .or_insert_with(|| writer.clone()) - .clone()) + .entry(path.clone()) + .or_default() + .clone(); + // A recovery actor and capture may open the same source concurrently. + // Tail repair must finish exactly once before either can append. + let writer = slot + .get_or_try_init(|| async { + let path = + journal::ensure_source_journal(&self.data_dir, source, session_id).await?; + let writer = JournalWriter::open_path(&path).await?; + Ok::<_, anyhow::Error>(Arc::new(tokio::sync::Mutex::new(writer))) + }) + .await?; + Ok(writer.clone()) } - pub(super) async fn append_to_journal(&self, env: &mut Envelope) -> anyhow::Result<(u64, u64)> { - let writer = self - .journal_writer_for(&env.source, &env.session_id) - .await?; - let mut writer = writer.lock().await; + /// Take only the durable append position, never hold the writer while + /// reading/replaying records. Recheck after metadata I/O if it opened meanwhile. + pub(super) async fn journal_snapshot(&self, path: &std::path::Path) -> u64 { + let slot = self.journals.lock().get(path).cloned(); + if let Some(writer) = slot.and_then(|slot| slot.get().cloned()) { + return writer.lock().await.position(); + } + let recorded = journal::JournalReader::recorded_len(path).await; + let slot = self.journals.lock().get(path).cloned(); + if let Some(writer) = slot.and_then(|slot| slot.get().cloned()) { + recorded.min(writer.lock().await.position()) + } else { + recorded + } + } + + pub(super) fn source_input_for( + &self, + source: &str, + session_id: &str, + ) -> crate::pipeline::input::SourceInput { + self.source_inputs + .lock() + .entry(crate::ids::session_namespace(source, session_id)) + .or_default() + .clone() + } + + pub(super) async fn append_to_journal( + &self, + env: &Envelope, + writer: &mut JournalWriter, + ) -> anyhow::Result { let before = writer.position(); + if let Some(cutoffs) = self.startup_cutoffs.lock().as_mut() { + let path = journal::source_journal_path(&self.data_dir, &env.source, &env.session_id); + cutoffs.entry(path).or_insert(before); + } let through = writer.append(env).await?; - Ok((before, through)) + Ok(crate::pipeline::input::SourcePosition { before, through }) } pub(super) fn session_lock( diff --git a/bt-daemon/src/transport.rs b/bt-daemon/src/transport.rs index 6b7f829e..a1152b2c 100644 --- a/bt-daemon/src/transport.rs +++ b/bt-daemon/src/transport.rs @@ -9,6 +9,40 @@ use std::path::Path; #[cfg(windows)] use std::time::Duration; +/// Hold a stable lock file for the lifetime of a Unix listener. A timed-out +/// RPC cannot establish that the process which bound the socket has died. +#[cfg(unix)] +pub(crate) fn lock_endpoint(endpoint: &Path) -> anyhow::Result> { + let mut name = endpoint.as_os_str().to_os_string(); + name.push(".lock"); + let file = std::fs::OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(std::path::PathBuf::from(name))?; + match fs2::FileExt::try_lock_exclusive(&file) { + Ok(()) => Ok(Some(file)), + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => Ok(None), + Err(error) => Err(error.into()), + } +} + +/// A departing process (or a short-lived inherited file descriptor) can hold +/// the lock briefly after removing the socket. Allow that handover to finish. +#[cfg(unix)] +pub(crate) async fn wait_for_endpoint_lock( + endpoint: &Path, +) -> anyhow::Result> { + for _ in 0..50 { + if let Some(lock) = lock_endpoint(endpoint)? { + return Ok(Some(lock)); + } + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + } + Ok(None) +} + #[cfg(unix)] pub(crate) type ClientStream = tokio::net::UnixStream; #[cfg(windows)] @@ -144,14 +178,14 @@ where #[cfg(unix)] async fn claim_unix( endpoint: &Path, - probe_alive: &mut F, + _probe_alive: &mut F, ) -> anyhow::Result> where F: FnMut() -> Fut, Fut: Future, { if endpoint.exists() { - if probe_alive().await { + if unix_endpoint_in_use(endpoint).await? { return Ok(None); } cleanup(endpoint); @@ -159,7 +193,7 @@ where match Listener::bind_raw(endpoint) { Ok(listener) => Ok(Some(listener)), Err(error) if error.kind() == std::io::ErrorKind::AddrInUse => { - if probe_alive().await { + if unix_endpoint_in_use(endpoint).await? { Ok(None) } else { cleanup(endpoint); @@ -170,6 +204,25 @@ where } } +/// A successful transport connection proves ownership even when an older +/// daemon has not started accepting RPCs. A timeout is ambiguous and must not +/// authorize unlinking a potentially live listener. +#[cfg(unix)] +async fn unix_endpoint_in_use(endpoint: &Path) -> anyhow::Result { + match tokio::time::timeout(std::time::Duration::from_secs(1), connect(endpoint)).await { + Ok(Ok(_)) | Err(_) => Ok(true), + Ok(Err(error)) + if matches!( + error.kind(), + std::io::ErrorKind::NotFound | std::io::ErrorKind::ConnectionRefused + ) => + { + Ok(false) + } + Ok(Err(error)) => Err(error.into()), + } +} + #[cfg(windows)] async fn claim_windows( endpoint: &Path, @@ -214,6 +267,47 @@ pub(crate) fn cleanup(endpoint: &Path) { #[cfg(windows)] pub(crate) fn cleanup(_endpoint: &Path) {} +#[cfg(all(test, unix))] +mod unix_tests { + #[tokio::test] + async fn startup_owner_excludes_a_contender_before_socket_bind() { + let tmp = tempfile::tempdir().unwrap(); + let endpoint = tmp.path().join("daemon.sock"); + let owner = super::lock_endpoint(&endpoint).unwrap().unwrap(); + assert!(super::lock_endpoint(&endpoint).unwrap().is_none()); + assert!(!endpoint.exists()); + let contender_path = endpoint.clone(); + let mut contender = tokio::spawn(async move { + super::wait_for_endpoint_lock(&contender_path) + .await + .unwrap() + }); + assert!( + tokio::time::timeout(std::time::Duration::from_millis(100), &mut contender) + .await + .is_err() + ); + assert!(!endpoint.exists()); + drop(owner); + assert!(contender.await.unwrap().is_some()); + } + + #[tokio::test] + async fn unresponsive_legacy_listener_is_not_unlinked() { + let tmp = tempfile::tempdir().unwrap(); + let endpoint = tmp.path().join("daemon.sock"); + let listener = tokio::net::UnixListener::bind(&endpoint).unwrap(); + // A legacy daemon doesn't take the new lock or answer initialize while + // recovering. Its successful transport connection still owns the path. + let claimed = super::claim(&endpoint, || async { false }).await.unwrap(); + assert!(claimed.is_none()); + assert!(endpoint.exists()); + drop(listener); + let claimed = super::claim(&endpoint, || async { false }).await.unwrap(); + assert!(claimed.is_some()); + } +} + #[cfg(all(test, windows))] mod tests { use super::*; diff --git a/bt-daemon/tests/distributed_tracing.rs b/bt-daemon/tests/distributed_tracing.rs index 32abad49..e06ca1d2 100644 --- a/bt-daemon/tests/distributed_tracing.rs +++ b/bt-daemon/tests/distributed_tracing.rs @@ -42,6 +42,7 @@ struct RecordingSinkFactory { } impl RecordingSinkFactory { + #[track_caller] fn session(&self, session_id: &str) -> Arc { self.sessions .lock() @@ -54,14 +55,20 @@ impl RecordingSinkFactory { impl SinkFactory for RecordingSinkFactory { fn create(&self, session_id: &str, source: &str) -> anyhow::Result> { - let record = Arc::new(RecordedSession { - source: source.into(), - ..RecordedSession::default() - }); - self.sessions + // Reopening a delivery sink must preserve events already accepted before + // a daemon restart, just as the destination does. + let record = self + .sessions .lock() .unwrap() - .insert(session_id.into(), record.clone()); + .entry(session_id.into()) + .or_insert_with(|| { + Arc::new(RecordedSession { + source: source.into(), + ..RecordedSession::default() + }) + }) + .clone(); Ok(Box::new(RecordingSink { record })) } } @@ -3025,13 +3032,28 @@ async fn pending_restart_preserves_current_candidates_and_decided_legacy_links() if format != "current" { let again_recording = Arc::new(RecordingSinkFactory::default()); let again = spawn_daemon(socket.clone(), data_dir, again_recording.clone()).await; + // Exercise new work after restart. A Stop for a turn already + // completed during recovery may correctly produce no output and + // therefore never create a delivery sink. + forward_all( + &mut fixtures.start_turn( + AgentKind::Claude, + "bounded-child", + &ProcessTree::child(40_002, 40_001, &ProcessTree::root(40_000)), + "new work after upgrade", + 1_702_100_000_050, + ), + &socket, + &host, + ) + .await; forward( fixtures.close_session( AgentKind::Claude, "bounded-child", &ProcessTree::child(40_002, 40_001, &ProcessTree::root(40_000)), "after upgrade", - 1_702_100_000_050, + 1_702_100_000_060, ), &socket, &host, @@ -3039,6 +3061,10 @@ async fn pending_restart_preserves_current_candidates_and_decided_legacy_links() .await; flush("bounded-child", &socket).await; let child = again_recording.session("bounded-child"); + assert!(inserted_rows(&child).iter().any(|row| row + .input + .as_ref() + .is_some_and(|input| input.to_string().contains("new work after upgrade")))); assert!(child .configs .lock() @@ -3307,10 +3333,14 @@ async fn flush(session: &str, socket: &Path) { let result = flush_session(session, socket, 5_000) .await .unwrap_or_else(|error| panic!("flush {session}: {error}")); - assert!( - result.flushed && result.pending == 0, - "session {session} did not fully flush: {result:?}" - ); + if !result.flushed || result.pending != 0 || result.accepted_sessions == 0 { + let status = run_status(StatusArgs { + socket: Some(socket.to_owned()), + session_id: Some(session.to_owned()), + }) + .await; + panic!("session {session} did not fully flush: {result:?}; status: {status:?}"); + } } async fn start_daemon() -> ( diff --git a/bt-daemon/tests/support/translation_streams.rs b/bt-daemon/tests/support/translation_streams.rs new file mode 100644 index 00000000..0de17dd5 --- /dev/null +++ b/bt-daemon/tests/support/translation_streams.rs @@ -0,0 +1,98 @@ +// Native event streams for storage and recovery equivalence tests. +use bt_daemon as fixture_sdk; +#[allow(dead_code)] +mod distributed { + use super::fixture_sdk as bt_daemon; + include!(concat!( + env!("CARGO_MANIFEST_DIR"), + "/tests/support/distributed.rs" + )); +} +use bt_daemon::wire::Envelope; +use distributed::{AgentKind, DistributedFixtures, ProcessTree}; +use serde_json::{json, Value}; +fn event(source: &str, name: &str, ts: i64, payload: Value) -> Envelope { + serde_json::from_value(json!({"source":source,"session_id":"checkpoint","event":name,"ts_ms":ts,"payload":payload})).unwrap() +} +// Keep the fixtures alive while consuming a stream: Codex events refer to +// transcript files owned by this guard. +pub type NativeStream = (String, Vec, Option); + +pub fn native_streams() -> Vec { + let mut cases = Vec::::new(); + for kind in AgentKind::ALL { + let mut fixtures = DistributedFixtures::new(); + let tree = ProcessTree::root(1234); + let id = "checkpoint"; + let t = 1_700_000_000_000; + let mut events = fixtures.start_turn(kind, id, &tree, "first", t); + events.push(fixtures.open_tool(kind, id, &tree, "call", "work", t + 10)); + events.push(fixtures.close_tool(kind, id, &tree, "call", "work", "result", t + 20)); + events.push(fixtures.close_session(kind, id, &tree, "complete", t + 30)); + events.extend(fixtures.start_turn(kind, id, &tree, "second", t + 40)); + cases.push((events[0].source.clone(), events, Some(fixtures))); + } + let events = include_str!("../fixtures/muse-hooks-1.1.1.jsonl") + .lines() + .enumerate() + .map(|(i, line)| { + let payload: Value = serde_json::from_str(line).unwrap(); + let name = payload["hook_event_name"].as_str().unwrap().to_owned(); + event("muse", &name, i as i64 + 1, payload) + }) + .collect(); + cases.push(("muse".to_owned(), events, None)); + let events = + include_str!("../fixtures/cursor/real-cli/interactive-multi-turn-success/hooks.ndjson") + .lines() + .map(|line| { + let capture: Value = serde_json::from_str(line).unwrap(); + let payload = capture["payload"].clone(); + event( + "cursor", + payload["hook_event_name"].as_str().unwrap(), + capture["received_at_ms"].as_i64().unwrap(), + payload.clone(), + ) + }) + .collect(); + cases.push(("cursor".into(), events, None)); + let records = [ + json!({"step_index":0,"source":"USER_EXPLICIT","type":"USER_INPUT","status":"COMPLETED","content":"List files"}), + json!({"step_index":1,"source":"MODEL","type":"PLANNER_RESPONSE","status":"COMPLETED","content":"Inspecting","tool_calls":[{"name":"run_command","args":{"CommandLine":"ls"}}],"usage":{"inputTokens":12,"outputTokens":7,"totalTokens":19}}), + json!({"step_index":2,"source":"MODEL","type":"CORTEX_STEP_TYPE_RUN_COMMAND","status":"COMPLETED","content":"README","tool_calls":[{"name":"run_command","args":{"CommandLine":"ls"},"result":"README"}]}), + ]; + let mut contents = String::new(); + let mut boundaries = Vec::new(); + for record in records { + contents.push_str(&serde_json::to_string(&record).unwrap()); + contents.push('\n'); + boundaries.push(contents.len()); + } + let events = [("PreInvocation", boundaries[0]), ("PostInvocation", boundaries[1]), ("PostToolUse", boundaries[2]), ("Stop", boundaries[2])].into_iter().enumerate().map(|(i, (name, through))| { + event("antigravity", name, i as i64 + 1, json!({"conversationId":"checkpoint","invocationNum":0,"initialNumSteps":1,"stepIdx":2,"fullyIdle":true, + "transcriptPath":"/tmp/checkpoint/transcript.jsonl", + "_bt_transcript_observation":{"path":"/tmp/checkpoint/transcript.jsonl","observed_bytes":through,"full_path":"/tmp/checkpoint/transcript_full.jsonl","full_observed_bytes":through}, + "_bt_transcript_snapshot":{"path":"/tmp/checkpoint/transcript_full.jsonl","contents":contents}})) + }).collect(); + cases.push(("antigravity".to_owned(), events, None)); + let root = + std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/grok/transcript"); + let updates = root.join("updates.jsonl"); + let events = root.join("events.jsonl"); + let mut offset = 0; + let stream = std::fs::read(&updates).unwrap().split_inclusive(|byte| *byte == b'\n').map(|line| { + offset += line.len() as u64; + event("grok", "Wake", 1600, json!({"cwd":"/repo","_bt_grok_transcript_mirrors":{"updates":{"mirror":updates,"through":offset},"events":{"mirror":events,"through":std::fs::metadata(&events).unwrap().len()}}})) + }).collect(); + cases.push(("grok".to_owned(), stream, None)); + cases.push(( + "debug".into(), + vec![ + event("debug", "event", 1, json!({"value":1})), + event("debug", "event", 2, json!({"value":2})), + ], + None, + )); + cases +}