Skip to content

Consolidate durable subscription lifecycle and retention #188

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Stephen Belanger (Qard) merged 5 commits into from
Oct 19, 2026
Merged

Consolidate durable subscription lifecycle and retention #188

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
48eb5b4
Consolidate durable subscription lifecycle and retention
Qard Oct 13, 2026
3ae80f5
Close shared journal storage after cancelled reads on shutdown
Qard Oct 14, 2026
a876cf9
Preserve route isolation during refresh, recovery, and retention
Qard Oct 14, 2026
32a1f3c
Wait for shutdown receipt before disconnecting test client
Qard Oct 14, 2026
d689455
Align lifecycle recovery with fallible ledgers and panic bans
Qard Oct 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
67 changes: 39 additions & 28 deletions bt-daemon/src/delivery_ledger.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -263,8 +263,7 @@ impl std::error::Error for CommitFailure {
}

/// Wrap a sink with destination-scoped terminal-span suppression. If durable
/// ledger state cannot be loaded, callers retain normal delivery rather than
/// risking lost trace data.
/// ledger state cannot be loaded, retain input until its recovery condition is met.
pub(crate) struct LedgerSink {
inner: Box<dyn Sink>,
ledger: Option<DeliveryLedger>,
Expand All @@ -284,23 +283,18 @@ impl LedgerSink {
session_id: &str,
config: Option<&SessionConfig>,
bt_version: &str,
) -> Self {
) -> anyhow::Result<Self> {
let loaded = match (data_dir, config) {
(Some(data_dir), Some(config)) if config.destination.is_some() => {
DeliveryLedger::load(data_dir, source, session_id, config)
.await
.map_err(|error| {
tracing::warn!(source, session_id, "delivery ledger unavailable: {error}")
})
.ok()
Some(DeliveryLedger::load(data_dir, source, session_id, config).await?)
}
_ => None,
};
let (ledger, span_origins) = match loaded {
Some((ledger, origins)) => (Some(ledger), origins),
None => (None, HashMap::new()),
};
Self {
Ok(Self {
inner,
ledger,
span_origins,
Expand All @@ -309,7 +303,7 @@ impl LedgerSink {
source_version: None,
bt_version: bt_version.to_owned(),
attached_parent: config.and_then(|config| config.attached_span_ids().0),
}
})
}
}

Expand Down Expand Up @@ -546,7 +540,8 @@ mod tests {
Some(&config),
"test",
)
.await;
.await
.unwrap();
sink.emit(&[terminal("one")]).await.unwrap();
let path = sink.ledger.as_ref().unwrap().path.clone();
let original = std::fs::read(&path).unwrap();
Expand Down Expand Up @@ -604,7 +599,8 @@ mod tests {
Some(&config),
"bt1",
)
.await;
.await
.unwrap();
first.set_capture_versions(None, Some("native1"));
let turn = SpanRow {
span_id: "turn-1".into(),
Expand Down Expand Up @@ -652,7 +648,8 @@ mod tests {
Some(&config),
"bt2",
)
.await;
.await
.unwrap();
recovered.set_capture_versions(Some("v2"), Some("native2"));
recovered
.emit(&[
Expand Down Expand Up @@ -719,7 +716,8 @@ mod tests {
Some(&config),
"bt1",
)
.await;
.await
.unwrap();
sink.set_capture_versions(Some("v1"), Some("native1"));
let path = sink.ledger.as_ref().unwrap().path.clone();
// Block persistence: nothing may reach the sink until origins are saved.
Expand Down Expand Up @@ -754,7 +752,8 @@ mod tests {
Some(&config),
"bt2",
)
.await;
.await
.unwrap();
assert_eq!(recovered.span_origins.len(), 256);
assert_eq!(
recovered.span_origins["span-0"].plugin_version.as_deref(),
Expand Down Expand Up @@ -819,7 +818,8 @@ mod tests {
Some(&config),
if is_child { "bt2" } else { "bt1" },
)
.await;
.await
.unwrap();
sink.set_capture_versions(
Some(if is_child { "v2" } else { "v1" }),
Some(if is_child { "native2" } else { "native1" }),
Expand Down Expand Up @@ -936,7 +936,8 @@ mod tests {
Some(&config("project-a")),
"test",
)
.await;
.await
.unwrap();
first.emit(&[terminal("span-1")]).await.unwrap();
first.flush().await.unwrap();
assert_eq!(first_output.lock().len(), 1);
Expand All @@ -955,7 +956,8 @@ mod tests {
Some(&same_destination),
"test",
)
.await;
.await
.unwrap();
assert_eq!(repeated.emit(&[terminal("span-1")]).await.unwrap(), 0);
assert_eq!(repeated.emit(&[partial_merge("span-1")]).await.unwrap(), 0);
repeated.flush().await.unwrap();
Expand All @@ -974,7 +976,8 @@ mod tests {
Some(&config("project-a")),
"test",
)
.await;
.await
.unwrap();
assert_eq!(first.emit(&[terminal("span-1")]).await.unwrap(), 1);
first.flush().await.unwrap();

Expand All @@ -987,7 +990,8 @@ mod tests {
Some(&config("project-a")),
"test",
)
.await;
.await
.unwrap();
assert_eq!(
second
.emit(&[late_merge("span-1", "output")])
Expand Down Expand Up @@ -1017,7 +1021,8 @@ mod tests {
Some(&config("project-a")),
"test",
)
.await;
.await
.unwrap();
assert_eq!(
third.emit(&[late_merge("span-1", "output")]).await.unwrap(),
0
Expand All @@ -1040,7 +1045,8 @@ mod tests {
Some(&config("project-a")),
"test",
)
.await;
.await
.unwrap();
assert_eq!(
first
.emit(&[late_terminal_merge("root", "session:stop:3", 3)])
Expand All @@ -1059,7 +1065,8 @@ mod tests {
Some(&config("project-a")),
"test",
)
.await;
.await
.unwrap();
assert_eq!(
resumed
.emit(&[late_terminal_merge("root", "session:stop:3", 3)])
Expand Down Expand Up @@ -1092,7 +1099,8 @@ mod tests {
Some(&config("project-a")),
"test",
)
.await;
.await
.unwrap();
initial.emit(&[terminal("span-1")]).await.unwrap();
initial.flush().await.unwrap();

Expand All @@ -1108,7 +1116,8 @@ mod tests {
Some(&config("project-b")),
"test",
)
.await;
.await
.unwrap();
assert_eq!(replay.emit(&[terminal("span-1")]).await.unwrap(), 1);
replay.flush().await.unwrap();
assert_eq!(replay_output.lock().len(), 1);
Expand All @@ -1125,7 +1134,8 @@ mod tests {
Some(&config("project-a")),
"test",
)
.await;
.await
.unwrap();
let mut authoritative = terminal("span-1");
let SpanOp::Insert(row) = &mut authoritative else {
unreachable!();
Expand All @@ -1143,7 +1153,8 @@ mod tests {
Some(&config("project-a")),
"test",
)
.await;
.await
.unwrap();
assert_eq!(repeated.emit(&[terminal("span-1")]).await.unwrap(), 0);
assert_eq!(
repeated
Expand Down
23 changes: 11 additions & 12 deletions bt-daemon/src/dispatch.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,4 @@
//! Per-session dispatch. Each session owns an ordered queue and a single actor
//! task that runs its translator + sink serially, so events for one session
//! are processed strictly in arrival order. Different sessions run
//! concurrently.
//! Session facade over ordered source publication and independent route delivery.
//!
//! Hook acknowledgement happens before this layer, immediately after the raw
//! event is durably journaled. Translation, correlation, and delivery here are
Expand All @@ -12,7 +9,6 @@ use crate::pipeline::{
translation::Translation,
};
use crate::sink::SinkFactory;
use crate::translate::Registry;
use crate::wire::Envelope;
use parking_lot::Mutex;
use std::path::PathBuf;
Expand Down Expand Up @@ -70,17 +66,17 @@ pub struct Session {

impl Session {
/// Spawn a session's actor task and return its handle.
pub fn spawn(
options: SessionOptions,
_translators: Arc<Registry>,
sink_factory: Arc<dyn SinkFactory>,
) -> Arc<Session> {
pub fn spawn(options: SessionOptions, sink_factory: Arc<dyn SinkFactory>) -> Arc<Session> {
let (tx, mut rx) = mpsc::channel(QUEUE_CAPACITY);
let counters = Arc::new(Counters::default());
let recovery = Arc::new(Mutex::new(None));
let last_error = Arc::new(Mutex::new(None));
let permalink = Arc::new(Mutex::new(None));
let source = options.source.clone();
let observer_key = format!(
"{}\u{1f}{}\u{1f}{}",
options.source, options.session_id, options.cursor.route
);
let route = Consumer::spawn(
options.translation.journal.clone(),
options.cursor,
Expand All @@ -103,6 +99,7 @@ impl Session {
let source_recovery = producer.recovery.clone();
let source_progress = producer.progress.clone();
let errors = last_error.clone();
let source_errors = source_error.clone();
let queued = counters.clone();
tokio::spawn(async move {
while let Some(message) = rx.recv().await {
Expand All @@ -113,7 +110,7 @@ impl Session {
route.configure(config);
}
if let Err(error) = producer.advance(through).await {
*errors.lock() = Some(error.to_string());
*source_errors.lock() = Some(error.to_string());
}
}
SessionMsg::Configure(config, reply) => {
Expand All @@ -134,6 +131,7 @@ impl Session {
let route = route.clone();
let queued = queued.clone();
let errors = errors.clone();
let source_errors = source_errors.clone();
tokio::spawn(async move {
let mut reply = reply;
let waiting = async {
Expand All @@ -146,7 +144,7 @@ impl Session {
}
},
Err(error) => {
*errors.lock() = Some(error);
*source_errors.lock() = Some(error);
queued.queued.load(Ordering::Relaxed).max(1)
}
}
Expand All @@ -171,6 +169,7 @@ impl Session {
)
.await;
}
producer.unobserve(observer_key.clone()).await;
route.shutdown().await;
if producer.routes.load(Ordering::Relaxed) == 0 {
producer.shutdown().await;
Expand Down
49 changes: 48 additions & 1 deletion bt-daemon/src/journal.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -599,14 +599,61 @@ pub async fn gc_old_journals(data_dir: &Path, max_age: std::time::Duration) {
}
None => false,
};
if old && !protected {
if old && !protected && legacy_collectible(data_dir, &path).await {
if let Err(e) = tokio::fs::remove_file(&path).await {
tracing::warn!(path = %path.display(), "failed to remove stale journal: {e}");
}
}
}
}

// Pre-SpanOp journals can be collected only with explicit completion evidence.
// Unknown or uncheckpointed history stays available for migration/reconstruction.
async fn legacy_collectible(data_dir: &Path, path: &Path) -> bool {
let Ok(Some(mut reader)) =
JournalReader::open(path, JournalReader::recorded_len(path).await).await
else {
return false;
};
let mut observed = std::collections::HashMap::<String, u64>::new();
let mut completed = std::collections::HashMap::<String, u64>::new();
loop {
match reader.next_record().await {
Ok(Some(entry)) => match entry.record {
JournalRecord::Event(event) => {
let partition = data_dir.join("pipeline").join(format!(
"{}.sqlite",
crate::ids::session_storage_id(&event.source, &event.session_id)
));
if tokio::fs::try_exists(partition).await.unwrap_or(true) {
return false;
}
let Some(route) = event.route else {
return false;
};
observed.insert(
serde_json::to_string(&route).unwrap_or_default(),
entry.through,
);
}
JournalRecord::DeliveryCheckpoint { route, through } => {
completed
.entry(serde_json::to_string(&route).unwrap_or_default())
.and_modify(|previous| *previous = (*previous).max(through))
.or_insert(through);
}
JournalRecord::Control(_) => return false,
},
Ok(None) => break,
Err(_) => return false,
}
}
!observed.is_empty()
&& observed
.iter()
.all(|(route, through)| completed.get(route).copied().unwrap_or(0) >= *through)
}

/// Reconstruct a translator-usable [`Envelope`] from a redacted journal entry.
/// The live token is gone (redacted), so `auth.token` is empty — fine for
/// rebuilding translator state; the sink must be re-supplied live credentials
Expand Down
Loading
Loading