From 582f89df3bc1e74e007a79406ae0ac7494834305 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Tue, 3 Nov 2026 19:36:44 +0800 Subject: [PATCH 1/4] Experiment with lossless journal history prefixes --- bt-daemon/Cargo.toml | 2 +- bt-daemon/src/journal.rs | 170 +++-- bt-daemon/src/journal/history.rs | 1082 +++++++++++++++++++++++++++ bt-daemon/src/pipeline/input.rs | 23 +- bt-daemon/src/server/maintenance.rs | 41 + bt-daemon/src/server/sessions.rs | 5 +- 6 files changed, 1269 insertions(+), 54 deletions(-) create mode 100644 bt-daemon/src/journal/history.rs diff --git a/bt-daemon/Cargo.toml b/bt-daemon/Cargo.toml index d38f914c..ee1993bf 100644 --- a/bt-daemon/Cargo.toml +++ b/bt-daemon/Cargo.toml @@ -36,7 +36,7 @@ regex = "1" rquickjs = "0.12.2" rquickjs-serde = "0.6.1" serde = { version = "1", features = ["derive", "rc"] } -serde_json = { version = "1", features = ["raw_value"] } +serde_json = { version = "1", features = ["raw_value", "float_roundtrip"] } semver = "1" sha2 = "0.10" sysinfo = { version = "0.33.1", default-features = false, features = ["system"] } diff --git a/bt-daemon/src/journal.rs b/bt-daemon/src/journal.rs index 809a3316..890129e8 100644 --- a/bt-daemon/src/journal.rs +++ b/bt-daemon/src/journal.rs @@ -2,7 +2,7 @@ //! (auth-redacted) before the caller is acked, so a restarted daemon can //! rebuild session state by replaying the journal through the translator. //! -//! Format: one [`RedactedEnvelope`] JSON value per line in +//! Format: one plain envelope or versioned history-reference record per line in //! `/journal/----.ndjson`, where an //! overlong sanitized session id keeps only its tail to fit one file name. //! @@ -10,11 +10,13 @@ //! still tell which delivery pipelines a managed child produced after the //! daemon that accepted them has restarted or idle-exited. +mod history; + use crate::wire::{BackendAuth, Envelope, RedactedEnvelope, SessionRoute}; use serde::{Deserialize, Serialize}; use std::collections::HashSet; use std::path::{Path, PathBuf}; -use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt}; +use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncSeekExt, AsyncWriteExt}; pub fn journal_dir(data_dir: &Path) -> PathBuf { data_dir.join("journal") @@ -219,9 +221,29 @@ pub async fn gc_old_managed_runs(data_dir: &Path, max_age: std::time::Duration) /// Append-only journal writer for one session. pub struct JournalWriter { + path: PathBuf, file: tokio::fs::File, position: u64, - pi_context: Vec, + histories: history::Encoder, + failed_write: bool, +} + +// A crash can leave only the unacknowledged final record incomplete. Never +// discard a complete record: malformed complete records must fail recovery. +async fn complete_position(file: &mut tokio::fs::File) -> std::io::Result { + let mut end = file.metadata().await?.len(); + let mut buffer = vec![0; 64 * 1024]; + while end > 0 { + let start = end.saturating_sub(buffer.len() as u64); + file.seek(std::io::SeekFrom::Start(start)).await?; + let bytes = &mut buffer[..(end - start) as usize]; + file.read_exact(bytes).await?; + if let Some(index) = bytes.iter().rposition(|byte| *byte == b'\n') { + return Ok(start + index as u64 + 1); + } + end = start; + } + Ok(0) } impl JournalWriter { @@ -229,16 +251,29 @@ impl JournalWriter { if let Some(dir) = path.parent() { tokio::fs::create_dir_all(dir).await?; } - let file = tokio::fs::OpenOptions::new() + let mut file = tokio::fs::OpenOptions::new() .create(true) + .read(true) + .write(true) + .truncate(false) + .open(path) + .await?; + let position = complete_position(&mut file).await?; + if position < file.metadata().await?.len() { + file.set_len(position).await?; + file.sync_data().await?; + } + drop(file); + let file = tokio::fs::OpenOptions::new() .append(true) .open(path) .await?; - let position = file.metadata().await?.len(); Ok(Self { + path: path.to_owned(), file, position, - pi_context: Vec::new(), + histories: history::Encoder::default(), + failed_write: false, }) } @@ -246,22 +281,53 @@ impl JournalWriter { self.position } + async fn append_line(&mut self, line: &[u8]) -> anyhow::Result { + if self.failed_write { + // A cancelled write job may still be running on Tokio's blocking + // pool. Join it before truncating through a separate repair handle. + let _ = self.file.flush().await; + // Windows append handles cannot truncate; acquire write access + // only for rollback, keeping normal writes append-only. + let repair = tokio::fs::OpenOptions::new() + .write(true) + .open(&self.path) + .await?; + repair.set_len(self.position).await?; + repair.sync_data().await?; + self.failed_write = false; + } + // Mark dirty before awaiting I/O: cancellation must leave the same + // rollback obligation as a reported write/flush/sync error. + self.failed_write = true; + self.file.write_all(line).await?; + self.file.flush().await?; + self.file.sync_data().await?; + self.position += line.len() as u64; + self.failed_write = false; + Ok(self.position) + } + /// Append one redacted event and confirm its durable write before capture acknowledges it. /// - /// The journal is a durability record and is never capped or truncated: - /// dropping entries would silently cost recovery fidelity. Its size is + /// Complete records are never capped or truncated. Only an unacknowledged, + /// unterminated trailing write is discarded before appending. Its size is /// bounded instead by writing each transcript byte once (see /// [`crate::transcript_mirror`]) and by age-based GC, and replay reads it /// as a stream so a large journal never becomes a large allocation. pub async fn append(&mut self, env: &Envelope) -> anyhow::Result { let mut redacted = env.redacted(); - compact_pi_payload(&mut redacted, &mut self.pi_context); - let mut line = serde_json::to_vec(&redacted)?; + compact_pi_payload(&mut redacted); + let mut histories = self.histories.clone(); + let position = self.position; + // Hashing a large incoming history must not occupy the IPC runtime. + let (histories, mut line) = tokio::task::spawn_blocking(move || -> anyhow::Result<_> { + let line = histories.encode(redacted, position)?; + Ok((histories, line)) + }) + .await??; line.push(b'\n'); - self.file.write_all(&line).await?; - self.file.flush().await?; - self.file.sync_data().await?; - self.position += line.len() as u64; + self.append_line(&line).await?; + self.histories = histories; Ok(self.position) } @@ -273,11 +339,7 @@ impl JournalWriter { value["_bt_record_type"] = "pipeline_control".into(); let mut line = serde_json::to_vec(&value)?; line.push(b'\n'); - self.file.write_all(&line).await?; - self.file.flush().await?; - self.file.sync_data().await?; - self.position += line.len() as u64; - Ok(self.position) + self.append_line(&line).await } } @@ -330,6 +392,7 @@ pub struct JournalReader { line_no: usize, position: u64, pi_context: Vec, + histories: history::Decoder, } impl JournalReader { @@ -372,16 +435,17 @@ impl JournalReader { line_no: 0, position: 0, pi_context: Vec::new(), + histories: history::Decoder::default(), })) } /// The journal's current length, which is the bound a session created now /// should replay through. A missing journal replays nothing. pub async fn recorded_len(path: &Path) -> u64 { - tokio::fs::metadata(path) - .await - .map(|meta| meta.len()) - .unwrap_or(0) + let Ok(mut file) = tokio::fs::File::open(path).await else { + return 0; + }; + complete_position(&mut file).await.unwrap_or(0) } /// The next entry, or `None` at end of file. @@ -409,12 +473,21 @@ impl JournalReader { 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) + let mut histories = std::mem::take(&mut self.histories); + let (pi_context, histories, record) = tokio::task::spawn_blocking(move || { + let record = decode_record( + &bytes, + &path, + line_no, + position, + &mut pi_context, + &mut histories, + ); + (pi_context, histories, record) }) .await?; self.pi_context = pi_context; + self.histories = histories; if let Some(record) = record? { return Ok(Some(record)); } @@ -457,6 +530,7 @@ fn decode_record( line_no: usize, position: u64, pi_context: &mut Vec, + histories: &mut history::Decoder, ) -> anyhow::Result> { let line = std::str::from_utf8(bytes) .map_err(|error| anyhow::anyhow!("journal {}:{}: {error}", path.display(), line_no))? @@ -483,7 +557,24 @@ fn decode_record( == Some("pipeline_control") { JournalRecord::Control(serde_json::from_value(value)?) + } else if value + .get("_bt_record_type") + .and_then(serde_json::Value::as_str) + == Some(history::FORMAT) + { + let expanded = histories.expand(value, path, position - bytes.len() as u64, bytes.len())?; + let event: RedactedEnvelope = serde_json::from_value(expanded)?; + if event.source == "pi" && event.event == "context" { + // New contexts reconstruct independently; don't pin the last + // pre-upgrade history throughout the rest of a long recovery. + *pi_context = Vec::new(); + } + JournalRecord::Event(event) } else { + anyhow::ensure!( + value.get("_bt_record_type").is_none(), + "unsupported journal record format" + ); 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) @@ -498,7 +589,7 @@ fn decode_record( const PI_MESSAGES_DELTA: &str = "_bt_messages_delta"; -fn compact_pi_payload(event: &mut RedactedEnvelope, previous: &mut Vec) { +fn compact_pi_payload(event: &mut RedactedEnvelope) { if event.source != "pi" { return; } @@ -510,27 +601,6 @@ fn compact_pi_payload(event: &mut RedactedEnvelope, previous: &mut Vec { - let Some(messages) = native - .remove("messages") - .and_then(|messages| messages.as_array().cloned()) - else { - return; - }; - let common_prefix = previous - .iter() - .zip(&messages) - .take_while(|(left, right)| left == right) - .count(); - native.insert( - PI_MESSAGES_DELTA.into(), - serde_json::json!({ - "common_prefix": common_prefix, - "suffix": messages[common_prefix..], - }), - ); - *previous = messages; - } "before_provider_request" => { // The translator records only the allowlisted request // configuration and tool definitions, never the prompt or messages. @@ -740,7 +810,7 @@ mod tests { } #[tokio::test] - async fn pi_context_history_is_delta_encoded_and_replayed() { + async fn pi_context_history_is_prefix_encoded_and_replayed() { let temp = tempfile::tempdir().unwrap(); let path = temp.path().join("pi.ndjson"); let mut writer = JournalWriter::open_path(&path).await.unwrap(); @@ -770,7 +840,7 @@ mod tests { "Pi journal did not eliminate cumulative context copies: stored={} naive={naive_bytes}", stored.len() ); - assert!(!String::from_utf8_lossy(&stored).contains("\"messages\"")); + assert!(String::from_utf8_lossy(&stored).contains("_bt_history_refs")); let through = stored.len() as u64; let mut reader = JournalReader::open(&path, through).await.unwrap().unwrap(); diff --git a/bt-daemon/src/journal/history.rs b/bt-daemon/src/journal/history.rs new file mode 100644 index 00000000..29531ae9 --- /dev/null +++ b/bt-daemon/src/journal/history.rs @@ -0,0 +1,1082 @@ +//! Lossless history prefixes. References point only into the same append-only +//! journal; caches affect efficiency, never the data available for recovery. +use crate::wire::RedactedEnvelope; +use anyhow::{ensure, Context}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use sha2::{Digest, Sha256}; +use std::{ + collections::{BTreeMap, VecDeque}, + io::{BufRead, Seek}, + path::Path, + sync::Arc, +}; + +pub(super) const FORMAT: &str = "event_history_v1"; +const STREAMS: usize = 16; +const PREFIXES: usize = 128; +const CACHE_BYTES: usize = 8 * 1024 * 1024; +const CACHE_RECORDS: usize = 64; + +#[derive(Clone)] +struct Prefix { + hash: [u8; 32], + wire: [u8; 32], + count: usize, + bytes: usize, +} +#[derive(Clone)] +struct History { + key: [u8; 32], + offset: u64, + anchor: u64, + delta_bytes: usize, + prefixes: VecDeque, +} +#[derive(Clone, Default)] +pub(super) struct Encoder(VecDeque); +#[derive(Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct Reference { + path: String, + base: u64, + anchor: u64, + keep: usize, + hash: String, +} + +// No native keys are reserved: reference metadata lives outside payload. +fn paths(value: &Value, path: &str, found: &mut Vec) { + if found.len() == STREAMS { + return; + } + match value { + Value::Object(object) => { + for (key, value) in object { + let next = format!("{path}/{}", key.replace('~', "~0").replace('/', "~1")); + if matches!(key.as_str(), "messages" | "history") && value.is_array() { + found.push(next); + } else { + paths(value, &next, found); + } + if found.len() == STREAMS { + break; + } + } + } + Value::Array(array) => { + for (i, value) in array.iter().enumerate() { + paths(value, &format!("{path}/{i}"), found); + if found.len() == STREAMS { + break; + } + } + } + _ => {} + } +} +struct Counter(usize); +impl std::io::Write for Counter { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.0 += bytes.len(); + Ok(bytes.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} +// Stable type/length framing and sorted object keys make persisted checksums +// independent of serde_json's preserve_order feature. The separate, memory-only +// wire hash preserves object order and number spellings a translator may stringify. +// Normalize equivalent decimal spellings without rounding through f64. Very +// large exponents fall back to their exact spelling; no precision is discarded. +fn hash_number(text: &str, hash: &mut Sha256) { + let (mantissa, exponent) = text.split_once(['e', 'E']).unwrap_or((text, "0")); + let digits = mantissa.trim_start_matches('-').replace('.', ""); + let digits = digits.trim_start_matches('0'); + let coefficient = digits.trim_end_matches('0'); + let fraction = mantissa + .split_once('.') + .map_or(0, |(_, fraction)| fraction.len()); + let exponent = exponent + .parse::() + .ok() + .and_then(|e| e.checked_sub(fraction as i64)) + .and_then(|e| e.checked_add((digits.len() - coefficient.len()) as i64)); + if coefficient.is_empty() { + hash.update(b"0"); + } else if let Some(exponent) = exponent { + hash.update(if text.starts_with('-') { b"-" } else { b"+" }); + hash.update((coefficient.len() as u64).to_be_bytes()); + hash.update(coefficient.as_bytes()); + hash.update(exponent.to_be_bytes()); + } else { + hash.update(b"x"); + hash.update((text.len() as u64).to_be_bytes()); + hash.update(text.as_bytes()); + } +} +fn hash_value(value: &Value, hash: &mut Sha256, wire: &mut Sha256) { + let sized = |hash: &mut Sha256, bytes: &[u8]| { + hash.update((bytes.len() as u64).to_be_bytes()); + hash.update(bytes); + }; + match value { + Value::Null => hash.update(b"n"), + Value::Bool(value) => hash.update(if *value { b"t" } else { b"f" }), + Value::Number(number) => { + let text = number.to_string(); + hash.update(b"#"); + hash_number(&text, hash); + wire.update(b"#"); + sized(wire, text.as_bytes()); + } + Value::String(text) => { + hash.update(b"s"); + sized(hash, text.as_bytes()); + } + Value::Array(array) => { + hash.update(b"["); + hash.update((array.len() as u64).to_be_bytes()); + for value in array { + hash_value(value, hash, wire); + } + } + Value::Object(object) => { + hash.update(b"{"); + hash.update((object.len() as u64).to_be_bytes()); + for key in object.keys() { + sized(wire, key.as_bytes()); + } + let mut entries: Vec<_> = object.iter().collect(); + entries.sort_unstable_by(|a, b| a.0.cmp(b.0)); + for (key, value) in entries { + sized(hash, key.as_bytes()); + hash_value(value, hash, wire); + } + } + } +} +fn hex(hash: [u8; 32]) -> String { + hash.iter().map(|byte| format!("{byte:02x}")).collect() +} +fn prefixes(messages: &[Value]) -> anyhow::Result> { + extend_prefixes(VecDeque::new(), messages) +} +fn extend_prefixes( + mut result: VecDeque, + messages: &[Value], +) -> anyhow::Result> { + let mut hash = result.back().map_or([0; 32], |p| p.hash); + let mut wire = result.back().map_or([0; 32], |p| p.wire); + let mut bytes = result.back().map_or(2, |p| p.bytes); + let start = result.back().map_or(0, |p| p.count); + for (index, message) in messages.iter().enumerate() { + let count = start + index + 1; + let mut digest = Sha256::new(); + digest.update(hash); + let mut wire_digest = Sha256::new(); + wire_digest.update(wire); + hash_value(message, &mut digest, &mut wire_digest); + let mut counter = Counter(0); + serde_json::to_writer(&mut counter, message)?; + bytes += counter.0 + usize::from(count > 1); + hash = digest.finalize().into(); + wire = wire_digest.finalize().into(); + if result.len() == PREFIXES { + result.pop_front(); + } + result.push_back(Prefix { + hash, + wire, + count, + bytes, + }); + } + Ok(result) +} + +impl Encoder { + pub(super) fn encode( + &mut self, + mut value: RedactedEnvelope, + offset: u64, + ) -> anyhow::Result> { + let source = &value.source; + let session = &value.session_id; + let mut candidates = Vec::new(); + paths(&value.payload, "", &mut candidates); + if candidates.is_empty() && !(source == "pi" && value.event == "context") { + return Ok(serde_json::to_vec(&value)?); + } + let mut references = Vec::new(); + for pointer in candidates { + let path = format!("/payload{pointer}"); + let mut stream = Sha256::new(); + for part in [source.as_str(), session.as_str(), path.as_str()] { + stream.update((part.len() as u64).to_be_bytes()); + stream.update(part.as_bytes()); + } + let key: [u8; 32] = stream.finalize().into(); + let messages = value + .payload + .pointer(&pointer) + .and_then(Value::as_array) + .context("missing history")?; + let current = prefixes(messages)?; + // Keep a few older heads too: a truncated/branched context can + // return to an earlier prefix without writing its messages again. + let candidate = current.iter().rev().find_map(|prefix| { + self.0 + .iter() + .enumerate() + .filter(|(_, h)| { + h.key == key + && h.prefixes.iter().any(|old| { + old.count == prefix.count + && old.hash == prefix.hash + && old.wire == prefix.wire + }) + }) + .max_by_key(|(index, h)| { + ( + h.prefixes.back().map(|p| p.count) == Some(prefix.count), + *index, + ) + }) + .map(|(index, _)| (index, prefix.clone())) + }); + let previous = candidate + .as_ref() + .and_then(|(index, _)| self.0.remove(*index)); + let full_bytes = current.back().map_or(2, |p| p.bytes); + let mut anchor = offset; + let mut delta_bytes = 0; + let mut unchanged = false; + if let Some(previous) = &previous { + if let Some((_, common)) = &candidate { + let reference = Reference { + path: path.clone(), + base: previous.offset, + anchor: previous.anchor, + keep: common.count, + hash: hex(common.hash), + }; + let cost = serde_json::to_vec(&reference)?.len() + 32; + let accumulated = previous + .delta_bytes + .saturating_add(full_bytes.saturating_sub(common.bytes)) + .saturating_add(cost); + // Each stream pays for its own checkpoints. Charging global + // journal bytes would amplify checkpoints for multiple histories. + if common.bytes > cost && accumulated < full_bytes { + value + .payload + .pointer_mut(&pointer) + .and_then(Value::as_array_mut) + .context("missing history")? + .drain(..common.count); + references.push(reference); + anchor = previous.anchor; + delta_bytes = accumulated; + unchanged = previous.prefixes.back().map(|p| (p.count, p.hash, p.wire)) + == current.back().map(|p| (p.count, p.hash, p.wire)); + } + } + } + let mut entry = if unchanged { + previous.context("missing prefix base")? + } else { + if let Some(previous) = previous { + self.0.push_back(previous); + } + History { + key, + offset, + anchor, + delta_bytes, + prefixes: current, + } + }; + entry.delta_bytes = delta_bytes; + self.0.push_back(entry); + while self.0.len() > STREAMS { + self.0.pop_front(); + } + } + // A wrapper deliberately cannot deserialize as an older plain envelope: + // rollback must fail closed, rather than translate an unexpanded suffix. + #[derive(Serialize)] + struct Stored<'a> { + _bt_record_type: &'static str, + envelope: &'a RedactedEnvelope, + #[serde(skip_serializing_if = "Vec::is_empty")] + _bt_history_refs: Vec, + } + Ok(serde_json::to_vec(&Stored { + _bt_record_type: FORMAT, + envelope: &value, + _bt_history_refs: references, + })?) + } +} + +struct Cached { + value: Arc, + bytes: usize, + anchors: BTreeMap, + prefixes: BTreeMap>, +} +#[derive(Default)] +pub(super) struct Decoder { + records: BTreeMap, + order: VecDeque, + bytes: usize, +} +impl Decoder { + fn remember( + &mut self, + offset: u64, + value: &Value, + bytes: usize, + anchors: BTreeMap, + prefixes: BTreeMap>, + ) { + if self.records.contains_key(&offset) { + return; + } + // One oversized event travels alone, as on the live SpanOp feed. + while !self.records.is_empty() + && (self.bytes.saturating_add(bytes) > CACHE_BYTES + || self.records.len() >= CACHE_RECORDS) + { + if let Some(old) = self + .order + .pop_front() + .and_then(|key| self.records.remove(&key)) + { + self.bytes -= old.bytes; + } else { + break; + } + } + self.records.insert( + offset, + Cached { + value: Arc::new(value.clone()), + bytes, + anchors, + prefixes, + }, + ); + self.order.push_back(offset); + self.bytes += bytes; + } + fn read(file: &mut std::fs::File, offset: u64, limit: u64) -> anyhow::Result { + ensure!(offset < limit, "history reference is not backward"); + if offset > 0 { + file.seek(std::io::SeekFrom::Start(offset - 1))?; + let mut byte = [0]; + std::io::Read::read_exact(file, &mut byte)?; + ensure!( + byte[0] == b'\n', + "history reference is not at a record boundary" + ); + } + file.seek(std::io::SeekFrom::Start(offset))?; + let mut line = Vec::new(); + std::io::BufReader::new(std::io::Read::take(&mut *file, limit - offset)) + .read_until(b'\n', &mut line)?; + ensure!(line.last() == Some(&b'\n'), "incomplete history reference"); + Ok(serde_json::from_slice(&line)?) + } + pub(super) fn expand( + &mut self, + mut value: Value, + file_path: &Path, + offset: u64, + bytes: usize, + ) -> anyhow::Result { + let references: Vec = value + .get("_bt_history_refs") + .cloned() + .map(serde_json::from_value) + .transpose()? + .unwrap_or_default(); + let mut stored_paths = Vec::new(); + paths(&value["envelope"]["payload"], "/payload", &mut stored_paths); + // Empty suffixes cannot become new writer bases. Keep their actual + // referenced heads hot instead of caching duplicate complete histories. + let cacheable = stored_paths.iter().any(|path| { + value["envelope"] + .pointer(path) + .and_then(Value::as_array) + .is_some_and(|a| !a.is_empty()) + }); + let mut expanded_bytes = bytes; + let mut anchors = BTreeMap::new(); + let mut validated = BTreeMap::new(); + let mut value = value + .as_object_mut() + .and_then(|object| object.remove("envelope")) + .context("missing history envelope")?; + let mut file = None; + let mut seen = Vec::::new(); + for reference in references { + ensure!( + reference.path.starts_with("/payload/"), + "history reference outside payload" + ); + ensure!( + !seen.iter().any(|p| p == &reference.path + || p.starts_with(&format!("{}/", reference.path)) + || reference.path.starts_with(&format!("{p}/"))), + "overlapping history references" + ); + seen.push(reference.path.clone()); + ensure!( + reference.anchor <= reference.base && reference.base < offset && reference.keep > 0, + "invalid history reference range" + ); + let mut base = reference.base; + let mut limit = offset; + let mut count = reference.keep; + let mut parts = Vec::new(); + let mut cached_prefix = None; + loop { + if let Some(record) = self.records.get_mut(&base) { + ensure!( + record.anchors.get(&reference.path).copied().unwrap_or(base) + == reference.anchor + && record.value["source"] == value["source"] + && record.value["session_id"] == value["session_id"], + "history reference crossed its cached stream" + ); + let messages = record + .value + .pointer(&reference.path) + .and_then(Value::as_array) + .context("cached history is not an array")?; + ensure!(count <= messages.len(), "history prefix exceeds cache"); + if !record.prefixes.contains_key(&reference.path) { + record + .prefixes + .insert(reference.path.clone(), prefixes(messages)?); + } + let points = &record.prefixes[&reference.path]; + let points = if points.iter().any(|p| p.count == count) { + points + .iter() + .filter(|p| p.count <= count) + .cloned() + .collect() + } else { + prefixes(&messages[..count])? + }; + cached_prefix = Some(points); + parts.push(messages[..count].to_vec()); + self.order.retain(|key| *key != base); + self.order.push_back(base); + break; + } + let file = match &mut file { + Some(file) => file, + slot @ None => slot.insert(std::fs::File::open(file_path)?), + }; + let record = Self::read(file, base, limit)?; + ensure!( + record["_bt_record_type"] == FORMAT + && record["envelope"]["source"] == value["source"] + && record["envelope"]["session_id"] == value["session_id"], + "history reference crossed its stream" + ); + let messages = record["envelope"] + .pointer(&reference.path) + .and_then(Value::as_array) + .context("history reference target is not an array")?; + let refs: Vec = record + .get("_bt_history_refs") + .cloned() + .map(serde_json::from_value) + .transpose()? + .unwrap_or_default(); + if let Some(parent) = refs.iter().find(|r| r.path == reference.path) { + ensure!( + parent.anchor == reference.anchor + && parent.anchor <= parent.base + && parent.base < base + && parent.keep > 0, + "invalid history dependency" + ); + ensure!( + count <= parent.keep.saturating_add(messages.len()), + "history prefix exceeds target" + ); + parts.push(messages[..count.saturating_sub(parent.keep)].to_vec()); + count = count.min(parent.keep); + limit = base; + base = parent.base; + } else { + ensure!( + base == reference.anchor && count <= messages.len(), + "invalid history anchor" + ); + parts.push(messages[..count].to_vec()); + break; + } + } + let points = if let Some(mut points) = cached_prefix { + // The final part came from a validated cached prefix. Only + // uncached suffixes need hashing when unwinding dependencies. + for suffix in parts.iter().rev().skip(1) { + points = extend_prefixes(points, suffix)?; + } + points + } else { + let mut points = VecDeque::new(); + for part in parts.iter().rev() { + points = extend_prefixes(points, part)?; + } + points + }; + let prefix = points.back().context("empty referenced prefix")?; + ensure!( + hex(prefix.hash) == reference.hash, + "history prefix checksum mismatch" + ); + expanded_bytes = expanded_bytes.saturating_add(prefix.bytes); + let suffix = value + .pointer_mut(&reference.path) + .and_then(Value::as_array_mut) + .context("history suffix is not an array")?; + let points = extend_prefixes(points, suffix)?; + let mut prefix: Vec<_> = parts.into_iter().rev().flatten().collect(); + prefix.append(suffix); + *suffix = prefix; + anchors.insert(reference.path.clone(), reference.anchor); + validated.insert(reference.path, points); + } + if cacheable { + self.remember(offset, &value, expanded_bytes, anchors, validated); + } + Ok(value) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ + journal::{JournalReader, JournalWriter}, + wire::Envelope, + }; + use serde_json::json; + + fn event(messages: Vec) -> Envelope { + serde_json::from_value( + json!({"source":"muse","session_id":"test","event":"PreLLMCall","ts_ms":1, + "payload":{"messages":messages,"native":{"_bt_history_refs":"untouched"}}}), + ) + .unwrap() + } + fn message(index: usize) -> Value { + json!({"role":"user","content":format!("{index} {}", "unicode 📚\n".repeat(128))}) + } + + #[test] + fn fingerprints_are_stable_without_coalescing_wire_order_or_number_types() { + let left: Value = serde_json::from_str(r#"{"z":1,"a":[true,"é"]}"#).unwrap(); + let right: Value = serde_json::from_str(r#"{"a":[true,"é"],"z":1}"#).unwrap(); + let a = prefixes(std::slice::from_ref(&left)) + .unwrap() + .pop_back() + .unwrap(); + let b = prefixes(std::slice::from_ref(&right)) + .unwrap() + .pop_back() + .unwrap(); + // This is part of the on-disk v1 format, including cross-build key order. + assert_eq!( + hex(a.hash), + "77274ba03369e2db77acaabb7dc2d895be5b9106f7b9278ec521e259f1781138" + ); + assert_eq!(a.hash, b.hash); + if serde_json::to_string(&left).unwrap() != serde_json::to_string(&right).unwrap() { + assert_ne!(a.wire, b.wire); + } + let integer = prefixes(&[json!(1)]).unwrap().pop_back().unwrap(); + let decimal = prefixes(&[json!(1.0)]).unwrap().pop_back().unwrap(); + assert_eq!(integer.hash, decimal.hash); + assert_ne!(integer.wire, decimal.wire); + let digest = |number: &str| { + let mut h = Sha256::new(); + hash_number(number, &mut h); + hex(h.finalize().into()) + }; + assert_eq!(digest("0.00100"), digest("1e-3")); + assert_ne!(digest("1.0000000000000000001"), digest("1.0")); + } + + #[tokio::test] + async fn changes_branches_eviction_and_cold_reads_are_lossless() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("events.ndjson"); + let mut writer = JournalWriter::open_path(&path).await.unwrap(); + let mut histories = vec![vec![message(0)], vec![message(0), message(1)]]; + histories.push(histories[1].clone()); // identical + histories.push(vec![message(0), message(2)]); // edited tail + histories.push(vec![message(0)]); // truncation + histories.push(vec![message(1), message(0)]); // reordered / branched + histories.push(vec![json!({"role":"system","summary":"compacted"})]); + histories.push(vec![]); + let numbers: Vec<_> = (0..128u64) + .filter_map(|i| { + serde_json::Number::from_f64(f64::from_bits(i.wrapping_mul(6364136223846793005))) + }) + .map(Value::Number) + .collect(); + let numeric = vec![json!({"role":"tool","values":numbers,"zero":-0.0,"maximum":f64::MAX})]; + histories.push(numeric.clone()); + histories.push(numeric); + histories.extend((0..80).map(|i| vec![message(0), message(i)])); + let mut seed = 17u64; + for _ in 0..160 { + seed = seed.wrapping_mul(6364136223846793005).wrapping_add(1); + let mut branch = histories[(seed as usize >> 8) % histories.len()].clone(); + match seed % 3 { + 0 if branch.len() < 16 => branch.push(message(seed as usize)), + 1 if !branch.is_empty() => { + let index = seed as usize % branch.len(); + branch[index] = message(seed as usize); + } + _ => branch.truncate(branch.len() / 2), + } + histories.push(branch); + } + let mut expected = Vec::new(); + for (i, messages) in histories.into_iter().enumerate() { + if i == 5 { + writer = JournalWriter::open_path(&path).await.unwrap(); + } + let mut env = event(messages); + if i % 7 == 0 { + env.payload["request"] = json!({"history":[message(0)]}); + } + writer.append(&env).await.unwrap(); + expected.push(env.payload); + } + let stored = std::fs::read_to_string(&path).unwrap(); + assert!(stored.contains("_bt_history_refs")); + let mut offset = 0; + for (line, expected) in stored.split_inclusive('\n').zip(&expected) { + // No warmed decoder state: every reference must work from disk. + let value = Decoder::default() + .expand( + serde_json::from_str(line).unwrap(), + &path, + offset, + line.len(), + ) + .unwrap(); + assert_eq!(&value["payload"], expected); + offset += line.len() as u64; + } + let mut reader = JournalReader::open(&path, writer.position()) + .await + .unwrap() + .unwrap(); + for expected in expected { + assert_eq!( + reader.next_entry().await.unwrap().unwrap().payload, + expected + ); + } + assert!(reader.next_entry().await.unwrap().is_none()); + } + + #[tokio::test] + async fn returning_to_an_older_branch_reuses_its_complete_prefix() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("events.ndjson"); + let mut writer = JournalWriter::open_path(&path).await.unwrap(); + let full = event((0..64).map(message).collect()); + let short = event((0..8).map(message).collect()); + let first = writer.append(&full).await.unwrap(); + for _ in 0..100 { + for event in [&short, &full] { + writer.append(event).await.unwrap(); + } + } + assert!( + writer.position() < 3 * first, + "branch returns must reuse exact cached heads" + ); + let stored = std::fs::read_to_string(&path).unwrap(); + let record: Value = serde_json::from_str(stored.lines().nth(2).unwrap()).unwrap(); + assert_eq!(record["_bt_history_refs"][0]["keep"], 64); + assert_eq!(record["envelope"]["payload"]["messages"], json!([])); + let mut reader = JournalReader::open(&path, writer.position()) + .await + .unwrap() + .unwrap(); + assert_eq!( + reader.next_entry().await.unwrap().unwrap().payload, + full.payload + ); + for _ in 0..100 { + for expected in [&short, &full] { + assert_eq!( + reader.next_entry().await.unwrap().unwrap().payload, + expected.payload + ); + } + } + } + + #[tokio::test] + async fn references_are_generic_and_never_modify_the_live_envelope() { + let dir = tempfile::tempdir().unwrap(); + for source in crate::translate::Registry::default_agents().sources() { + let path = dir.path().join(format!("{source}.ndjson")); + let mut writer = JournalWriter::open_path(&path).await.unwrap(); + let mut env = event(vec![message(0)]); + env.source = source.to_owned(); + env.event = "context".into(); + env.payload["a~/b"] = json!({"history":[message(0)]}); + let original = env.payload.clone(); + writer.append(&env).await.unwrap(); + writer.append(&env).await.unwrap(); + assert_eq!(env.payload, original); + let stored = std::fs::read_to_string(&path).unwrap(); + assert!(stored.contains("_bt_history_refs"), "{source}"); + let mut reader = JournalReader::open(&path, writer.position()) + .await + .unwrap() + .unwrap(); + for _ in 0..2 { + assert_eq!( + reader.next_entry().await.unwrap().unwrap().payload, + original + ); + } + } + } + + #[tokio::test] + async fn invalid_references_fail_instead_of_silently_changing_history() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("events.ndjson"); + let mut writer = JournalWriter::open_path(&path).await.unwrap(); + writer.append(&event(vec![message(0)])).await.unwrap(); + let offset = writer + .append(&event(vec![message(0), message(1)])) + .await + .unwrap(); + writer + .append(&event(vec![message(0), message(1), message(2)])) + .await + .unwrap(); + let stored = std::fs::read_to_string(&path).unwrap(); + let value: Value = serde_json::from_str(stored.lines().nth(2).unwrap()).unwrap(); + assert!(value.get("_bt_history_refs").is_some()); + let mut bad = Vec::new(); + for (key, val) in [ + ("base", json!(offset)), + ("base", json!(1)), + ("anchor", json!(offset + 1)), + ("keep", json!(100)), + ("keep", json!(0)), + ("hash", json!("wrong")), + ("path", json!("/source")), + ("path", json!("/payload/missing")), + ] { + let mut copy = value.clone(); + copy["_bt_history_refs"][0][key] = val; + bad.push(copy); + } + let mut copy = value.clone(); + copy["_bt_history_refs"][0]["anchor"] = value["_bt_history_refs"][0]["base"].clone(); + bad.push(copy); + let mut copy = value.clone(); + copy["envelope"]["session_id"] = "other".into(); + bad.push(copy); + let mut copy = value.clone(); + copy["_bt_history_refs"] + .as_array_mut() + .unwrap() + .push(value["_bt_history_refs"][0].clone()); + bad.push(copy); + let mut warm = JournalReader::open(&path, offset).await.unwrap().unwrap(); + for _ in 0..2 { + warm.next_entry().await.unwrap().unwrap(); + } + for value in bad { + assert!(Decoder::default() + .expand(value.clone(), &path, offset, 1) + .is_err()); + assert!(warm.histories.expand(value, &path, offset, 1).is_err()); + } + } + + #[tokio::test] + async fn stream_cache_eviction_changes_only_storage_efficiency() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("events.ndjson"); + let mut writer = JournalWriter::open_path(&path).await.unwrap(); + for i in 0..2 * STREAMS { + let mut env = event(vec![message(0)]); + env.session_id = i.to_string(); + writer.append(&env).await.unwrap(); + } + let env = event(vec![message(0)]); + writer.append(&env).await.unwrap(); + let mut reader = JournalReader::open(&path, writer.position()) + .await + .unwrap() + .unwrap(); + for _ in 0..2 * STREAMS + 1 { + assert_eq!( + reader.next_entry().await.unwrap().unwrap().payload, + env.payload + ); + } + assert!(writer.histories.0.len() <= STREAMS); + assert!( + (reader.histories.bytes <= CACHE_BYTES || reader.histories.records.len() == 1) + && reader.histories.records.len() <= CACHE_RECORDS + ); + } + + #[tokio::test] + async fn released_pi_deltas_and_new_records_can_share_a_journal() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("events.ndjson"); + let mut old = event(vec![]); + old.source = "pi".into(); + old.event = "context".into(); + old.payload = + json!({"event":{"_bt_messages_delta":{"common_prefix":0,"suffix":[message(0)]}}}); + std::fs::write( + &path, + format!("{}\n", serde_json::to_string(&old.redacted()).unwrap()), + ) + .unwrap(); + let mut writer = JournalWriter::open_path(&path).await.unwrap(); + let mut new = old.clone(); + new.payload = + json!({"event":{"messages":[message(0)],"_bt_messages_delta":"native value"}}); + writer.append(&new).await.unwrap(); + writer.append(&new).await.unwrap(); + let mut reader = JournalReader::open(&path, writer.position()) + .await + .unwrap() + .unwrap(); + assert_eq!( + reader.next_entry().await.unwrap().unwrap().payload["event"]["messages"], + json!([message(0)]) + ); + for _ in 0..2 { + assert_eq!( + reader.next_entry().await.unwrap().unwrap().payload, + new.payload + ); + assert!( + reader.pi_context.is_empty(), + "new records must release the legacy history cache" + ); + } + } + + #[test] + fn growing_sessions_do_not_pay_for_full_snapshots_at_fixed_intervals() { + let size = |events: usize| { + let mut encoder = Encoder::default(); + let mut position = 0u64; + let mut messages = Vec::new(); + for index in 0..events { + messages + .push(json!({"role":"user","content":format!("{index}:{}", "x".repeat(128))})); + position += encoder + .encode(event(messages.clone()).redacted(), position) + .unwrap() + .len() as u64 + + 1; + } + position + }; + let small = size(256); + let large = size(512); + assert!( + large < 3 * small, + "doubling a growing session must not quadruple its history storage: {small} -> {large}" + ); + } + + #[tokio::test] + async fn cache_limits_never_truncate_history_or_native_fields() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("events.ndjson"); + let mut writer = JournalWriter::open_path(&path).await.unwrap(); + let mut env = event(vec![message(0)]); + for i in 0..2 * STREAMS { + env.payload[format!("field{i}")] = json!({"history":[message(i)]}); + } + let large = event(vec![ + json!({"role":"tool","content":"image-data".repeat(CACHE_BYTES/8)}), + ]); + let long = event((0..3 * PREFIXES).map(message).collect()); + let mut edited = long.clone(); + edited.payload["messages"][0] = message(999); + for event in [&env, &env, &large, &large, &long, &edited] { + writer.append(event).await.unwrap(); + } + let mut reader = JournalReader::open(&path, writer.position()) + .await + .unwrap() + .unwrap(); + for expected in [&env, &env, &large, &large, &long, &edited] { + assert_eq!( + reader.next_entry().await.unwrap().unwrap().payload, + expected.payload + ); + if expected.payload == large.payload { + assert!(reader.histories.bytes > CACHE_BYTES); + assert_eq!( + reader.histories.records.len(), + 1, + "keep a single oversized history hot" + ); + } + assert!( + (reader.histories.bytes <= CACHE_BYTES || reader.histories.records.len() == 1) + && reader.histories.records.len() <= CACHE_RECORDS + ); + } + assert!(writer + .histories + .0 + .iter() + .all(|h| h.prefixes.len() <= PREFIXES)); + } + + #[tokio::test] + async fn failed_and_partial_appends_never_publish_a_prefix_base() { + use tokio::io::AsyncWriteExt; + let dir = tempfile::tempdir().unwrap(); + for reopen in [false, true] { + let path = dir.path().join(format!("{reopen}.ndjson")); + let mut writer = JournalWriter::open_path(&path).await.unwrap(); + let first = event(vec![message(0)]); + let through = writer.append(&first).await.unwrap(); + writer.file = tokio::fs::File::open(&path).await.unwrap(); + assert!(writer + .append(&event(vec![message(0), message(1)])) + .await + .is_err()); + assert_eq!(writer.position(), through); + writer.file = tokio::fs::OpenOptions::new() + .append(true) + .open(&path) + .await + .unwrap(); + writer.file.write_all(b"{interrupted").await.unwrap(); + writer.file.flush().await.unwrap(); + if reopen { + writer = JournalWriter::open_path(&path).await.unwrap(); + } + let last = event(vec![message(0), message(1), message(2)]); + writer.append(&last).await.unwrap(); + let mut reader = JournalReader::open(&path, writer.position()) + .await + .unwrap() + .unwrap(); + assert_eq!( + reader.next_entry().await.unwrap().unwrap().payload, + first.payload + ); + assert_eq!( + reader.next_entry().await.unwrap().unwrap().payload, + last.payload + ); + assert!(reader.next_entry().await.unwrap().is_none()); + } + } + + #[tokio::test] + async fn cancelling_a_pending_write_preserves_the_last_durable_boundary() { + use std::{future::Future, task::Poll}; + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("events.ndjson"); + let mut writer = JournalWriter::open_path(&path).await.unwrap(); + let first = event(vec![message(0)]); + writer.append(&first).await.unwrap(); + let interrupted = event(vec![json!({"content":"x".repeat(16*1024*1024)})]); + let mut line = serde_json::to_vec(&interrupted.redacted()).unwrap(); + line.push(b'\n'); + let mut pending = Box::pin(writer.append_line(&line)); + let completed = std::future::poll_fn(|cx| { + Poll::Ready(match pending.as_mut().poll(cx) { + Poll::Ready(result) => { + result.unwrap(); + true + } + Poll::Pending => false, + }) + }) + .await; + drop(pending); + if !completed { + assert!(writer.failed_write); + } + let last = event(vec![message(0), message(1)]); + writer.append(&last).await.unwrap(); + let mut reader = JournalReader::open(&path, writer.position()) + .await + .unwrap() + .unwrap(); + assert_eq!( + reader.next_entry().await.unwrap().unwrap().payload, + first.payload + ); + if completed { + assert_eq!( + reader.next_entry().await.unwrap().unwrap().payload, + interrupted.payload + ); + } + assert_eq!( + reader.next_entry().await.unwrap().unwrap().payload, + last.payload + ); + assert!(reader.next_entry().await.unwrap().is_none()); + } + + #[tokio::test] + async fn older_readers_cannot_misinterpret_suffixes_as_complete_events() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("events.ndjson"); + let mut writer = JournalWriter::open_path(&path).await.unwrap(); + for _ in 0..2 { + writer.append(&event(vec![message(0)])).await.unwrap(); + } + for line in std::fs::read_to_string(&path).unwrap().lines() { + assert!(serde_json::from_str::(line).is_err()); + } + // Complete corruption remains visible; only unfinished trailing bytes + // may be discarded on reopen. + let mut file = std::fs::OpenOptions::new() + .append(true) + .open(&path) + .unwrap(); + std::io::Write::write_all(&mut file, b"{bad}\n").unwrap(); + let writer = JournalWriter::open_path(&path).await.unwrap(); + let mut reader = JournalReader::open(&path, writer.position()) + .await + .unwrap() + .unwrap(); + for _ in 0..2 { + reader.next_entry().await.unwrap().unwrap(); + } + assert!(reader.next_entry().await.is_err()); + } +} diff --git a/bt-daemon/src/pipeline/input.rs b/bt-daemon/src/pipeline/input.rs index 63d5586c..67a01a94 100644 --- a/bt-daemon/src/pipeline/input.rs +++ b/bt-daemon/src/pipeline/input.rs @@ -28,7 +28,25 @@ pub(crate) struct SourceRecord { impl SourceRecord { fn bytes(&self) -> u64 { - self.position.through.saturating_sub(self.position.before) + // Journal references can be tiny while the live payload is huge. + // Charge retained values, without serializing/copying their strings. + fn payload_bytes(value: &serde_json::Value) -> u64 { + std::mem::size_of::() as u64 + + match value { + serde_json::Value::String(text) => text.len() as u64, + serde_json::Value::Array(array) => array.iter().map(payload_bytes).sum(), + serde_json::Value::Object(object) => object + .iter() + .map(|(key, value)| key.len() as u64 + 64 + payload_bytes(value)) + .sum(), + _ => 0, + } + } + let stored = self.position.through.saturating_sub(self.position.before); + match &self.event { + SourceEvent::Event(event) => stored.max(payload_bytes(&event.payload)), + _ => stored, + } } } @@ -98,7 +116,8 @@ mod tests { 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; + // References have a small disk footprint but retain full live data. + let stride = 128; let position = |before| SourcePosition { before, through: before + stride, diff --git a/bt-daemon/src/server/maintenance.rs b/bt-daemon/src/server/maintenance.rs index 442b7df3..845f0948 100644 --- a/bt-daemon/src/server/maintenance.rs +++ b/bt-daemon/src/server/maintenance.rs @@ -551,6 +551,47 @@ mod tests { drain_all(&daemon).await; } + #[tokio::test] + async fn recovery_bound_excludes_an_interrupted_tail_and_its_live_replacement() { + use tokio::io::AsyncWriteExt; + let tmp = tempfile::tempdir().unwrap(); + let mut env = event("interrupted", 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 mut file = tokio::fs::OpenOptions::new() + .append(true) + .open(&path) + .await + .unwrap(); + file.write_all(b"{interrupted").await.unwrap(); + file.flush().await.unwrap(); + let (daemon, ready) = paused_ingress(tmp.path()); + let recovered = discover_unprocessed_journals(&daemon).await; + assert_eq!(recovered.len(), 1); + assert_eq!(recovered[0].journal_through, old_end); + env.event = "replacement".into(); + daemon.capture_event(env.clone()).await.unwrap(); + assert_eq!( + daemon.startup_cutoffs.lock().as_ref().unwrap()[&path], + 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.translation_for(&env).await.unwrap(); + source.barrier().await.unwrap(); + assert_eq!(source.progress.pending(), 0); + drain_all(&daemon).await; + } + #[tokio::test] async fn repeated_overflow_preserves_each_session_and_event_order() { let tmp = tempfile::tempdir().unwrap(); diff --git a/bt-daemon/src/server/sessions.rs b/bt-daemon/src/server/sessions.rs index 67e5000c..d622ea61 100644 --- a/bt-daemon/src/server/sessions.rs +++ b/bt-daemon/src/server/sessions.rs @@ -323,7 +323,10 @@ impl Daemon { 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); + cutoffs + .entry(path) + .and_modify(|cutoff| *cutoff = (*cutoff).min(before)) + .or_insert(before); } let through = writer.append(env).await?; Ok(crate::pipeline::input::SourcePosition { before, through }) From a58a3b41bd6c9a8c156f27b5d1a6d0ebb6a0c755 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Tue, 3 Nov 2026 19:40:28 +0800 Subject: [PATCH 2/4] Preserve native JSON depth during history recovery --- bt-daemon/src/journal.rs | 2 +- bt-daemon/src/journal/history.rs | 67 +++++++++++++++++++++++++++++++- 2 files changed, 67 insertions(+), 2 deletions(-) diff --git a/bt-daemon/src/journal.rs b/bt-daemon/src/journal.rs index 890129e8..1e4e61b8 100644 --- a/bt-daemon/src/journal.rs +++ b/bt-daemon/src/journal.rs @@ -538,7 +538,7 @@ fn decode_record( if line.is_empty() { return Ok(None); } - let value: serde_json::Value = serde_json::from_str(line) + let value: serde_json::Value = history::parse(line.as_bytes()) .map_err(|error| anyhow::anyhow!("journal {}:{}: {error}", path.display(), line_no))?; let record = if value .get("_bt_record_type") diff --git a/bt-daemon/src/journal/history.rs b/bt-daemon/src/journal/history.rs index 29531ae9..9fea0cab 100644 --- a/bt-daemon/src/journal/history.rs +++ b/bt-daemon/src/journal/history.rs @@ -18,6 +18,35 @@ const PREFIXES: usize = 128; const CACHE_BYTES: usize = 8 * 1024 * 1024; const CACHE_RECORDS: usize = 64; +pub(super) fn parse(bytes: &[u8]) -> anyhow::Result { + match serde_json::from_slice(bytes) { + Ok(value) => Ok(value), + Err(original) => { + // The storage wrapper must not reduce the native event's JSON + // depth allowance. Parse its raw envelope independently on this + // uncommon path, keeping the normal parser's recursion protection. + #[derive(Deserialize)] + struct Raw<'a> { + _bt_record_type: String, + #[serde(borrow)] + envelope: &'a serde_json::value::RawValue, + #[serde(default)] + _bt_history_refs: Vec, + } + let raw: Raw<'_> = serde_json::from_slice(bytes).map_err(|_| original)?; + ensure!( + raw._bt_record_type == FORMAT, + "unsupported journal record format" + ); + let envelope: Value = serde_json::from_str(raw.envelope.get())?; + Ok( + serde_json::json!({"_bt_record_type":FORMAT,"envelope":envelope, + "_bt_history_refs":raw._bt_history_refs}), + ) + } + } +} + #[derive(Clone)] struct Prefix { hash: [u8; 32], @@ -388,7 +417,7 @@ impl Decoder { std::io::BufReader::new(std::io::Read::take(&mut *file, limit - offset)) .read_until(b'\n', &mut line)?; ensure!(line.last() == Some(&b'\n'), "incomplete history reference"); - Ok(serde_json::from_slice(&line)?) + parse(&line) } pub(super) fn expand( &mut self, @@ -583,6 +612,42 @@ mod tests { json!({"role":"user","content":format!("{index} {}", "unicode 📚\n".repeat(128))}) } + #[tokio::test] + async fn storage_wrapping_preserves_the_native_json_depth_allowance() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("depth.ndjson"); + let mut writer = JournalWriter::open_path(&path).await.unwrap(); + let mut env = event(vec![message(0)]); + let mut deepest = env.clone(); + loop { + env.payload["nested"] = json!([env.payload.get("nested").cloned()]); + if serde_json::from_slice::(&serde_json::to_vec(&env).unwrap()).is_err() { + break; + } + deepest = env.clone(); + } + let first = writer.append(&deepest).await.unwrap(); + writer.append(&deepest).await.unwrap(); + let bytes = std::fs::read(&path).unwrap(); + assert!(serde_json::from_slice::(&bytes[..first as usize]).is_err()); + let mut reader = JournalReader::open(&path, writer.position()) + .await + .unwrap() + .unwrap(); + for _ in 0..2 { + assert_eq!( + reader.next_entry().await.unwrap().unwrap().payload, + deepest.payload + ); + } + // The backward-seek path must use the same depth allowance too. + let record = parse(&bytes[first as usize..]).unwrap(); + let recovered = Decoder::default() + .expand(record, &path, first, bytes.len() - first as usize) + .unwrap(); + assert_eq!(recovered["payload"], deepest.payload); + } + #[test] fn fingerprints_are_stable_without_coalescing_wire_order_or_number_types() { let left: Value = serde_json::from_str(r#"{"z":1,"a":[true,"é"]}"#).unwrap(); From c28fb5ae0df5f1a42ecd11308b7d3c4dc64c8d48 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Tue, 3 Nov 2026 20:30:46 +0800 Subject: [PATCH 3/4] Simplify journal history references and recovery caching --- bt-daemon/src/journal.rs | 7 +- bt-daemon/src/journal/history.rs | 451 ++++++++++++------------------- bt-daemon/src/pipeline/input.rs | 16 +- bt-daemon/src/wire/mod.rs | 14 + 4 files changed, 202 insertions(+), 286 deletions(-) diff --git a/bt-daemon/src/journal.rs b/bt-daemon/src/journal.rs index 1e4e61b8..164fe13f 100644 --- a/bt-daemon/src/journal.rs +++ b/bt-daemon/src/journal.rs @@ -562,7 +562,12 @@ fn decode_record( .and_then(serde_json::Value::as_str) == Some(history::FORMAT) { - let expanded = histories.expand(value, path, position - bytes.len() as u64, bytes.len())?; + let expanded = histories.expand( + value, + path, + position - bytes.len() as u64, + history::record_hash(bytes), + )?; let event: RedactedEnvelope = serde_json::from_value(expanded)?; if event.source == "pi" && event.event == "context" { // New contexts reconstruct independently; don't pin the last diff --git a/bt-daemon/src/journal/history.rs b/bt-daemon/src/journal/history.rs index 9fea0cab..425df258 100644 --- a/bt-daemon/src/journal/history.rs +++ b/bt-daemon/src/journal/history.rs @@ -6,13 +6,12 @@ use serde::{Deserialize, Serialize}; use serde_json::Value; use sha2::{Digest, Sha256}; use std::{ - collections::{BTreeMap, VecDeque}, + collections::VecDeque, io::{BufRead, Seek}, path::Path, - sync::Arc, }; -pub(super) const FORMAT: &str = "event_history_v1"; +pub(super) const FORMAT: &str = "event_history_v2"; const STREAMS: usize = 16; const PREFIXES: usize = 128; const CACHE_BYTES: usize = 8 * 1024 * 1024; @@ -50,7 +49,6 @@ pub(super) fn parse(bytes: &[u8]) -> anyhow::Result { #[derive(Clone)] struct Prefix { hash: [u8; 32], - wire: [u8; 32], count: usize, bytes: usize, } @@ -58,7 +56,7 @@ struct Prefix { struct History { key: [u8; 32], offset: u64, - anchor: u64, + record_hash: [u8; 32], delta_bytes: usize, prefixes: VecDeque, } @@ -69,9 +67,10 @@ pub(super) struct Encoder(VecDeque); struct Reference { path: String, base: u64, - anchor: u64, keep: usize, - hash: String, + // SHA-256 of the referenced record, excluding its terminating newline. + #[serde(rename = "hash")] + checksum: String, } // No native keys are reserved: reference metadata lives outside payload. @@ -104,121 +103,49 @@ fn paths(value: &Value, path: &str, found: &mut Vec) { _ => {} } } -struct Counter(usize); -impl std::io::Write for Counter { +// Hash the exact serialized representation while measuring its storage cost. +struct FingerprintWriter { + bytes: usize, + digest: Sha256, +} +impl std::io::Write for FingerprintWriter { fn write(&mut self, bytes: &[u8]) -> std::io::Result { - self.0 += bytes.len(); + self.bytes += bytes.len(); + self.digest.update(bytes); Ok(bytes.len()) } fn flush(&mut self) -> std::io::Result<()> { Ok(()) } } -// Stable type/length framing and sorted object keys make persisted checksums -// independent of serde_json's preserve_order feature. The separate, memory-only -// wire hash preserves object order and number spellings a translator may stringify. -// Normalize equivalent decimal spellings without rounding through f64. Very -// large exponents fall back to their exact spelling; no precision is discarded. -fn hash_number(text: &str, hash: &mut Sha256) { - let (mantissa, exponent) = text.split_once(['e', 'E']).unwrap_or((text, "0")); - let digits = mantissa.trim_start_matches('-').replace('.', ""); - let digits = digits.trim_start_matches('0'); - let coefficient = digits.trim_end_matches('0'); - let fraction = mantissa - .split_once('.') - .map_or(0, |(_, fraction)| fraction.len()); - let exponent = exponent - .parse::() - .ok() - .and_then(|e| e.checked_sub(fraction as i64)) - .and_then(|e| e.checked_add((digits.len() - coefficient.len()) as i64)); - if coefficient.is_empty() { - hash.update(b"0"); - } else if let Some(exponent) = exponent { - hash.update(if text.starts_with('-') { b"-" } else { b"+" }); - hash.update((coefficient.len() as u64).to_be_bytes()); - hash.update(coefficient.as_bytes()); - hash.update(exponent.to_be_bytes()); - } else { - hash.update(b"x"); - hash.update((text.len() as u64).to_be_bytes()); - hash.update(text.as_bytes()); - } -} -fn hash_value(value: &Value, hash: &mut Sha256, wire: &mut Sha256) { - let sized = |hash: &mut Sha256, bytes: &[u8]| { - hash.update((bytes.len() as u64).to_be_bytes()); - hash.update(bytes); - }; - match value { - Value::Null => hash.update(b"n"), - Value::Bool(value) => hash.update(if *value { b"t" } else { b"f" }), - Value::Number(number) => { - let text = number.to_string(); - hash.update(b"#"); - hash_number(&text, hash); - wire.update(b"#"); - sized(wire, text.as_bytes()); - } - Value::String(text) => { - hash.update(b"s"); - sized(hash, text.as_bytes()); - } - Value::Array(array) => { - hash.update(b"["); - hash.update((array.len() as u64).to_be_bytes()); - for value in array { - hash_value(value, hash, wire); - } - } - Value::Object(object) => { - hash.update(b"{"); - hash.update((object.len() as u64).to_be_bytes()); - for key in object.keys() { - sized(wire, key.as_bytes()); - } - let mut entries: Vec<_> = object.iter().collect(); - entries.sort_unstable_by(|a, b| a.0.cmp(b.0)); - for (key, value) in entries { - sized(hash, key.as_bytes()); - hash_value(value, hash, wire); - } - } - } -} fn hex(hash: [u8; 32]) -> String { hash.iter().map(|byte| format!("{byte:02x}")).collect() } -fn prefixes(messages: &[Value]) -> anyhow::Result> { - extend_prefixes(VecDeque::new(), messages) + +// Persisted checksums cover stored bytes, so JSON parser features cannot +// change their meaning. Prefix fingerprints are only an in-memory optimization. +pub(super) fn record_hash(bytes: &[u8]) -> [u8; 32] { + Sha256::digest(bytes.strip_suffix(b"\n").unwrap_or(bytes)).into() } -fn extend_prefixes( - mut result: VecDeque, - messages: &[Value], -) -> anyhow::Result> { - let mut hash = result.back().map_or([0; 32], |p| p.hash); - let mut wire = result.back().map_or([0; 32], |p| p.wire); - let mut bytes = result.back().map_or(2, |p| p.bytes); - let start = result.back().map_or(0, |p| p.count); +fn prefixes(messages: &[Value]) -> anyhow::Result> { + let mut result = VecDeque::new(); + let mut hash = [0; 32]; + let mut bytes = 2; for (index, message) in messages.iter().enumerate() { - let count = start + index + 1; - let mut digest = Sha256::new(); - digest.update(hash); - let mut wire_digest = Sha256::new(); - wire_digest.update(wire); - hash_value(message, &mut digest, &mut wire_digest); - let mut counter = Counter(0); - serde_json::to_writer(&mut counter, message)?; - bytes += counter.0 + usize::from(count > 1); - hash = digest.finalize().into(); - wire = wire_digest.finalize().into(); + let mut encoded = FingerprintWriter { + bytes: 0, + digest: Sha256::new(), + }; + encoded.digest.update(hash); + serde_json::to_writer(&mut encoded, message)?; + hash = encoded.digest.finalize().into(); + bytes += encoded.bytes + usize::from(index > 0); if result.len() == PREFIXES { result.pop_front(); } result.push_back(Prefix { hash, - wire, - count, + count: index + 1, bytes, }); } @@ -261,11 +188,9 @@ impl Encoder { .enumerate() .filter(|(_, h)| { h.key == key - && h.prefixes.iter().any(|old| { - old.count == prefix.count - && old.hash == prefix.hash - && old.wire == prefix.wire - }) + && h.prefixes + .iter() + .any(|old| old.count == prefix.count && old.hash == prefix.hash) }) .max_by_key(|(index, h)| { ( @@ -276,53 +201,47 @@ impl Encoder { .map(|(index, _)| (index, prefix.clone())) }); let previous = candidate - .as_ref() - .and_then(|(index, _)| self.0.remove(*index)); + .and_then(|(index, common)| self.0.remove(index).map(|history| (history, common))); let full_bytes = current.back().map_or(2, |p| p.bytes); - let mut anchor = offset; let mut delta_bytes = 0; let mut unchanged = false; - if let Some(previous) = &previous { - if let Some((_, common)) = &candidate { - let reference = Reference { - path: path.clone(), - base: previous.offset, - anchor: previous.anchor, - keep: common.count, - hash: hex(common.hash), - }; - let cost = serde_json::to_vec(&reference)?.len() + 32; - let accumulated = previous - .delta_bytes - .saturating_add(full_bytes.saturating_sub(common.bytes)) - .saturating_add(cost); - // Each stream pays for its own checkpoints. Charging global - // journal bytes would amplify checkpoints for multiple histories. - if common.bytes > cost && accumulated < full_bytes { - value - .payload - .pointer_mut(&pointer) - .and_then(Value::as_array_mut) - .context("missing history")? - .drain(..common.count); - references.push(reference); - anchor = previous.anchor; - delta_bytes = accumulated; - unchanged = previous.prefixes.back().map(|p| (p.count, p.hash, p.wire)) - == current.back().map(|p| (p.count, p.hash, p.wire)); - } + if let Some((previous, common)) = &previous { + let reference = Reference { + path: path.clone(), + base: previous.offset, + keep: common.count, + checksum: hex(previous.record_hash), + }; + let cost = serde_json::to_vec(&reference)?.len() + 32; + let accumulated = previous + .delta_bytes + .saturating_add(full_bytes.saturating_sub(common.bytes)) + .saturating_add(cost); + // Each stream pays for its own checkpoints. Charging global + // journal bytes would amplify checkpoints for multiple histories. + if common.bytes > cost && accumulated < full_bytes { + value + .payload + .pointer_mut(&pointer) + .and_then(Value::as_array_mut) + .context("missing history")? + .drain(..common.count); + references.push(reference); + delta_bytes = accumulated; + unchanged = previous.prefixes.back().map(|p| (p.count, p.hash)) + == current.back().map(|p| (p.count, p.hash)); } } let mut entry = if unchanged { - previous.context("missing prefix base")? + previous.context("missing prefix base")?.0 } else { - if let Some(previous) = previous { + if let Some((previous, _)) = previous { self.0.push_back(previous); } History { key, offset, - anchor, + record_hash: [0; 32], delta_bytes, prefixes: current, } @@ -342,36 +261,32 @@ impl Encoder { #[serde(skip_serializing_if = "Vec::is_empty")] _bt_history_refs: Vec, } - Ok(serde_json::to_vec(&Stored { + let bytes = serde_json::to_vec(&Stored { _bt_record_type: FORMAT, envelope: &value, _bt_history_refs: references, - })?) + })?; + let hash = record_hash(&bytes); + for history in self.0.iter_mut().filter(|h| h.offset == offset) { + history.record_hash = hash; + } + Ok(bytes) } } struct Cached { - value: Arc, + value: Value, bytes: usize, - anchors: BTreeMap, - prefixes: BTreeMap>, + hash: [u8; 32], } #[derive(Default)] pub(super) struct Decoder { - records: BTreeMap, - order: VecDeque, + records: VecDeque<(u64, Cached)>, bytes: usize, } impl Decoder { - fn remember( - &mut self, - offset: u64, - value: &Value, - bytes: usize, - anchors: BTreeMap, - prefixes: BTreeMap>, - ) { - if self.records.contains_key(&offset) { + fn remember(&mut self, offset: u64, value: &Value, bytes: usize, hash: [u8; 32]) { + if self.records.iter().any(|(key, _)| *key == offset) { return; } // One oversized event travels alone, as on the live SpanOp feed. @@ -379,29 +294,25 @@ impl Decoder { && (self.bytes.saturating_add(bytes) > CACHE_BYTES || self.records.len() >= CACHE_RECORDS) { - if let Some(old) = self - .order - .pop_front() - .and_then(|key| self.records.remove(&key)) - { - self.bytes -= old.bytes; - } else { - break; + if let Some((_, evicted)) = self.records.pop_front() { + self.bytes -= evicted.bytes; } } - self.records.insert( + self.records.push_back(( offset, Cached { - value: Arc::new(value.clone()), + value: value.clone(), bytes, - anchors, - prefixes, + hash, }, - ); - self.order.push_back(offset); + )); self.bytes += bytes; } - fn read(file: &mut std::fs::File, offset: u64, limit: u64) -> anyhow::Result { + fn read( + file: &mut std::fs::File, + offset: u64, + limit: u64, + ) -> anyhow::Result<(Value, [u8; 32])> { ensure!(offset < limit, "history reference is not backward"); if offset > 0 { file.seek(std::io::SeekFrom::Start(offset - 1))?; @@ -417,14 +328,14 @@ impl Decoder { std::io::BufReader::new(std::io::Read::take(&mut *file, limit - offset)) .read_until(b'\n', &mut line)?; ensure!(line.last() == Some(&b'\n'), "incomplete history reference"); - parse(&line) + Ok((parse(&line)?, record_hash(&line))) } pub(super) fn expand( &mut self, mut value: Value, file_path: &Path, offset: u64, - bytes: usize, + hash: [u8; 32], ) -> anyhow::Result { let references: Vec = value .get("_bt_history_refs") @@ -442,9 +353,6 @@ impl Decoder { .and_then(Value::as_array) .is_some_and(|a| !a.is_empty()) }); - let mut expanded_bytes = bytes; - let mut anchors = BTreeMap::new(); - let mut validated = BTreeMap::new(); let mut value = value .as_object_mut() .and_then(|object| object.remove("envelope")) @@ -464,20 +372,19 @@ impl Decoder { ); seen.push(reference.path.clone()); ensure!( - reference.anchor <= reference.base && reference.base < offset && reference.keep > 0, + reference.base < offset && reference.keep > 0, "invalid history reference range" ); let mut base = reference.base; let mut limit = offset; let mut count = reference.keep; let mut parts = Vec::new(); - let mut cached_prefix = None; + let mut expected_hash = reference.checksum; loop { - if let Some(record) = self.records.get_mut(&base) { + if let Some(index) = self.records.iter().position(|(key, _)| *key == base) { + let record = &mut self.records[index].1; ensure!( - record.anchors.get(&reference.path).copied().unwrap_or(base) - == reference.anchor - && record.value["source"] == value["source"] + record.value["source"] == value["source"] && record.value["session_id"] == value["session_id"], "history reference crossed its cached stream" ); @@ -487,32 +394,27 @@ impl Decoder { .and_then(Value::as_array) .context("cached history is not an array")?; ensure!(count <= messages.len(), "history prefix exceeds cache"); - if !record.prefixes.contains_key(&reference.path) { - record - .prefixes - .insert(reference.path.clone(), prefixes(messages)?); - } - let points = &record.prefixes[&reference.path]; - let points = if points.iter().any(|p| p.count == count) { - points - .iter() - .filter(|p| p.count <= count) - .cloned() - .collect() - } else { - prefixes(&messages[..count])? - }; - cached_prefix = Some(points); + ensure!( + hex(record.hash) == expected_hash, + "history record checksum mismatch" + ); parts.push(messages[..count].to_vec()); - self.order.retain(|key| *key != base); - self.order.push_back(base); + let cached = self + .records + .remove(index) + .context("missing cached record")?; + self.records.push_back(cached); break; } let file = match &mut file { Some(file) => file, slot @ None => slot.insert(std::fs::File::open(file_path)?), }; - let record = Self::read(file, base, limit)?; + let (record, hash) = Self::read(file, base, limit)?; + ensure!( + hex(hash) == expected_hash, + "history record checksum mismatch" + ); ensure!( record["_bt_record_type"] == FORMAT && record["envelope"]["source"] == value["source"] @@ -531,10 +433,7 @@ impl Decoder { .unwrap_or_default(); if let Some(parent) = refs.iter().find(|r| r.path == reference.path) { ensure!( - parent.anchor == reference.anchor - && parent.anchor <= parent.base - && parent.base < base - && parent.keep > 0, + parent.base < base && parent.keep > 0, "invalid history dependency" ); ensure!( @@ -545,48 +444,24 @@ impl Decoder { count = count.min(parent.keep); limit = base; base = parent.base; + expected_hash = parent.checksum.clone(); } else { - ensure!( - base == reference.anchor && count <= messages.len(), - "invalid history anchor" - ); + ensure!(count <= messages.len(), "history prefix exceeds literal"); parts.push(messages[..count].to_vec()); break; } } - let points = if let Some(mut points) = cached_prefix { - // The final part came from a validated cached prefix. Only - // uncached suffixes need hashing when unwinding dependencies. - for suffix in parts.iter().rev().skip(1) { - points = extend_prefixes(points, suffix)?; - } - points - } else { - let mut points = VecDeque::new(); - for part in parts.iter().rev() { - points = extend_prefixes(points, part)?; - } - points - }; - let prefix = points.back().context("empty referenced prefix")?; - ensure!( - hex(prefix.hash) == reference.hash, - "history prefix checksum mismatch" - ); - expanded_bytes = expanded_bytes.saturating_add(prefix.bytes); let suffix = value .pointer_mut(&reference.path) .and_then(Value::as_array_mut) .context("history suffix is not an array")?; - let points = extend_prefixes(points, suffix)?; let mut prefix: Vec<_> = parts.into_iter().rev().flatten().collect(); prefix.append(suffix); *suffix = prefix; - anchors.insert(reference.path.clone(), reference.anchor); - validated.insert(reference.path, points); } if cacheable { - self.remember(offset, &value, expanded_bytes, anchors, validated); + let bytes = crate::wire::retained_json_bytes(&value) as usize; + self.remember(offset, &value, bytes, hash); } Ok(value) } @@ -643,43 +518,75 @@ mod tests { // The backward-seek path must use the same depth allowance too. let record = parse(&bytes[first as usize..]).unwrap(); let recovered = Decoder::default() - .expand(record, &path, first, bytes.len() - first as usize) + .expand(record, &path, first, record_hash(&bytes[first as usize..])) .unwrap(); assert_eq!(recovered["payload"], deepest.payload); } #[test] - fn fingerprints_are_stable_without_coalescing_wire_order_or_number_types() { + fn prefix_matching_preserves_serialized_order_and_number_types() { + let fingerprint = |value: &Value| { + prefixes(std::slice::from_ref(value)) + .unwrap() + .pop_back() + .unwrap() + .hash + }; let left: Value = serde_json::from_str(r#"{"z":1,"a":[true,"é"]}"#).unwrap(); let right: Value = serde_json::from_str(r#"{"a":[true,"é"],"z":1}"#).unwrap(); - let a = prefixes(std::slice::from_ref(&left)) - .unwrap() - .pop_back() - .unwrap(); - let b = prefixes(std::slice::from_ref(&right)) - .unwrap() - .pop_back() - .unwrap(); - // This is part of the on-disk v1 format, including cross-build key order. assert_eq!( - hex(a.hash), - "77274ba03369e2db77acaabb7dc2d895be5b9106f7b9278ec521e259f1781138" + fingerprint(&left) == fingerprint(&right), + serde_json::to_vec(&left).unwrap() == serde_json::to_vec(&right).unwrap() + ); + assert_ne!(fingerprint(&json!(1)), fingerprint(&json!(1.0))); + } + + #[tokio::test] + async fn references_verify_stored_bytes_independently_of_json_features() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("bytes.ndjson"); + // Deliberately noncanonical key order and numeric spelling. Re-encoding + // this Value may differ, depending on the embedding CLI's JSON features. + let base = format!( + r#"{{"_bt_record_type":"{FORMAT}","envelope":{{"source":"muse","session_id":"test","event":"native","ts_ms":1,"payload":{{"messages":[{{"z":1e+00,"a":"text"}}]}}}}}}"# + ); + let value = parse(base.as_bytes()).unwrap(); + let expected = value["envelope"]["payload"].clone(); + let mut next = value.clone(); + next["envelope"]["payload"]["messages"] = json!([]); + next["_bt_history_refs"] = json!([{"path":"/payload/messages","base":0, + "keep":1,"hash":hex(record_hash(base.as_bytes()))}]); + let next_bytes = serde_json::to_vec(&next).unwrap(); + let offset = base.len() as u64 + 1; + std::fs::write( + &path, + format!( + "{base}\n{}\n", + String::from_utf8(next_bytes.clone()).unwrap() + ), + ) + .unwrap(); + assert_eq!( + Decoder::default() + .expand(next.clone(), &path, offset, record_hash(&next_bytes)) + .unwrap()["payload"], + expected ); - assert_eq!(a.hash, b.hash); - if serde_json::to_string(&left).unwrap() != serde_json::to_string(&right).unwrap() { - assert_ne!(a.wire, b.wire); + let mut reader = JournalReader::open(&path, std::fs::metadata(&path).unwrap().len()) + .await + .unwrap() + .unwrap(); + for _ in 0..2 { + assert_eq!( + reader.next_entry().await.unwrap().unwrap().payload, + expected + ); } - let integer = prefixes(&[json!(1)]).unwrap().pop_back().unwrap(); - let decimal = prefixes(&[json!(1.0)]).unwrap().pop_back().unwrap(); - assert_eq!(integer.hash, decimal.hash); - assert_ne!(integer.wire, decimal.wire); - let digest = |number: &str| { - let mut h = Sha256::new(); - hash_number(number, &mut h); - hex(h.finalize().into()) - }; - assert_eq!(digest("0.00100"), digest("1e-3")); - assert_ne!(digest("1.0000000000000000001"), digest("1.0")); + // A valid JSON edit must still invalidate the reference's checksum. + std::fs::write(&path, format!("{}\n", base.replace("text", "edit"))).unwrap(); + assert!(Decoder::default() + .expand(next, &path, offset, record_hash(&next_bytes)) + .is_err()); } #[tokio::test] @@ -740,7 +647,7 @@ mod tests { serde_json::from_str(line).unwrap(), &path, offset, - line.len(), + record_hash(line.as_bytes()), ) .unwrap(); assert_eq!(&value["payload"], expected); @@ -848,7 +755,7 @@ mod tests { for (key, val) in [ ("base", json!(offset)), ("base", json!(1)), - ("anchor", json!(offset + 1)), + ("unknown", json!(offset + 1)), ("keep", json!(100)), ("keep", json!(0)), ("hash", json!("wrong")), @@ -860,9 +767,6 @@ mod tests { bad.push(copy); } let mut copy = value.clone(); - copy["_bt_history_refs"][0]["anchor"] = value["_bt_history_refs"][0]["base"].clone(); - bad.push(copy); - let mut copy = value.clone(); copy["envelope"]["session_id"] = "other".into(); bad.push(copy); let mut copy = value.clone(); @@ -877,9 +781,12 @@ mod tests { } for value in bad { assert!(Decoder::default() - .expand(value.clone(), &path, offset, 1) + .expand(value.clone(), &path, offset, record_hash(b"test")) + .is_err()); + assert!(warm + .histories + .expand(value, &path, offset, record_hash(b"test")) .is_err()); - assert!(warm.histories.expand(value, &path, offset, 1).is_err()); } } diff --git a/bt-daemon/src/pipeline/input.rs b/bt-daemon/src/pipeline/input.rs index 67a01a94..197c6606 100644 --- a/bt-daemon/src/pipeline/input.rs +++ b/bt-daemon/src/pipeline/input.rs @@ -30,21 +30,11 @@ impl SourceRecord { fn bytes(&self) -> u64 { // Journal references can be tiny while the live payload is huge. // Charge retained values, without serializing/copying their strings. - fn payload_bytes(value: &serde_json::Value) -> u64 { - std::mem::size_of::() as u64 - + match value { - serde_json::Value::String(text) => text.len() as u64, - serde_json::Value::Array(array) => array.iter().map(payload_bytes).sum(), - serde_json::Value::Object(object) => object - .iter() - .map(|(key, value)| key.len() as u64 + 64 + payload_bytes(value)) - .sum(), - _ => 0, - } - } let stored = self.position.through.saturating_sub(self.position.before); match &self.event { - SourceEvent::Event(event) => stored.max(payload_bytes(&event.payload)), + SourceEvent::Event(event) => { + stored.max(crate::wire::retained_json_bytes(&event.payload)) + } _ => stored, } } diff --git a/bt-daemon/src/wire/mod.rs b/bt-daemon/src/wire/mod.rs index 48e9dfa4..360998b9 100644 --- a/bt-daemon/src/wire/mod.rs +++ b/bt-daemon/src/wire/mod.rs @@ -23,3 +23,17 @@ pub use rpc::{error_code, Message, Request, RequestId, Response, RpcError}; /// The protocol version this build speaks. Bumped on any breaking change to /// the envelope or method contracts. See `docs/protocol.md`. pub const PROTOCOL_VERSION: u32 = 1; + +// Approximate retained JSON memory without copying or serializing payloads. +pub(crate) fn retained_json_bytes(value: &serde_json::Value) -> u64 { + std::mem::size_of::() as u64 + + match value { + serde_json::Value::String(text) => text.len() as u64, + serde_json::Value::Array(array) => array.iter().map(retained_json_bytes).sum(), + serde_json::Value::Object(object) => object + .iter() + .map(|(key, value)| key.len() as u64 + 64 + retained_json_bytes(value)) + .sum(), + _ => 0, + } +} From c79402a2edade5e5719fe606d2f082605df1cafa Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Tue, 3 Nov 2026 21:29:33 +0800 Subject: [PATCH 4/4] Validate journal history through concrete record types --- bt-daemon/src/journal.rs | 139 +++++++++++--------- bt-daemon/src/journal/history.rs | 210 +++++++++++++++++++------------ 2 files changed, 215 insertions(+), 134 deletions(-) diff --git a/bt-daemon/src/journal.rs b/bt-daemon/src/journal.rs index 164fe13f..3326cfbd 100644 --- a/bt-daemon/src/journal.rs +++ b/bt-daemon/src/journal.rs @@ -524,6 +524,19 @@ impl JournalReader { } } +#[derive(Default, Deserialize)] +enum RecordKind { + #[default] + #[serde(skip_deserializing)] + Legacy, + #[serde(rename = "delivery_checkpoint")] + DeliveryCheckpoint, + #[serde(rename = "pipeline_control")] + Control, + #[serde(rename = "event_history_v2")] + History, +} + fn decode_record( bytes: &[u8], path: &Path, @@ -538,53 +551,42 @@ fn decode_record( if line.is_empty() { return Ok(None); } - let value: serde_json::Value = history::parse(line.as_bytes()) + #[derive(Deserialize)] + struct Header { + #[serde(rename = "_bt_record_type", default)] + kind: RecordKind, + } + let header: Header = 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, + let record = match header.kind { + RecordKind::DeliveryCheckpoint => { + let checkpoint: DeliveryCheckpointRecord = serde_json::from_str(line)?; + 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 if value - .get("_bt_record_type") - .and_then(serde_json::Value::as_str) - == Some(history::FORMAT) - { - let expanded = histories.expand( - value, - path, - position - bytes.len() as u64, - history::record_hash(bytes), - )?; - let event: RedactedEnvelope = serde_json::from_value(expanded)?; - if event.source == "pi" && event.event == "context" { - // New contexts reconstruct independently; don't pin the last - // pre-upgrade history throughout the rest of a long recovery. - *pi_context = Vec::new(); + RecordKind::Control => JournalRecord::Control(serde_json::from_str(line)?), + RecordKind::History => { + let event = histories.expand( + history::parse(line.as_bytes())?, + path, + position - bytes.len() as u64, + history::record_hash(bytes), + )?; + if event.source == "pi" && event.event == "context" { + // Release the last pre-upgrade context once recovery enters the new format. + *pi_context = Vec::new(); + } + JournalRecord::Event(event) + } + RecordKind::Legacy => { + let mut event: RedactedEnvelope = serde_json::from_str(line)?; + expand_pi_payload(&mut event, pi_context).map_err(|error| { + anyhow::anyhow!("journal {}:{}: {error}", path.display(), line_no) + })?; + JournalRecord::Event(event) } - JournalRecord::Event(event) - } else { - anyhow::ensure!( - value.get("_bt_record_type").is_none(), - "unsupported journal record format" - ); - 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, @@ -655,21 +657,18 @@ fn expand_pi_payload( } return Ok(()); }; - let common_prefix = delta - .get("common_prefix") - .and_then(serde_json::Value::as_u64) - .and_then(|value| usize::try_from(value).ok()) - .ok_or_else(|| anyhow::anyhow!("invalid Pi context common prefix"))?; + #[derive(Deserialize)] + struct MessagesDelta { + common_prefix: usize, + suffix: Vec, + } + let delta: MessagesDelta = serde_json::from_value(delta)?; anyhow::ensure!( - common_prefix <= previous.len(), + delta.common_prefix <= previous.len(), "Pi context common prefix exceeds prior context" ); - let suffix = delta - .get("suffix") - .and_then(serde_json::Value::as_array) - .ok_or_else(|| anyhow::anyhow!("invalid Pi context suffix"))?; - previous.truncate(common_prefix); - previous.extend(suffix.iter().cloned()); + previous.truncate(delta.common_prefix); + previous.extend(delta.suffix); native.insert( "messages".into(), serde_json::Value::Array(previous.clone()), @@ -798,6 +797,34 @@ pub fn envelope_from_redacted(r: RedactedEnvelope) -> Envelope { mod tests { use super::*; + #[test] + fn invalid_record_discriminators_cannot_fall_through_to_legacy_events() { + let event = serde_json::json!({"source":"muse","session_id":"test", + "event":"native","ts_ms":1,"payload":{}}); + let decode = |value: &serde_json::Value| { + let bytes = serde_json::to_vec(value).unwrap(); + decode_record( + &bytes, + Path::new("unused.ndjson"), + 1, + bytes.len() as u64, + &mut Vec::new(), + &mut history::Decoder::default(), + ) + }; + assert!(decode(&event).is_ok()); + for kind in [ + serde_json::json!(null), + serde_json::json!(42), + serde_json::json!("Legacy"), + serde_json::json!("unknown"), + ] { + let mut invalid = event.clone(); + invalid["_bt_record_type"] = kind; + assert!(decode(&invalid).is_err()); + } + } + fn pi_context(messages: Vec, ts_ms: i64) -> Envelope { Envelope { source: "pi".into(), diff --git a/bt-daemon/src/journal/history.rs b/bt-daemon/src/journal/history.rs index 425df258..4d32c7b2 100644 --- a/bt-daemon/src/journal/history.rs +++ b/bt-daemon/src/journal/history.rs @@ -17,33 +17,32 @@ const PREFIXES: usize = 128; const CACHE_BYTES: usize = 8 * 1024 * 1024; const CACHE_RECORDS: usize = 64; -pub(super) fn parse(bytes: &[u8]) -> anyhow::Result { - match serde_json::from_slice(bytes) { - Ok(value) => Ok(value), - Err(original) => { - // The storage wrapper must not reduce the native event's JSON - // depth allowance. Parse its raw envelope independently on this - // uncommon path, keeping the normal parser's recursion protection. - #[derive(Deserialize)] - struct Raw<'a> { - _bt_record_type: String, - #[serde(borrow)] - envelope: &'a serde_json::value::RawValue, - #[serde(default)] - _bt_history_refs: Vec, - } - let raw: Raw<'_> = serde_json::from_slice(bytes).map_err(|_| original)?; - ensure!( - raw._bt_record_type == FORMAT, - "unsupported journal record format" - ); - let envelope: Value = serde_json::from_str(raw.envelope.get())?; - Ok( - serde_json::json!({"_bt_record_type":FORMAT,"envelope":envelope, - "_bt_history_refs":raw._bt_history_refs}), - ) - } +pub(super) struct Record { + envelope: RedactedEnvelope, + references: Vec, +} + +pub(super) fn parse(bytes: &[u8]) -> anyhow::Result { + // Parse the envelope independently so the storage wrapper does not consume + // any of the native event's JSON depth allowance. This is the only path. + #[derive(Deserialize)] + #[serde(deny_unknown_fields)] + struct Raw<'a> { + _bt_record_type: super::RecordKind, + #[serde(borrow)] + envelope: &'a serde_json::value::RawValue, + #[serde(default)] + _bt_history_refs: Vec, } + let raw: Raw<'_> = serde_json::from_slice(bytes)?; + ensure!( + matches!(raw._bt_record_type, super::RecordKind::History), + "not a history record" + ); + Ok(Record { + envelope: serde_json::from_str(raw.envelope.get())?, + references: raw._bt_history_refs, + }) } #[derive(Clone)] @@ -275,7 +274,9 @@ impl Encoder { } struct Cached { - value: Value, + source: String, + session_id: String, + payload: Value, bytes: usize, hash: [u8; 32], } @@ -285,10 +286,14 @@ pub(super) struct Decoder { bytes: usize, } impl Decoder { - fn remember(&mut self, offset: u64, value: &Value, bytes: usize, hash: [u8; 32]) { + fn remember(&mut self, offset: u64, value: &RedactedEnvelope, hash: [u8; 32]) { if self.records.iter().any(|(key, _)| *key == offset) { return; } + let bytes = crate::wire::retained_json_bytes(&value.payload) as usize + + value.source.len() + + value.session_id.len() + + std::mem::size_of::(); // One oversized event travels alone, as on the live SpanOp feed. while !self.records.is_empty() && (self.bytes.saturating_add(bytes) > CACHE_BYTES @@ -301,7 +306,9 @@ impl Decoder { self.records.push_back(( offset, Cached { - value: value.clone(), + source: value.source.clone(), + session_id: value.session_id.clone(), + payload: value.payload.clone(), bytes, hash, }, @@ -312,7 +319,7 @@ impl Decoder { file: &mut std::fs::File, offset: u64, limit: u64, - ) -> anyhow::Result<(Value, [u8; 32])> { + ) -> anyhow::Result<(Record, [u8; 32])> { ensure!(offset < limit, "history reference is not backward"); if offset > 0 { file.seek(std::io::SeekFrom::Start(offset - 1))?; @@ -332,31 +339,26 @@ impl Decoder { } pub(super) fn expand( &mut self, - mut value: Value, + record: Record, file_path: &Path, offset: u64, hash: [u8; 32], - ) -> anyhow::Result { - let references: Vec = value - .get("_bt_history_refs") - .cloned() - .map(serde_json::from_value) - .transpose()? - .unwrap_or_default(); + ) -> anyhow::Result { + let Record { + envelope: mut value, + references, + } = record; let mut stored_paths = Vec::new(); - paths(&value["envelope"]["payload"], "/payload", &mut stored_paths); + paths(&value.payload, "", &mut stored_paths); // Empty suffixes cannot become new writer bases. Keep their actual // referenced heads hot instead of caching duplicate complete histories. let cacheable = stored_paths.iter().any(|path| { - value["envelope"] + value + .payload .pointer(path) .and_then(Value::as_array) .is_some_and(|a| !a.is_empty()) }); - let mut value = value - .as_object_mut() - .and_then(|object| object.remove("envelope")) - .context("missing history envelope")?; let mut file = None; let mut seen = Vec::::new(); for reference in references { @@ -371,6 +373,10 @@ impl Decoder { "overlapping history references" ); seen.push(reference.path.clone()); + let pointer = reference + .path + .strip_prefix("/payload") + .context("history reference outside payload")?; ensure!( reference.base < offset && reference.keep > 0, "invalid history reference range" @@ -384,13 +390,12 @@ impl Decoder { if let Some(index) = self.records.iter().position(|(key, _)| *key == base) { let record = &mut self.records[index].1; ensure!( - record.value["source"] == value["source"] - && record.value["session_id"] == value["session_id"], + record.source == value.source && record.session_id == value.session_id, "history reference crossed its cached stream" ); let messages = record - .value - .pointer(&reference.path) + .payload + .pointer(pointer) .and_then(Value::as_array) .context("cached history is not an array")?; ensure!(count <= messages.len(), "history prefix exceeds cache"); @@ -416,22 +421,17 @@ impl Decoder { "history record checksum mismatch" ); ensure!( - record["_bt_record_type"] == FORMAT - && record["envelope"]["source"] == value["source"] - && record["envelope"]["session_id"] == value["session_id"], + record.envelope.source == value.source + && record.envelope.session_id == value.session_id, "history reference crossed its stream" ); - let messages = record["envelope"] - .pointer(&reference.path) + let messages = record + .envelope + .payload + .pointer(pointer) .and_then(Value::as_array) .context("history reference target is not an array")?; - let refs: Vec = record - .get("_bt_history_refs") - .cloned() - .map(serde_json::from_value) - .transpose()? - .unwrap_or_default(); - if let Some(parent) = refs.iter().find(|r| r.path == reference.path) { + if let Some(parent) = record.references.iter().find(|r| r.path == reference.path) { ensure!( parent.base < base && parent.keep > 0, "invalid history dependency" @@ -452,7 +452,8 @@ impl Decoder { } } let suffix = value - .pointer_mut(&reference.path) + .payload + .pointer_mut(pointer) .and_then(Value::as_array_mut) .context("history suffix is not an array")?; let mut prefix: Vec<_> = parts.into_iter().rev().flatten().collect(); @@ -460,8 +461,7 @@ impl Decoder { *suffix = prefix; } if cacheable { - let bytes = crate::wire::retained_json_bytes(&value) as usize; - self.remember(offset, &value, bytes, hash); + self.remember(offset, &value, hash); } Ok(value) } @@ -487,6 +487,40 @@ mod tests { json!({"role":"user","content":format!("{index} {}", "unicode 📚\n".repeat(128))}) } + #[test] + fn record_structure_is_validated_before_recovery() { + let valid = json!({"_bt_record_type":FORMAT,"envelope":event(vec![]).redacted(), + "_bt_history_refs":[{"path":"/payload/messages","base":0,"keep":1,"hash":"abc"}]}); + assert!(parse(&serde_json::to_vec(&valid).unwrap()).is_ok()); + for (pointer, wrong) in [ + ("/_bt_record_type", json!(null)), + ("/_bt_record_type", json!("event_history_v1")), + ("/envelope", json!({})), + ("/envelope", json!(null)), + ("/envelope/source", json!(42)), + ("/envelope/session_id", json!(null)), + ("/envelope/ts_ms", json!("now")), + ("/_bt_history_refs", json!({})), + ("/_bt_history_refs/0/base", json!(-1)), + ("/_bt_history_refs/0/keep", json!("1")), + ] { + let mut bad = valid.clone(); + *bad.pointer_mut(pointer).unwrap() = wrong; + assert!( + parse(&serde_json::to_vec(&bad).unwrap()).is_err(), + "{pointer}" + ); + } + for field in ["source", "session_id", "event", "ts_ms", "payload"] { + let mut bad = valid.clone(); + bad["envelope"].as_object_mut().unwrap().remove(field); + assert!( + parse(&serde_json::to_vec(&bad).unwrap()).is_err(), + "{field}" + ); + } + } + #[tokio::test] async fn storage_wrapping_preserves_the_native_json_depth_allowance() { let dir = tempfile::tempdir().unwrap(); @@ -496,7 +530,7 @@ mod tests { let mut deepest = env.clone(); loop { env.payload["nested"] = json!([env.payload.get("nested").cloned()]); - if serde_json::from_slice::(&serde_json::to_vec(&env).unwrap()).is_err() { + if serde_json::from_slice::(&serde_json::to_vec(&env).unwrap()).is_err() { break; } deepest = env.clone(); @@ -504,7 +538,6 @@ mod tests { let first = writer.append(&deepest).await.unwrap(); writer.append(&deepest).await.unwrap(); let bytes = std::fs::read(&path).unwrap(); - assert!(serde_json::from_slice::(&bytes[..first as usize]).is_err()); let mut reader = JournalReader::open(&path, writer.position()) .await .unwrap() @@ -520,7 +553,7 @@ mod tests { let recovered = Decoder::default() .expand(record, &path, first, record_hash(&bytes[first as usize..])) .unwrap(); - assert_eq!(recovered["payload"], deepest.payload); + assert_eq!(recovered.payload, deepest.payload); } #[test] @@ -551,8 +584,8 @@ mod tests { r#"{{"_bt_record_type":"{FORMAT}","envelope":{{"source":"muse","session_id":"test","event":"native","ts_ms":1,"payload":{{"messages":[{{"z":1e+00,"a":"text"}}]}}}}}}"# ); let value = parse(base.as_bytes()).unwrap(); - let expected = value["envelope"]["payload"].clone(); - let mut next = value.clone(); + let expected = value.envelope.payload.clone(); + let mut next: Value = serde_json::from_str(&base).unwrap(); next["envelope"]["payload"]["messages"] = json!([]); next["_bt_history_refs"] = json!([{"path":"/payload/messages","base":0, "keep":1,"hash":hex(record_hash(base.as_bytes()))}]); @@ -568,8 +601,14 @@ mod tests { .unwrap(); assert_eq!( Decoder::default() - .expand(next.clone(), &path, offset, record_hash(&next_bytes)) - .unwrap()["payload"], + .expand( + parse(&next_bytes).unwrap(), + &path, + offset, + record_hash(&next_bytes) + ) + .unwrap() + .payload, expected ); let mut reader = JournalReader::open(&path, std::fs::metadata(&path).unwrap().len()) @@ -585,7 +624,12 @@ mod tests { // A valid JSON edit must still invalidate the reference's checksum. std::fs::write(&path, format!("{}\n", base.replace("text", "edit"))).unwrap(); assert!(Decoder::default() - .expand(next, &path, offset, record_hash(&next_bytes)) + .expand( + parse(&next_bytes).unwrap(), + &path, + offset, + record_hash(&next_bytes) + ) .is_err()); } @@ -644,13 +688,13 @@ mod tests { // No warmed decoder state: every reference must work from disk. let value = Decoder::default() .expand( - serde_json::from_str(line).unwrap(), + parse(line.as_bytes()).unwrap(), &path, offset, record_hash(line.as_bytes()), ) .unwrap(); - assert_eq!(&value["payload"], expected); + assert_eq!(&value.payload, expected); offset += line.len() as u64; } let mut reader = JournalReader::open(&path, writer.position()) @@ -780,12 +824,22 @@ mod tests { warm.next_entry().await.unwrap().unwrap(); } for value in bad { - assert!(Decoder::default() - .expand(value.clone(), &path, offset, record_hash(b"test")) + let bytes = serde_json::to_vec(&value).unwrap(); + assert!(parse(&bytes) + .and_then(|record| Decoder::default().expand( + record, + &path, + offset, + record_hash(&bytes) + )) .is_err()); - assert!(warm - .histories - .expand(value, &path, offset, record_hash(b"test")) + assert!(parse(&bytes) + .and_then(|record| warm.histories.expand( + record, + &path, + offset, + record_hash(&bytes) + )) .is_err()); } }