From 6a1411adfa58960124982f98816e380e027b3822 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Tue, 13 Oct 2026 23:16:51 +0800 Subject: [PATCH 1/3] Isolate routing state behind a bounded actor port --- bt-daemon/src/dispatch.rs | 63 +++-------- bt-daemon/src/pipeline/delivery.rs | 37 +++++- bt-daemon/src/pipeline/mod.rs | 1 + bt-daemon/src/pipeline/routing.rs | 175 +++++++++++++++++++++++++++++ 4 files changed, 229 insertions(+), 47 deletions(-) create mode 100644 bt-daemon/src/pipeline/routing.rs diff --git a/bt-daemon/src/dispatch.rs b/bt-daemon/src/dispatch.rs index 55896cd9..6224d45b 100644 --- a/bt-daemon/src/dispatch.rs +++ b/bt-daemon/src/dispatch.rs @@ -9,8 +9,8 @@ use crate::delivery_ledger::LedgerSink; use crate::journal::JournalWriter; -use crate::pipeline::plugins::{PluginOutcome, Plugins}; -use crate::pipeline::{Completion, Delivery, Stage}; +use crate::pipeline::routing::RoutingSink; +use crate::pipeline::{Completion, Delivery}; use crate::sink::SinkFactory; use crate::translate::{Registry, SessionCtx}; use crate::wire::{Envelope, SessionRoute}; @@ -545,7 +545,7 @@ impl SessionActor { return; } }; - let mut sink: Box = Box::new( + let sink: Box = Box::new( LedgerSink::new( sink, Some(&self.data_dir), @@ -556,6 +556,13 @@ impl SessionActor { ) .await, ); + let mut sink: Box = Box::new(RoutingSink::spawn( + sink, + self.source.clone(), + self.session_id.clone(), + self.data_dir.clone(), + self.last_error.clone(), + )); let mut ctx = SessionCtx { session_id: self.session_id.clone(), config: Some(self.config.clone()), @@ -724,48 +731,8 @@ impl SessionActor { self.persist_correlation_if_changed(true).await; } } - let plugin_paths = ctx - .config - .as_ref() - .map(|config| config.span_plugins.as_slice()) - .unwrap_or_default(); - let mut stage = Plugins { - paths: plugin_paths.to_vec(), - source: self.source.clone(), - session_id: self.session_id.clone(), - }; - let mut processed = Vec::with_capacity(ops.len()); - for op in &ops { - match stage.process(op).await { - PluginOutcome::Prepared(op) => processed.push(*op), - PluginOutcome::Filtered => {} - PluginOutcome::Failed(failure) => { - delivered = false; - if failure.newly_seen { - if let Err(error) = crate::plugin_diagnostics::record( - &self.data_dir, - &self.source, - &failure.path, - &failure.message, - ) { - tracing::warn!(session_id = %self.session_id, - "failed to persist span plugin diagnostic: {error}"); - } - self.set_error(format!( - "span plugin {} failed; span operations are being discarded: {}", - failure.path.display(), - failure.message - )); - } - } - PluginOutcome::Unavailable(error) => { - delivered = false; - self.set_error(format!("span plugin processor failed: {error}")); - } - } - } - if !processed.is_empty() { - match sink.emit(&processed).await { + if !ops.is_empty() { + match sink.emit(&ops).await { Ok(n) => { self.counters.spans_emitted.fetch_add(n, Ordering::Relaxed); } @@ -964,7 +931,11 @@ impl SessionActor { ); match pending.into_completed().await { Ok(through) => *acknowledged_through = through, - Err(failed) => self.set_error(format!("delivery checkpoint failed: {}", failed.error)), + Err(failed) => self.set_error(format!( + "delivery checkpoint at {} failed: {}", + failed.delivery.value(), + failed.error + )), } true } diff --git a/bt-daemon/src/pipeline/delivery.rs b/bt-daemon/src/pipeline/delivery.rs index d6cd35cb..da9371cf 100644 --- a/bt-daemon/src/pipeline/delivery.rs +++ b/bt-daemon/src/pipeline/delivery.rs @@ -1,3 +1,5 @@ +use tokio::sync::mpsc; + /// A message owns its completion obligation until terminal handling succeeds. pub(crate) struct Delivery { value: T, @@ -25,7 +27,6 @@ impl Delivery { } } - #[cfg_attr(not(test), allow(dead_code))] pub fn value(&self) -> &T { &self.value } @@ -49,6 +50,40 @@ impl Delivery { } } +// async_trait adds a redundant must_use annotation to its boxed futures. +#[allow(clippy::double_must_use)] +#[async_trait::async_trait] +pub(crate) trait InputPort: Send { + type Item: Send; + async fn receive(&mut self) -> Option>; +} + +// async_trait adds a redundant must_use annotation to its boxed futures. +#[allow(clippy::double_must_use)] +#[async_trait::async_trait] +pub(crate) trait OutputPort: Send { + type Item: Send; + async fn send(&mut self, delivery: Delivery) -> Result<(), Delivery>; +} + +#[async_trait::async_trait] +impl InputPort for mpsc::Receiver> { + type Item = T; + async fn receive(&mut self) -> Option> { + self.recv().await + } +} + +#[async_trait::async_trait] +impl OutputPort for mpsc::Sender> { + type Item = T; + async fn send(&mut self, delivery: Delivery) -> Result<(), Delivery> { + mpsc::Sender::send(self, delivery) + .await + .map_err(|error| error.0) + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/bt-daemon/src/pipeline/mod.rs b/bt-daemon/src/pipeline/mod.rs index 6180279e..3adefa05 100644 --- a/bt-daemon/src/pipeline/mod.rs +++ b/bt-daemon/src/pipeline/mod.rs @@ -2,6 +2,7 @@ pub(crate) mod delivery; pub(crate) mod plugins; +pub(crate) mod routing; pub(crate) use delivery::{Completion, Delivery}; diff --git a/bt-daemon/src/pipeline/routing.rs b/bt-daemon/src/pipeline/routing.rs new file mode 100644 index 00000000..22806849 --- /dev/null +++ b/bt-daemon/src/pipeline/routing.rs @@ -0,0 +1,175 @@ +//! Routing owns its sink behind a bounded typed port. Session dispatch is its facade. +use super::delivery::{Delivery, InputPort, OutputPort}; +use super::plugins::{PluginOutcome, Plugins}; +use super::{Completion, Stage}; +use crate::{sink::Sink, translate::SpanOp, wire::SessionConfig}; +use std::{ + path::PathBuf, + sync::{Arc, Mutex}, +}; +use tokio::sync::{mpsc, oneshot}; + +struct Request { + config: SessionConfig, + versions: (Option, Option), + operation: Operation, + reply: oneshot::Sender, +} + +enum Operation { + Emit(Vec), + Flush, +} +struct Response { + result: anyhow::Result, + pending: bool, + permalink: Option, +} + +struct Forward; +#[async_trait::async_trait] +impl Completion for Forward { + async fn complete(&mut self) -> anyhow::Result<()> { + Ok(()) + } +} + +pub(crate) struct RoutingSink { + tx: mpsc::Sender>, + config: Option, + versions: (Option, Option), + pending: bool, + permalink: Option, +} + +impl RoutingSink { + pub fn spawn( + mut sink: Box, + source: String, + session_id: String, + data_dir: PathBuf, + last_error: Arc>>, + ) -> Self { + let (tx, mut rx) = mpsc::channel::>(16); + tokio::spawn(async move { + while let Some(delivery) = InputPort::receive(&mut rx).await { + let request = delivery.value(); + sink.configure(&request.config); + sink.set_capture_versions( + request.versions.0.as_deref(), + request.versions.1.as_deref(), + ); + let result = match &request.operation { + Operation::Emit(ops) => { + let mut stage = Plugins { + paths: request.config.span_plugins.clone(), + source: source.clone(), + session_id: session_id.clone(), + }; + let mut prepared = Vec::with_capacity(ops.len()); + let mut failure = None; + for op in ops { + match stage.process(op).await { + PluginOutcome::Prepared(op) => prepared.push(*op), + PluginOutcome::Filtered => {} + PluginOutcome::Failed(error) => { + if error.newly_seen { + if let Err(diagnostic) = crate::plugin_diagnostics::record( + &data_dir, + &source, + &error.path, + &error.message, + ) { + tracing::warn!(%diagnostic, "plugin diagnostic could not be saved"); + } + } + failure = Some(anyhow::anyhow!( + "span plugin {} failed; span operations are being discarded: {}", + error.path.display(), error.message)); + } + PluginOutcome::Unavailable(error) => failure = Some(error), + } + } + let emitted = if prepared.is_empty() { + Ok(0) + } else { + sink.emit(&prepared).await + }; + match failure { + Some(error) => Err(error), + None => emitted, + } + } + Operation::Flush => sink.flush().await.map(|()| 0), + }; + if let Err(error) = &result { + *last_error.lock().unwrap() = Some(error.to_string()); + } + let request = delivery + .into_completed() + .await + .ok() + .expect("volatile delivery"); + let _ = request.reply.send(Response { + result, + pending: sink.has_pending_delivery(), + permalink: sink.permalink(), + }); + } + }); + Self { + tx, + config: None, + versions: (None, None), + pending: false, + permalink: None, + } + } + + async fn request(&mut self, operation: Operation) -> anyhow::Result { + let config = self + .config + .clone() + .ok_or_else(|| anyhow::anyhow!("route is not configured"))?; + let (reply, receive) = oneshot::channel(); + let delivery = Delivery::new(operation, Forward).map(|operation| Request { + config, + versions: self.versions.clone(), + operation, + reply, + }); + // Retain ownership on a failed send; no upstream completion is implied. + if let Err(undelivered) = OutputPort::send(&mut self.tx, delivery).await { + let _ = undelivered.value(); + anyhow::bail!("routing actor is gone"); + } + let response = receive + .await + .map_err(|_| anyhow::anyhow!("routing actor dropped its reply"))?; + self.pending = response.pending; + self.permalink = response.permalink; + response.result + } +} + +#[async_trait::async_trait] +impl Sink for RoutingSink { + fn configure(&mut self, config: &SessionConfig) { + self.config = Some(config.clone()); + } + fn set_capture_versions(&mut self, plugin: Option<&str>, source: Option<&str>) { + self.versions = (plugin.map(str::to_owned), source.map(str::to_owned)); + } + async fn emit(&mut self, ops: &[SpanOp]) -> anyhow::Result { + self.request(Operation::Emit(ops.to_vec())).await + } + async fn flush(&mut self) -> anyhow::Result<()> { + self.request(Operation::Flush).await.map(|_| ()) + } + fn has_pending_delivery(&self) -> bool { + self.pending + } + fn permalink(&self) -> Option { + self.permalink.clone() + } +} From b59815c348c0524fad7ee1cfd4ac521f2aab6f93 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Wed, 14 Oct 2026 01:13:52 +0800 Subject: [PATCH 2/3] Preserve partial routing counts and report cached plugin failures once --- bt-daemon/src/dispatch.rs | 13 +++- bt-daemon/src/pipeline/routing.rs | 109 ++++++++++++++++++++++++++++-- bt-daemon/tests/pipeline.rs | 36 ++++++++++ 3 files changed, 153 insertions(+), 5 deletions(-) diff --git a/bt-daemon/src/dispatch.rs b/bt-daemon/src/dispatch.rs index 6224d45b..08d01e05 100644 --- a/bt-daemon/src/dispatch.rs +++ b/bt-daemon/src/dispatch.rs @@ -738,7 +738,18 @@ impl SessionActor { } Err(e) => { delivered = false; - self.set_error(format!("{emit_error}: {e}")); + if let Some(failure) = + e.downcast_ref::() + { + self.counters + .spans_emitted + .fetch_add(failure.emitted, Ordering::Relaxed); + if failure.report { + self.set_error(format!("{emit_error}: {e}")); + } + } else { + self.set_error(format!("{emit_error}: {e}")); + } } } } diff --git a/bt-daemon/src/pipeline/routing.rs b/bt-daemon/src/pipeline/routing.rs index 22806849..191c4d94 100644 --- a/bt-daemon/src/pipeline/routing.rs +++ b/bt-daemon/src/pipeline/routing.rs @@ -26,6 +26,24 @@ struct Response { permalink: Option, } +/// Rows accepted before a mandatory plugin failed still count as emitted. +#[derive(Debug)] +pub(crate) struct EmitFailure { + pub emitted: u64, + pub report: bool, + pub cause: anyhow::Error, +} +impl std::fmt::Display for EmitFailure { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + self.cause.fmt(f) + } +} +impl std::error::Error for EmitFailure { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + Some(self.cause.as_ref()) + } +} + struct Forward; #[async_trait::async_trait] impl Completion for Forward { @@ -68,11 +86,13 @@ impl RoutingSink { }; let mut prepared = Vec::with_capacity(ops.len()); let mut failure = None; + let mut report = false; for op in ops { match stage.process(op).await { PluginOutcome::Prepared(op) => prepared.push(*op), PluginOutcome::Filtered => {} PluginOutcome::Failed(error) => { + report |= error.newly_seen; if error.newly_seen { if let Err(diagnostic) = crate::plugin_diagnostics::record( &data_dir, @@ -87,7 +107,10 @@ impl RoutingSink { "span plugin {} failed; span operations are being discarded: {}", error.path.display(), error.message)); } - PluginOutcome::Unavailable(error) => failure = Some(error), + PluginOutcome::Unavailable(error) => { + report = true; + failure = Some(error); + } } } let emitted = if prepared.is_empty() { @@ -95,9 +118,15 @@ impl RoutingSink { } else { sink.emit(&prepared).await }; - match failure { - Some(error) => Err(error), - None => emitted, + match (emitted, failure) { + (Err(error), _) => Err(error), + (Ok(emitted), Some(cause)) => Err(EmitFailure { + emitted, + report, + cause, + } + .into()), + (Ok(emitted), None) => Ok(emitted), } } Operation::Flush => sink.flush().await.map(|()| 0), @@ -173,3 +202,75 @@ impl Sink for RoutingSink { self.permalink.clone() } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ + translate::SpanRow, + wire::{BackendAuth, SessionRoute}, + }; + use std::sync::atomic::{AtomicU64, Ordering}; + + struct CountingSink(Arc); + #[async_trait::async_trait] + impl Sink for CountingSink { + async fn emit(&mut self, ops: &[SpanOp]) -> anyhow::Result { + self.0.fetch_add(ops.len() as u64, Ordering::Relaxed); + Ok(ops.len() as u64) + } + async fn flush(&mut self) -> anyhow::Result<()> { + Ok(()) + } + } + + #[tokio::test] + async fn partial_plugin_failure_preserves_count_and_reports_only_once() { + let tmp = tempfile::tempdir().unwrap(); + let plugin = tmp.path().join("selective.mjs"); + std::fs::write(&plugin, "export default span => { if (span.name === 'reject') throw new Error('rejected'); return span; }").unwrap(); + let emitted = Arc::new(AtomicU64::new(0)); + let error = Arc::new(Mutex::new(None)); + let mut sink = RoutingSink::spawn( + Box::new(CountingSink(emitted.clone())), + "debug".into(), + "partial".into(), + tmp.path().into(), + error.clone(), + ); + sink.configure( + &SessionRoute { + span_plugins: vec![plugin], + ..SessionRoute::default() + } + .with_auth(BackendAuth { + token: "test".into(), + api_url: None, + app_url: None, + org_name: None, + org_id: None, + }), + ); + let op = |name: &str| { + SpanOp::Insert(SpanRow { + span_id: name.into(), + name: name.into(), + ..SpanRow::default() + }) + }; + let failure = sink.emit(&[op("accept"), op("reject")]).await.unwrap_err(); + let first = failure.downcast_ref::().unwrap(); + assert_eq!(first.emitted, 1); + assert!(first.report); + assert_eq!(emitted.load(Ordering::Relaxed), 1); + let failure = sink.emit(&[op("accept")]).await.unwrap_err(); + let cached = failure.downcast_ref::().unwrap(); + assert_eq!(cached.emitted, 0); + assert!( + !cached.report, + "cached failures must not request another warning" + ); + assert!(error.lock().unwrap().as_ref().unwrap().contains("rejected")); + assert_eq!(emitted.load(Ordering::Relaxed), 1); + } +} diff --git a/bt-daemon/tests/pipeline.rs b/bt-daemon/tests/pipeline.rs index 06f81e45..c284d3d8 100644 --- a/bt-daemon/tests/pipeline.rs +++ b/bt-daemon/tests/pipeline.rs @@ -3292,3 +3292,39 @@ async fn packaged_grok_hook_replays_bounded_transcripts_to_isolated_debug_routes shutdown(&socket).await; second.await.unwrap(); } + +#[tokio::test] +async fn status_counts_rows_emitted_before_a_plugin_failure() { + let (data_dir, socket, handle, tmp) = start_daemon().await; + let plugin = tmp.path().join("partial.mjs"); + std::fs::write(&plugin, "export default span => { if (span.name === 'SessionStart') throw new Error('rejected event'); return span; }").unwrap(); + let mut env = envelope("partial-plugin", "SessionStart", 1); + env.route.as_mut().unwrap().span_plugins = vec![plugin]; + forward_envelope(&env, &socket, &dummy_host(), false) + .await + .unwrap(); + flush_session("partial-plugin", &socket, 5_000) + .await + .unwrap(); + let status = run_status(StatusArgs { + socket: Some(socket.clone()), + session_id: Some("partial-plugin".into()), + }) + .await + .unwrap() + .unwrap(); + let rows = std::fs::read_to_string(data_dir.join("spans/partial-plugin.ndjson")).unwrap(); + assert_eq!( + rows.lines().count(), + 1, + "the root was emitted before the plugin rejected its event" + ); + assert_eq!(status.sessions[0].spans_emitted, 1); + assert!(status.sessions[0] + .last_error + .as_ref() + .unwrap() + .contains("rejected event")); + shutdown(&socket).await; + handle.await.unwrap(); +} From cb30038ffcfd7b10edad71c3fd2b243b325057e5 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Fri, 16 Oct 2026 22:24:29 +0800 Subject: [PATCH 3/3] Align routing actor with panic-free bookkeeping --- bt-daemon/src/pipeline/routing.rs | 25 ++++++++++++++----------- 1 file changed, 14 insertions(+), 11 deletions(-) diff --git a/bt-daemon/src/pipeline/routing.rs b/bt-daemon/src/pipeline/routing.rs index 191c4d94..ffb65498 100644 --- a/bt-daemon/src/pipeline/routing.rs +++ b/bt-daemon/src/pipeline/routing.rs @@ -3,10 +3,8 @@ use super::delivery::{Delivery, InputPort, OutputPort}; use super::plugins::{PluginOutcome, Plugins}; use super::{Completion, Stage}; use crate::{sink::Sink, translate::SpanOp, wire::SessionConfig}; -use std::{ - path::PathBuf, - sync::{Arc, Mutex}, -}; +use parking_lot::Mutex; +use std::{path::PathBuf, sync::Arc}; use tokio::sync::{mpsc, oneshot}; struct Request { @@ -132,13 +130,18 @@ impl RoutingSink { Operation::Flush => sink.flush().await.map(|()| 0), }; if let Err(error) = &result { - *last_error.lock().unwrap() = Some(error.to_string()); + *last_error.lock() = Some(error.to_string()); } - let request = delivery - .into_completed() - .await - .ok() - .expect("volatile delivery"); + let request = match delivery.into_completed().await { + Ok(request) => request, + Err(failed) => { + *last_error.lock() = + Some(format!("routing completion failed: {}", failed.error)); + // Dropping the unfinished request closes its reply; the + // caller receives an error instead of a success response. + continue; + } + }; let _ = request.reply.send(Response { result, pending: sink.has_pending_delivery(), @@ -270,7 +273,7 @@ mod tests { !cached.report, "cached failures must not request another warning" ); - assert!(error.lock().unwrap().as_ref().unwrap().contains("rejected")); + assert!(error.lock().as_ref().unwrap().contains("rejected")); assert_eq!(emitted.load(Ordering::Relaxed), 1); } }