From 0e488418ab27f4f74439c573854934545a37b10f Mon Sep 17 00:00:00 2001 From: Snow Pettersen Date: Fri, 7 Aug 2026 08:50:50 -0400 Subject: [PATCH 01/23] bd-workflows: implement type-based routing within the workflow engine Instead of always evaluating all workflows for all logs, this tracks a running set of candidate workflows based on the log type of the log. This allows us to bypass workflows whose next match transitions require a different log type than the inbound log --- bd-log-matcher/src/matcher.rs | 106 ++++++++++ bd-log-matcher/src/matcher_test.rs | 61 +++++- bd-workflows/src/config.rs | 16 ++ bd-workflows/src/engine.rs | 264 +++++++++++++++++------- bd-workflows/src/engine_test.rs | 92 ++++++++- bd-workflows/src/engine_test_helpers.rs | 3 +- bd-workflows/src/lib.rs | 1 + bd-workflows/src/routing.rs | 216 +++++++++++++++++++ bd-workflows/src/routing_test.rs | 72 +++++++ bd-workflows/src/test.rs | 9 + bd-workflows/src/workflow.rs | 80 +++++++ 11 files changed, 841 insertions(+), 79 deletions(-) create mode 100644 bd-workflows/src/routing.rs create mode 100644 bd-workflows/src/routing_test.rs diff --git a/bd-log-matcher/src/matcher.rs b/bd-log-matcher/src/matcher.rs index ed35778e3..a3cf906b5 100644 --- a/bd-log-matcher/src/matcher.rs +++ b/bd-log-matcher/src/matcher.rs @@ -47,6 +47,7 @@ use bd_proto::protos::value_matcher::value_matcher::json_path_value_match::{ use bd_state::Scope; use log_matcher::LogMatcher; use log_matcher::log_matcher::{BaseLogMatcher, Matcher, base_log_matcher}; +use protobuf::Enum; use rand::RngExt; use std::borrow::Cow; @@ -54,6 +55,67 @@ const LOG_LEVEL_KEY: &str = "log_level"; const LOG_TYPE_KEY: &str = "log_type"; pub const SAMPLE_RATE_DENOMINATOR: u32 = 1_000_000; +// +// LogTypeSet +// + +/// An allocation-free bitset of SDK log types. +/// +/// Values outside the current `LogType` enum are represented by an empty set because the runtime +/// cannot receive them. Keeping the bitmask behind this type makes the enum-to-bit conversion +/// explicit without exposing numeric masks to routing code. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct LogTypeSet(u16); + +impl LogTypeSet { + #[must_use] + pub fn from_log_type(log_type: LogType) -> Self { + Self( + u32::try_from(log_type.value()) + .ok() + .and_then(|value| u16::checked_shl(1, value)) + .unwrap_or_default(), + ) + } + + fn from_log_type_value(log_type: u32) -> Self { + i32::try_from(log_type) + .ok() + .and_then(LogType::from_i32) + .map_or_else(Self::default, Self::from_log_type) + } + + #[must_use] + pub fn is_empty(self) -> bool { + self.0 == 0 + } + + pub fn union(&mut self, other: Self) { + self.0 |= other.0; + } + + pub fn intersect(&mut self, other: Self) { + self.0 &= other.0; + } + + pub fn iter(self) -> impl Iterator { + let mut bits = self.0; + std::iter::from_fn(move || { + loop { + let value = bits.trailing_zeros(); + if value >= u16::BITS { + return None; + } + + bits &= bits - 1; + if let Some(log_type) = i32::try_from(value).ok().and_then(LogType::from_i32) { + return Some(log_type); + } + } + }) + } +} + pub trait RandomNumberGenerator { fn random_u32(&mut self, upper_bound_exclusive: u32) -> u32; } @@ -175,6 +237,50 @@ impl Tree { ) } + /// Returns a safe upper bound on the log types this tree can match without evaluating a log. + /// + /// `Some(types)` means no log outside `types` can match, but a log in `types` still needs normal + /// matcher evaluation. `None` means the tree's log-type restriction cannot be proven, so a + /// router must use its fallback path and evaluate every log. + /// + /// This is deliberately conservative: returning a narrower set than the matcher can accept + /// would incorrectly skip a workflow. + #[must_use] + pub fn possible_log_types(&self) -> Option { + match self { + Self::Base(Leaf::LogType(log_type)) => Some(LogTypeSet::from_log_type_value(*log_type)), + // Base leaves without an exact type predicate may match any type. A negated expression can + // exclude a type, but its other conditions can still make that type match. Keep both on the + // fallback path rather than attempting to infer exclusions. + Self::Base(_) | Self::Not(_) => None, + Self::Or(matchers) => { + let mut log_types = LogTypeSet::default(); + for matcher in matchers { + // Every OR branch needs a known upper bound. One unrestricted branch makes the whole + // expression unrestricted. + log_types.union(matcher.possible_log_types()?); + } + Some(log_types) + }, + Self::And(matchers) => { + let mut log_types = None; + for matcher in matchers { + if let Some(matcher_log_types) = matcher.possible_log_types() { + // An unrestricted AND branch adds no type information; intersect only the known + // restrictions. + log_types = Some( + log_types.map_or(matcher_log_types, |mut log_types: LogTypeSet| { + log_types.intersect(matcher_log_types); + log_types + }), + ); + } + } + log_types + }, + } + } + #[must_use] pub fn do_match_with_rng( &self, diff --git a/bd-log-matcher/src/matcher_test.rs b/bd-log-matcher/src/matcher_test.rs index c0765e61a..85c1f625f 100644 --- a/bd-log-matcher/src/matcher_test.rs +++ b/bd-log-matcher/src/matcher_test.rs @@ -7,7 +7,7 @@ use crate::builder; use crate::matcher::base_log_matcher::tag_match::Value_match::DoubleValueMatch; -use crate::matcher::{RandomNumberGenerator, Tree}; +use crate::matcher::{LogTypeSet, RandomNumberGenerator, Tree}; use crate::test::TestMatcher; use ahash::AHashMap; use bd_log_primitives::tiny_set::TinyMap; @@ -1840,3 +1840,62 @@ fn double_matcher_with_u64_field() { ], ); } + +#[test] +fn log_type_set_iterates_set_bits_and_supports_all_protobuf_types() { + let mut log_types = LogTypeSet::from_log_type(LogType::RESOURCE); + log_types.union(LogTypeSet::from_log_type(LogType::LIFECYCLE)); + + assert_eq!( + vec![LogType::LIFECYCLE, LogType::RESOURCE], + log_types.iter().collect::>() + ); + assert!( + LogType::VALUES + .iter() + .all(|log_type| { u32::try_from(log_type.value()).is_ok_and(|value| value < u16::BITS) }) + ); +} + +#[test] +fn possible_log_types_are_conservative() { + let lifecycle = builder::log_type_equals(LogType::LIFECYCLE); + let resource = builder::log_type_equals(LogType::RESOURCE); + + assert_eq!( + Some(LogTypeSet::from_log_type(LogType::LIFECYCLE)), + Tree::new(&lifecycle).unwrap().possible_log_types() + ); + assert_eq!( + Some(LogTypeSet::from_log_type(LogType::RESOURCE)), + Tree::new(&builder::and(vec![ + resource.clone(), + builder::message_equals("resource") + ])) + .unwrap() + .possible_log_types() + ); + let mut lifecycle_or_resource = LogTypeSet::from_log_type(LogType::LIFECYCLE); + lifecycle_or_resource.union(LogTypeSet::from_log_type(LogType::RESOURCE)); + assert_eq!( + Some(lifecycle_or_resource), + Tree::new(&builder::or(vec![lifecycle.clone(), resource])) + .unwrap() + .possible_log_types() + ); + assert_eq!( + None, + Tree::new(&builder::or(vec![ + lifecycle.clone(), + builder::message_equals("fallback") + ])) + .unwrap() + .possible_log_types() + ); + assert_eq!( + None, + Tree::new(&builder::not(lifecycle)) + .unwrap() + .possible_log_types() + ); +} diff --git a/bd-workflows/src/config.rs b/bd-workflows/src/config.rs index 75cf39370..cc2622bcb 100644 --- a/bd-workflows/src/config.rs +++ b/bd-workflows/src/config.rs @@ -217,6 +217,10 @@ impl Config { &self.inner } + pub(crate) fn possible_log_types(&self) -> bd_log_matcher::matcher::LogTypeSet { + self.inner.possible_log_types() + } + pub(crate) const fn mode(&self) -> WorkflowDebugMode { self.mode } @@ -327,6 +331,18 @@ impl InnerConfig { .map(|state| state.transitions.as_slice()) } + pub(crate) fn possible_log_types(&self) -> bd_log_matcher::matcher::LogTypeSet { + let mut log_types = bd_log_matcher::matcher::LogTypeSet::default(); + for transition in self.states.iter().flat_map(State::transitions) { + if let Predicate::LogMatch { matcher, .. } = transition.rule() + && let Some(matcher_log_types) = matcher.possible_log_types() + { + log_types.union(matcher_log_types); + } + } + log_types + } + pub(crate) fn actions_for_traversal( &self, traversal: &Traversal, diff --git a/bd-workflows/src/engine.rs b/bd-workflows/src/engine.rs index a549dd819..ccde9750c 100644 --- a/bd-workflows/src/engine.rs +++ b/bd-workflows/src/engine.rs @@ -36,6 +36,7 @@ use crate::config::{ WorkflowsConfiguration, }; use crate::metrics::MetricsCollector; +use crate::routing::WorkflowLogRouter; use crate::sankey_diagram::{self, PendingSankeyPathUpload}; use crate::workflow::{ SankeyPath, @@ -120,6 +121,7 @@ pub struct WorkflowsEngine { // at index `i`. configs: Vec, state: WorkflowsState, + log_router: WorkflowLogRouter, // Tracks the immediately preceding session for detecting out-of-order logs that return to it. // This is process local as the most relevant case of this is during startup when sequencing // crash logs that occurred in a previous session. @@ -219,6 +221,7 @@ impl WorkflowsEngine { let workflows_engine = Self { configs: vec![], state: WorkflowsState::default(), + log_router: WorkflowLogRouter::default(), previous_session_id: String::new(), stats: WorkflowsEngineStats::new(&scope), state_store, @@ -339,6 +342,7 @@ impl WorkflowsEngine { } self.state.refresh_tracing_counts_from_state(); + Self::rebuild_log_routes(&mut self.log_router, &self.state.workflows, &self.configs); log::debug!( "started workflows engine with {} workflow(s); {} pending processing action(s); {} pending \ @@ -466,6 +470,8 @@ impl WorkflowsEngine { self.state.streaming_actions = self .flush_buffers_actions_resolver .standardize_streaming_buffers(self.state.streaming_actions.clone()); + // Regenerate routes whenever the workflow configuration is reloaded. + Self::rebuild_log_routes(&mut self.log_router, &self.state.workflows, &self.configs); log::debug!( "consumed received workflows config update; workflows engine contains {} workflow(s)", @@ -548,6 +554,107 @@ impl WorkflowsEngine { } } + #[allow(clippy::too_many_arguments)] + fn process_workflow_event<'config>( + configs: &'config [Config], + workflows: &mut [Workflow], + stats: &WorkflowsEngineStats, + needs_state_persistence: &mut bool, + active_run_tracing_count: &mut u32, + index: usize, + event: WorkflowEvent<'_>, + state_reader: &dyn bd_state::StateReader, + now: OffsetDateTime, + sampled_roll: u32, + output: &mut WorkflowEventOutput<'config>, + ) { + let Some(config) = configs.get(index) else { + return; + }; + let Some(workflow) = workflows.get_mut(index) else { + return; + }; + + let was_in_initial_state = workflow.is_in_initial_state(); + let result = workflow.process_event(config, event, state_reader, now, sampled_roll); + + if result.stats().matched_logs_count > 0 { + stats + .matched_logs_total + .inc_by(u64::from(result.stats().matched_logs_count)); + } + + *active_run_tracing_count = active_run_tracing_count + .saturating_add(result.stats().tracing_starts) + .saturating_sub(result.stats().tracing_ends); + + // Not every case of a workflow making a progress needs a state persistence. + // If the workflow was in an initial state prior to processing a log and is in + // an initial state after processing the log then the state of workflow did not change + // as the result of processing a log and does not have to be persisted. An example for when + // a workflow makes progress but does not needs persistence is the following workflow + // with 2 nodes/states - a start log matching node and a final emit metric node, + // such workflow may: + // * be in an initial state + // * match a single log + // * execute emit metric action + // * be an initial state + if result.stats().did_make_progress() + && !(was_in_initial_state && workflow.is_in_initial_state()) + { + *needs_state_persistence = true; + } + + // In debug only mode we do not trigger any actions, but we still inject logs so that + // workflows continue to advance if they depend on the injected logs. + let ( + triggered_actions, + workflow_logs_to_inject, + cumulative_workflow_debug_state, + incremental_workflow_debug_state, + workflow_tracing_carryover_flush_action_ids, + ) = result.into_parts(); + if !matches!(config.mode(), WorkflowDebugMode::DebugOnly) { + output + .prepared_actions + .incorporate_workflow_actions(index, triggered_actions); + } + output.logs_to_inject.extend(workflow_logs_to_inject); + if let Some(cumulative_workflow_debug_state) = cumulative_workflow_debug_state { + output + .all_cumulative_workflow_debug_state + .push((workflow.id().to_string(), cumulative_workflow_debug_state)); + } + output.all_incremental_workflow_debug_state.extend( + incremental_workflow_debug_state + .into_iter() + .map(|state_key| WorkflowDebugKey { + workflow_id: workflow.id().to_string(), + state_key, + }), + ); + output + .tracing_carryover_flush_action_ids + .extend(workflow_tracing_carryover_flush_action_ids); + } + + /// Rebuilds routes after an event or configuration change that can affect every workflow. + fn rebuild_log_routes( + log_router: &mut WorkflowLogRouter, + workflows: &[Workflow], + configs: &[Config], + ) { + let mut known_log_types = bd_log_matcher::matcher::LogTypeSet::default(); + for config in configs { + known_log_types.union(config.possible_log_types()); + } + log_router.prepare(workflows.len(), known_log_types); + + for (index, (workflow, config)) in workflows.iter().zip(configs).enumerate() { + log_router.insert(index, workflow.log_route(config)); + } + } + /// Attempts to persist the client-side state of the workflows to disk while ignoring any errors. /// Does nothing if state has not changed since the last time it was saved. /// @@ -796,88 +903,86 @@ impl WorkflowsEngine { }; } - let mut prepared_actions = PreparedActions::default(); - let mut logs_to_inject: TinyMap<&'a str, Log> = TinyMap::default(); - let mut all_cumulative_workflow_debug_state = vec![]; - let mut all_incremental_workflow_debug_state = vec![]; - let mut tracing_carryover_flush_action_ids: TinySet = TinySet::default(); + let mut workflow_event_output = WorkflowEventOutput::default(); // Sampling decisions should be stable for a single event across every workflow/run that // evaluates it. Roll once here and thread the same value through matcher evaluation. let sampled_roll = bd_log_matcher::matcher::random_sample_roll(); - let mut has_debug_workflows = false; - for (index, workflow) in &mut self.state.workflows.iter_mut().enumerate() { - let Some(config) = self.configs.get(index) else { - continue; - }; - - let was_in_initial_state = workflow.is_in_initial_state(); - let result = workflow.process_event(config, event, state_reader, now, sampled_roll); - - macro_rules! inc_by { - ($field:ident, $value:ident) => { - self.stats.$field.inc_by(u64::from(result.stats().$value)); - }; - } - - if result.stats().matched_logs_count > 0 { - inc_by!(matched_logs_total, matched_logs_count); - } + let has_debug_workflows = self + .configs + .iter() + .any(|config| config.mode() != WorkflowDebugMode::None); - self.state.active_run_tracing_count = self - .state - .active_run_tracing_count - .saturating_add(result.stats().tracing_starts); - self.state.active_run_tracing_count = self - .state - .active_run_tracing_count - .saturating_sub(result.stats().tracing_ends); - - // Not every case of a workflow making a progress needs a state persistence. - // If the workflow was in an initial state prior to processing a log and is in - // an initial state after processing the log then the state of workflow did not change - // as the result of processing a log and does not have to be persisted. An example for when - // a workflow makes progress but does not needs persistence is the following workflow - // with 2 nodes/states - a start log matching node and a final emit metric node, - // such workflow may: - // * be in an initial state - // * match a single log - // * execute emit metric action - // * be an initial state - if result.stats().did_make_progress() - && !(was_in_initial_state && workflow.is_in_initial_state()) - { - self.needs_state_persistence = true; - } + let Self { + state, + configs, + stats, + needs_state_persistence, + log_router, + .. + } = self; + let workflows = &mut state.workflows; + let workflow_count = workflows.len(); + let active_run_tracing_count = &mut state.active_run_tracing_count; + let routed_log = matches!(event, WorkflowEvent::Log(_)); + + // A routed log only evaluates the indices selected by the router. Other event kinds are + // broadcast because they can advance any workflow or change every route. + match event { + WorkflowEvent::Log(log) => { + for &index in log_router.select_candidates_for_log_type(log.log_type) { + Self::process_workflow_event( + configs, + workflows, + stats, + needs_state_persistence, + active_run_tracing_count, + index, + event, + state_reader, + now, + sampled_roll, + &mut workflow_event_output, + ); + } + }, + WorkflowEvent::SessionStart(_) | WorkflowEvent::StateChange(..) => { + for index in 0 .. workflow_count { + Self::process_workflow_event( + configs, + workflows, + stats, + needs_state_persistence, + active_run_tracing_count, + index, + event, + state_reader, + now, + sampled_roll, + &mut workflow_event_output, + ); + } + }, + } - // In debug only mode we do not trigger any actions, but we still inject logs so that - // workflows continue to advance if they depend on the injected logs. - has_debug_workflows |= config.mode() != WorkflowDebugMode::None; - let ( - triggered_actions, - workflow_logs_to_inject, - cumulative_workflow_debug_state, - incremental_workflow_debug_state, - workflow_tracing_carryover_flush_action_ids, - ) = result.into_parts(); - if !matches!(config.mode(), WorkflowDebugMode::DebugOnly) { - prepared_actions.incorporate_workflow_actions(index, triggered_actions); - } - logs_to_inject.extend(workflow_logs_to_inject); - if let Some(cumulative_workflow_debug_state) = cumulative_workflow_debug_state { - all_cumulative_workflow_debug_state - .push((workflow.id().to_string(), cumulative_workflow_debug_state)); - } - all_incremental_workflow_debug_state.extend( - incremental_workflow_debug_state - .into_iter() - .map(|state_key| WorkflowDebugKey { - workflow_id: workflow.id().to_string(), - state_key, - }), - ); - tracing_carryover_flush_action_ids.extend(workflow_tracing_carryover_flush_action_ids); + if routed_log { + log_router.refresh_selected_routes(|index| { + workflows + .get(index) + .zip(configs.get(index)) + .map(|(workflow, config)| workflow.log_route(config)) + .unwrap_or_default() + }); + } else { + Self::rebuild_log_routes(log_router, workflows, configs); } + let WorkflowEventOutput { + prepared_actions, + logs_to_inject, + all_cumulative_workflow_debug_state, + all_incremental_workflow_debug_state, + tracing_carryover_flush_action_ids, + } = workflow_event_output; let PreparedActions { mut flush_buffers_actions, emit_metric_action_counts, @@ -1054,6 +1159,15 @@ impl Drop for WorkflowsEngine { } } +#[derive(Default)] +struct WorkflowEventOutput<'a> { + prepared_actions: PreparedActions<'a>, + logs_to_inject: TinyMap<&'a str, Log>, + all_cumulative_workflow_debug_state: AllWorkflowsDebugState, + all_incremental_workflow_debug_state: Vec, + tracing_carryover_flush_action_ids: TinySet, +} + #[derive(Default)] struct PreparedActions<'a> { flush_buffers_actions: BTreeSet>, diff --git a/bd-workflows/src/engine_test.rs b/bd-workflows/src/engine_test.rs index 9fa890af3..d0cf482c5 100644 --- a/bd-workflows/src/engine_test.rs +++ b/bd-workflows/src/engine_test.rs @@ -24,7 +24,7 @@ use bd_client_stats::FlushTrigger; use bd_client_stats_store::Collector; use bd_client_stats_store::test::StatsHelper; use bd_error_reporter::reporter::{Reporter, UnexpectedErrorHandler}; -use bd_log_matcher::builder::{field_equals, message_equals, or}; +use bd_log_matcher::builder::{field_equals, log_type_equals, message_equals, or}; use bd_log_primitives::tiny_set::{TinyMap, TinySet}; use bd_log_primitives::{Log, LogFields, LogMessage, log_level}; use bd_proto::protos::client::api::sankey_path_upload_request::Node; @@ -3487,3 +3487,93 @@ async fn start_tracing_carries_into_streaming_until_streaming_ends() { assert!(!result.is_tracing_active); assert!(!engine.engine.is_tracing_active()); } + +#[tokio::test] +async fn log_type_router_updates_after_workflow_advances() { + let c = state("C"); + let b = state("B").declare_transition(&c, rule!(log_type_equals(LogType::RESOURCE))); + let a = state("A").declare_transition(&b, rule!(log_type_equals(LogType::LIFECYCLE))); + let workflow = WorkflowBuilder::new("workflow", &[&a, &b, &c]).make_config(); + + let setup = Setup::new(); + let mut engine = setup + .make_workflows_engine(WorkflowsEngineConfig::new_with_workflow_configurations( + vec![workflow], + )) + .await; + + engine.process_log(TestLog::new("lifecycle").with_log_type(LogType::LIFECYCLE)); + engine_assert_active_runs!(engine; 0; "B"); + assert_eq!( + &[0], + engine + .engine + .log_router + .select_candidates_for_log_type(LogType::NORMAL) + ); + + engine.process_log(TestLog::new("resource").with_log_type(LogType::RESOURCE)); + engine_assert_active_runs!(engine; 0; "A"); + assert_eq!( + &[] as &[usize], + engine + .engine + .log_router + .select_candidates_for_log_type(LogType::NORMAL) + ); + setup + .collector + .assert_counter_eq(2, "workflows:matched_logs_total", labels! {}); +} + +#[tokio::test] +async fn log_type_router_keeps_debug_workflows_on_the_fallback_route() { + let b = state("B"); + let a = state("A").declare_transition(&b, rule!(log_type_equals(LogType::LIFECYCLE))); + let workflow = WorkflowBuilder::new("workflow", &[&a, &b]) + .make_config_with_debug_mode(WorkflowDebugMode::DebugOnly); + + let setup = Setup::new(); + let mut engine = setup + .make_workflows_engine(WorkflowsEngineConfig::new_with_workflow_configurations( + vec![workflow], + )) + .await; + + // The initial delivery fallback is consumed here. The subsequent fallback is specifically for + // the active debug workflow. + engine.process_log(TestLog::new("normal").with_log_type(LogType::NORMAL)); + + assert_eq!( + &[0], + engine + .engine + .log_router + .select_candidates_for_log_type(LogType::NORMAL) + ); +} + +#[tokio::test] +async fn log_type_router_keeps_active_timeouts_on_the_fallback_route() { + let c = state("C"); + let b = state("B").with_timeout(&c, 1.minutes(), &[]); + let a = state("A").declare_transition(&b, rule!(log_type_equals(LogType::LIFECYCLE))); + let workflow = WorkflowBuilder::new("workflow", &[&a, &b, &c]).make_config(); + + let setup = Setup::new(); + let mut engine = setup + .make_workflows_engine(WorkflowsEngineConfig::new_with_workflow_configurations( + vec![workflow], + )) + .await; + + engine.process_log(TestLog::new("lifecycle").with_log_type(LogType::LIFECYCLE)); + + assert_eq!( + &[0], + engine + .engine + .log_router + .select_candidates_for_log_type(LogType::NORMAL) + ); +} diff --git a/bd-workflows/src/engine_test_helpers.rs b/bd-workflows/src/engine_test_helpers.rs index 85a543b88..045848de0 100644 --- a/bd-workflows/src/engine_test_helpers.rs +++ b/bd-workflows/src/engine_test_helpers.rs @@ -29,7 +29,6 @@ use bd_proto::protos::client::api::{ SankeyPathUploadRequest, log_upload_intent_request, }; -use bd_proto::protos::logging::payload::LogType; use bd_runtime::runtime::ConfigLoader; use bd_stats_common::{Counter, Histogram}; use bd_time::TimeDurationExt; @@ -149,7 +148,7 @@ impl AnnotatedWorkflowsEngine { pub fn process_log(&mut self, log: TestLog) -> WorkflowsEngineResult<'_> { self.engine.process_event( WorkflowEvent::Log(&bd_log_primitives::Log { - log_type: LogType::NORMAL, + log_type: log.log_type, log_level: log_level::DEBUG, message: LogMessage::String(log.message), session_id: log.session.unwrap_or_else(|| self.session_id.clone()), diff --git a/bd-workflows/src/lib.rs b/bd-workflows/src/lib.rs index 4907e8409..e1909c27d 100644 --- a/bd-workflows/src/lib.rs +++ b/bd-workflows/src/lib.rs @@ -21,6 +21,7 @@ pub mod config; pub mod engine; mod generate_log; pub mod metrics; +mod routing; mod sankey_diagram; // External users can opt into the test fixtures without making test helpers part of production // dependency closures. Unit tests receive the module through Cargo's `cfg(test)` automatically. diff --git a/bd-workflows/src/routing.rs b/bd-workflows/src/routing.rs new file mode 100644 index 000000000..89f1c5e43 --- /dev/null +++ b/bd-workflows/src/routing.rs @@ -0,0 +1,216 @@ +// shared-core - bitdrift's common client/server libraries +// Copyright Bitdrift, Inc. All rights reserved. +// +// Use of this source code is governed by a source available license that can be found in the +// LICENSE file or at: +// https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt + +#[cfg(test)] +#[path = "./routing_test.rs"] +mod routing_test; + +// +// WorkflowLogRoute +// + +use bd_log_matcher::matcher::LogTypeSet; +use bd_proto::protos::logging::payload::LogType; + +/// Describes which log types can require a workflow to process a log. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub enum WorkflowLogRoute { + /// No active traversal can react to a log. + #[default] + None, + /// The workflow must inspect every log to preserve its existing behavior. + Fallback, + /// The workflow only needs logs whose type is included in the set. + Types(LogTypeSet), +} + +// +// LogTypeBucket +// + +#[derive(Debug)] +struct LogTypeBucket { + log_type: LogType, + workflow_indices: Vec, +} + +// +// WorkflowLogRouter +// + +/// Maintains sorted workflow indices for efficient, allocation-free log routing. +/// +/// Workflow indices remain stable between configuration updates. The router is rebuilt at those +/// boundaries and otherwise only updates the workflows which processed the current log. +#[derive(Debug, Default)] +pub struct WorkflowLogRouter { + routes: Vec, + fallback_workflow_indices: Vec, + type_buckets: Vec, + candidate_indices: Vec, +} + +impl WorkflowLogRouter { + pub(crate) fn prepare(&mut self, workflow_count: usize, known_log_types: LogTypeSet) { + self.routes.clear(); + self.routes.reserve(workflow_count); + self.fallback_workflow_indices.clear(); + reserve_to(&mut self.fallback_workflow_indices, workflow_count); + self.candidate_indices.clear(); + reserve_to(&mut self.candidate_indices, workflow_count); + + for log_type in known_log_types.iter() { + if self + .type_buckets + .iter() + .all(|bucket| bucket.log_type != log_type) + { + self.type_buckets.push(LogTypeBucket { + log_type, + workflow_indices: Vec::with_capacity(workflow_count), + }); + } + } + + for bucket in &mut self.type_buckets { + bucket.workflow_indices.clear(); + reserve_to(&mut bucket.workflow_indices, workflow_count); + } + } + + pub(crate) fn insert(&mut self, workflow_index: usize, route: WorkflowLogRoute) { + debug_assert_eq!(workflow_index, self.routes.len()); + self.routes.push(route); + self.insert_route(workflow_index, route); + } + + /// Selects the workflows which need to evaluate a log with this type. + /// + /// The returned indices remain valid until the next selection or route refresh. + pub(crate) fn select_candidates_for_log_type(&mut self, log_type: LogType) -> &[usize] { + self.candidate_indices.clear(); + let type_indices = self + .type_buckets + .iter() + .find(|bucket| bucket.log_type == log_type) + .map_or(&[][..], |bucket| bucket.workflow_indices.as_slice()); + + merge_sorted_indices( + &self.fallback_workflow_indices, + type_indices, + &mut self.candidate_indices, + ); + &self.candidate_indices + } + + /// Updates routes for the workflows selected by the preceding log selection. + pub(crate) fn refresh_selected_routes(&mut self, mut route_for_index: F) + where + F: FnMut(usize) -> WorkflowLogRoute, + { + for candidate_position in 0 .. self.candidate_indices.len() { + let Some(workflow_index) = self.candidate_indices.get(candidate_position).copied() else { + continue; + }; + let route = route_for_index(workflow_index); + self.replace(workflow_index, route); + } + } + + fn replace(&mut self, workflow_index: usize, route: WorkflowLogRoute) { + let Some(previous_route) = self.routes.get(workflow_index).copied() else { + return; + }; + + if previous_route == route { + return; + } + + self.remove_route(workflow_index, previous_route); + if let Some(previous_route) = self.routes.get_mut(workflow_index) { + *previous_route = route; + } + self.insert_route(workflow_index, route); + } + + fn insert_route(&mut self, workflow_index: usize, route: WorkflowLogRoute) { + match route { + WorkflowLogRoute::None => {}, + WorkflowLogRoute::Fallback => { + insert_sorted(&mut self.fallback_workflow_indices, workflow_index); + }, + WorkflowLogRoute::Types(log_types) => { + for log_type in log_types.iter() { + if let Some(bucket) = self + .type_buckets + .iter_mut() + .find(|bucket| bucket.log_type == log_type) + { + insert_sorted(&mut bucket.workflow_indices, workflow_index); + } + } + }, + } + } + + fn remove_route(&mut self, workflow_index: usize, route: WorkflowLogRoute) { + match route { + WorkflowLogRoute::None => {}, + WorkflowLogRoute::Fallback => { + remove_sorted(&mut self.fallback_workflow_indices, workflow_index); + }, + WorkflowLogRoute::Types(log_types) => { + for log_type in log_types.iter() { + if let Some(bucket) = self + .type_buckets + .iter_mut() + .find(|bucket| bucket.log_type == log_type) + { + remove_sorted(&mut bucket.workflow_indices, workflow_index); + } + } + }, + } + } +} + +fn reserve_to(values: &mut Vec, capacity: usize) { + if values.capacity() < capacity { + values.reserve(capacity - values.capacity()); + } +} + +fn insert_sorted(values: &mut Vec, value: usize) { + match values.binary_search(&value) { + Ok(_) => {}, + Err(position) => values.insert(position, value), + } +} + +fn remove_sorted(values: &mut Vec, value: usize) { + if let Ok(position) = values.binary_search(&value) { + values.remove(position); + } +} + +fn merge_sorted_indices(first: &[usize], second: &[usize], output: &mut Vec) { + let mut first = first.iter().copied().peekable(); + let mut second = second.iter().copied().peekable(); + + while let (Some(first_index), Some(second_index)) = (first.peek(), second.peek()) { + if first_index < second_index { + if let Some(index) = first.next() { + output.push(index); + } + } else if let Some(index) = second.next() { + output.push(index); + } + } + + output.extend(first); + output.extend(second); +} diff --git a/bd-workflows/src/routing_test.rs b/bd-workflows/src/routing_test.rs new file mode 100644 index 000000000..ce05a1b1c --- /dev/null +++ b/bd-workflows/src/routing_test.rs @@ -0,0 +1,72 @@ +// shared-core - bitdrift's common client/server libraries +// Copyright Bitdrift, Inc. All rights reserved. +// +// Use of this source code is governed by a source available license that can be found in the +// LICENSE file or at: +// https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt + +use super::{WorkflowLogRoute, WorkflowLogRouter}; +use bd_log_matcher::matcher::LogTypeSet; +use bd_proto::protos::logging::payload::LogType; +use pretty_assertions::assert_eq; + +fn log_types(types: &[LogType]) -> LogTypeSet { + let mut log_types = LogTypeSet::default(); + for log_type in types { + log_types.union(LogTypeSet::from_log_type(*log_type)); + } + log_types +} + +#[test] +fn selects_only_matching_and_fallback_workflows_in_index_order() { + let lifecycle = log_types(&[LogType::LIFECYCLE]); + let lifecycle_or_resource = log_types(&[LogType::LIFECYCLE, LogType::RESOURCE]); + let mut router = WorkflowLogRouter::default(); + router.prepare(4, lifecycle_or_resource); + router.insert(0, WorkflowLogRoute::Types(lifecycle)); + router.insert(1, WorkflowLogRoute::Fallback); + router.insert(2, WorkflowLogRoute::Types(lifecycle_or_resource)); + router.insert(3, WorkflowLogRoute::None); + + assert_eq!( + &[0, 1, 2], + router.select_candidates_for_log_type(LogType::LIFECYCLE) + ); + assert_eq!( + &[1, 2], + router.select_candidates_for_log_type(LogType::RESOURCE) + ); + assert_eq!(&[1], router.select_candidates_for_log_type(LogType::NORMAL)); +} + +#[test] +fn refreshes_only_the_workflows_selected_for_a_log() { + let lifecycle = log_types(&[LogType::LIFECYCLE]); + let resource = log_types(&[LogType::RESOURCE]); + let lifecycle_or_resource = log_types(&[LogType::LIFECYCLE, LogType::RESOURCE]); + let mut router = WorkflowLogRouter::default(); + router.prepare(3, lifecycle_or_resource); + router.insert(0, WorkflowLogRoute::Types(lifecycle)); + router.insert(1, WorkflowLogRoute::Fallback); + router.insert(2, WorkflowLogRoute::Types(resource)); + + assert_eq!( + &[0, 1], + router.select_candidates_for_log_type(LogType::LIFECYCLE) + ); + router.refresh_selected_routes(|index| match index { + 0 => WorkflowLogRoute::Types(resource), + 1 => WorkflowLogRoute::None, + _ => unreachable!(), + }); + + assert_eq!( + &[] as &[usize], + router.select_candidates_for_log_type(LogType::LIFECYCLE) + ); + assert_eq!( + &[0, 2], + router.select_candidates_for_log_type(LogType::RESOURCE) + ); +} diff --git a/bd-workflows/src/test.rs b/bd-workflows/src/test.rs index 899dad37d..fb551af7a 100644 --- a/bd-workflows/src/test.rs +++ b/bd-workflows/src/test.rs @@ -9,6 +9,7 @@ #![allow(clippy::unwrap_used)] use crate::config::{Config, WorkflowDebugMode}; +use bd_proto::protos::logging::payload::LogType; use bd_test_helpers::workflow::WorkflowBuilder; use std::collections::BTreeMap; use time::OffsetDateTime; @@ -30,6 +31,7 @@ impl MakeConfig for WorkflowBuilder { pub struct TestLog { pub message: String, + pub log_type: LogType, pub occurred_at: OffsetDateTime, pub now: OffsetDateTime, pub tags: BTreeMap, @@ -42,6 +44,7 @@ impl TestLog { let now = OffsetDateTime::now_utc(); Self { message: message.to_string(), + log_type: LogType::NORMAL, occurred_at: now, now, tags: BTreeMap::new(), @@ -55,6 +58,12 @@ impl TestLog { self } + #[must_use] + pub const fn with_log_type(mut self, log_type: LogType) -> Self { + self.log_type = log_type; + self + } + #[must_use] pub fn with_tags(mut self, tags: BTreeMap) -> Self { self.tags = tags; diff --git a/bd-workflows/src/workflow.rs b/bd-workflows/src/workflow.rs index 8ab015fd7..d9aec10ca 100644 --- a/bd-workflows/src/workflow.rs +++ b/bd-workflows/src/workflow.rs @@ -22,6 +22,7 @@ use crate::config::{ WorkflowDebugMode, }; use crate::generate_log::generate_log_action; +use crate::routing::WorkflowLogRoute; use bd_log_primitives::tiny_set::{TinyMap, TinySet}; use bd_log_primitives::{FieldsRef, Log, log_level}; use bd_proto::protos::logging::payload::LogType; @@ -245,6 +246,52 @@ impl WorkflowEvent<'_> { } } +// +// WorkflowLogRouteBuilder +// + +#[derive(Default)] +struct WorkflowLogRouteBuilder { + log_types: bd_log_matcher::matcher::LogTypeSet, + needs_fallback: bool, +} + +impl WorkflowLogRouteBuilder { + fn add_state(&mut self, config: &Config, state_index: usize) { + let Some(state) = config.inner().states().get(state_index) else { + return; + }; + + // A state timeout is checked when any event arrives, not only when a transition's matcher + // accepts that event. Keep such states on the fallback path. + if state.timeout().is_some() { + self.needs_fallback = true; + } + + for transition in state.transitions() { + let Predicate::LogMatch { matcher, .. } = transition.rule() else { + continue; + }; + + if let Some(log_types) = matcher.possible_log_types() { + self.log_types.union(log_types); + } else { + self.needs_fallback = true; + } + } + } + + fn finish(self) -> WorkflowLogRoute { + if self.needs_fallback { + WorkflowLogRoute::Fallback + } else if self.log_types.is_empty() { + WorkflowLogRoute::None + } else { + WorkflowLogRoute::Types(self.log_types) + } + } +} + // // Workflow // @@ -329,6 +376,39 @@ impl Workflow { self.workflow_debug_state = None; } + /// Returns the log types that can advance the workflow's current state. + /// + /// This deliberately keeps debug workflows, active timeouts, and initial delivery state on the + /// fallback route so their behavior remains unchanged while a debug session is active. + pub(crate) fn log_route(&self, config: &Config) -> WorkflowLogRoute { + if self.needs_start_metric || config.mode() != WorkflowDebugMode::None { + return WorkflowLogRoute::Fallback; + } + + let mut route = WorkflowLogRouteBuilder::default(); + + // Workflow execution lazily creates an initial run while processing every log. It must remain + // on the fallback path until that run exists, even when the initial state has no log matcher. + if self.needs_new_run() { + return WorkflowLogRoute::Fallback; + } + + for run in &self.runs { + if run.first_progress_occurred_at.is_some() && config.inner().duration_limit().is_some() { + return WorkflowLogRoute::Fallback; + } + + for traversal in &run.traversals { + if traversal.timeout_unix_ms.is_some() { + return WorkflowLogRoute::Fallback; + } + route.add_state(config, traversal.state_index); + } + } + + route.finish() + } + pub(crate) fn process_event<'a>( &mut self, config: &'a Config, From 02b42ff961327af947208f8f03808ed778524951 Mon Sep 17 00:00:00 2001 From: Snow Pettersen Date: Fri, 7 Aug 2026 09:30:54 -0400 Subject: [PATCH 02/23] Fix workflow routing test lint Co-Authored-By: GPT-5 --- bd-workflows/src/routing_test.rs | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/bd-workflows/src/routing_test.rs b/bd-workflows/src/routing_test.rs index ce05a1b1c..cf991d158 100644 --- a/bd-workflows/src/routing_test.rs +++ b/bd-workflows/src/routing_test.rs @@ -55,11 +55,8 @@ fn refreshes_only_the_workflows_selected_for_a_log() { &[0, 1], router.select_candidates_for_log_type(LogType::LIFECYCLE) ); - router.refresh_selected_routes(|index| match index { - 0 => WorkflowLogRoute::Types(resource), - 1 => WorkflowLogRoute::None, - _ => unreachable!(), - }); + let refreshed_routes = [WorkflowLogRoute::Types(resource), WorkflowLogRoute::None]; + router.refresh_selected_routes(|index| refreshed_routes[index]); assert_eq!( &[] as &[usize], From dc80326f406ea3581f1f3460a009afe20452cd0a Mon Sep 17 00:00:00 2001 From: Snow Pettersen Date: Fri, 7 Aug 2026 09:41:19 -0400 Subject: [PATCH 03/23] Track processed debug workflows Co-Authored-By: GPT-5 --- bd-workflows/src/engine.rs | 7 +++---- bd-workflows/src/engine_test.rs | 6 +++++- 2 files changed, 8 insertions(+), 5 deletions(-) diff --git a/bd-workflows/src/engine.rs b/bd-workflows/src/engine.rs index ccde9750c..7c3778d32 100644 --- a/bd-workflows/src/engine.rs +++ b/bd-workflows/src/engine.rs @@ -577,6 +577,7 @@ impl WorkflowsEngine { let was_in_initial_state = workflow.is_in_initial_state(); let result = workflow.process_event(config, event, state_reader, now, sampled_roll); + output.has_debug_workflows |= config.mode() != WorkflowDebugMode::None; if result.stats().matched_logs_count > 0 { stats @@ -907,10 +908,6 @@ impl WorkflowsEngine { // Sampling decisions should be stable for a single event across every workflow/run that // evaluates it. Roll once here and thread the same value through matcher evaluation. let sampled_roll = bd_log_matcher::matcher::random_sample_roll(); - let has_debug_workflows = self - .configs - .iter() - .any(|config| config.mode() != WorkflowDebugMode::None); let Self { state, @@ -982,6 +979,7 @@ impl WorkflowsEngine { all_cumulative_workflow_debug_state, all_incremental_workflow_debug_state, tracing_carryover_flush_action_ids, + has_debug_workflows, } = workflow_event_output; let PreparedActions { mut flush_buffers_actions, @@ -1166,6 +1164,7 @@ struct WorkflowEventOutput<'a> { all_cumulative_workflow_debug_state: AllWorkflowsDebugState, all_incremental_workflow_debug_state: Vec, tracing_carryover_flush_action_ids: TinySet, + has_debug_workflows: bool, } #[derive(Default)] diff --git a/bd-workflows/src/engine_test.rs b/bd-workflows/src/engine_test.rs index d0cf482c5..234842894 100644 --- a/bd-workflows/src/engine_test.rs +++ b/bd-workflows/src/engine_test.rs @@ -3542,7 +3542,11 @@ async fn log_type_router_keeps_debug_workflows_on_the_fallback_route() { // The initial delivery fallback is consumed here. The subsequent fallback is specifically for // the active debug workflow. - engine.process_log(TestLog::new("normal").with_log_type(LogType::NORMAL)); + assert!( + engine + .process_log(TestLog::new("normal").with_log_type(LogType::NORMAL)) + .has_debug_workflows + ); assert_eq!( &[0], From c7b1aaedd2a234cff24b01603217c8b5c9186374 Mon Sep 17 00:00:00 2001 From: Snow Pettersen Date: Fri, 7 Aug 2026 09:45:38 -0400 Subject: [PATCH 04/23] Document workflow duration routing limitation Co-Authored-By: GPT-5 --- bd-workflows/src/workflow.rs | 3 +++ 1 file changed, 3 insertions(+) diff --git a/bd-workflows/src/workflow.rs b/bd-workflows/src/workflow.rs index d9aec10ca..173e4913d 100644 --- a/bd-workflows/src/workflow.rs +++ b/bd-workflows/src/workflow.rs @@ -394,6 +394,9 @@ impl Workflow { } for run in &self.runs { + // TODO: Decouple duration expiry from inbound log processing so a progressed workflow can + // remain on its type-specific route. This needs an expiry mechanism that preserves run + // termination and tracing updates without evaluating every log. if run.first_progress_occurred_at.is_some() && config.inner().duration_limit().is_some() { return WorkflowLogRoute::Fallback; } From f6b354b70398d943c3a08ec81a1ae444d790793c Mon Sep 17 00:00:00 2001 From: Snow Pettersen Date: Tue, 11 Aug 2026 11:06:47 -0400 Subject: [PATCH 05/23] comment changes --- bd-log-matcher/src/matcher.rs | 7 ++----- bd-workflows/src/routing.rs | 13 +++++++++---- 2 files changed, 11 insertions(+), 9 deletions(-) diff --git a/bd-log-matcher/src/matcher.rs b/bd-log-matcher/src/matcher.rs index a3cf906b5..67b723062 100644 --- a/bd-log-matcher/src/matcher.rs +++ b/bd-log-matcher/src/matcher.rs @@ -59,11 +59,8 @@ pub const SAMPLE_RATE_DENOMINATOR: u32 = 1_000_000; // LogTypeSet // -/// An allocation-free bitset of SDK log types. -/// -/// Values outside the current `LogType` enum are represented by an empty set because the runtime -/// cannot receive them. Keeping the bitmask behind this type makes the enum-to-bit conversion -/// explicit without exposing numeric masks to routing code. +/// An allocation-free bitset of SDK log types. This is used to track the set of log types that a +/// workflow can match against. #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] pub struct LogTypeSet(u16); diff --git a/bd-workflows/src/routing.rs b/bd-workflows/src/routing.rs index 89f1c5e43..1357c054c 100644 --- a/bd-workflows/src/routing.rs +++ b/bd-workflows/src/routing.rs @@ -9,20 +9,25 @@ #[path = "./routing_test.rs"] mod routing_test; +use bd_log_matcher::matcher::LogTypeSet; +use bd_proto::protos::logging::payload::LogType; + // // WorkflowLogRoute // -use bd_log_matcher::matcher::LogTypeSet; -use bd_proto::protos::logging::payload::LogType; +// TODO(snowp): We should be able to apply this routing logic to other event types to dramatically +// reduce the number of workflows attempted for state events. For now state events are rare so we +// limit this to logs. /// Describes which log types can require a workflow to process a log. #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] pub enum WorkflowLogRoute { - /// No active traversal can react to a log. + /// No active traversal can react to a log. This can happen if the workflow is matching on a state + /// change. #[default] None, - /// The workflow must inspect every log to preserve its existing behavior. + /// The workflow must inspect every log. Fallback, /// The workflow only needs logs whose type is included in the set. Types(LogTypeSet), From 61ecf8f20190f25216bc1504241cdb8f756ca2ff Mon Sep 17 00:00:00 2001 From: Snow Pettersen Date: Tue, 11 Aug 2026 11:13:18 -0400 Subject: [PATCH 06/23] license headers --- bd-workflows/src/routing.rs | 2 +- bd-workflows/src/routing_test.rs | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/bd-workflows/src/routing.rs b/bd-workflows/src/routing.rs index 1357c054c..42d0e17c0 100644 --- a/bd-workflows/src/routing.rs +++ b/bd-workflows/src/routing.rs @@ -2,7 +2,7 @@ // Copyright Bitdrift, Inc. All rights reserved. // // Use of this source code is governed by a source available license that can be found in the -// LICENSE file or at: +// LICENSE.polyform file or at: // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt #[cfg(test)] diff --git a/bd-workflows/src/routing_test.rs b/bd-workflows/src/routing_test.rs index cf991d158..e64c1438f 100644 --- a/bd-workflows/src/routing_test.rs +++ b/bd-workflows/src/routing_test.rs @@ -2,7 +2,7 @@ // Copyright Bitdrift, Inc. All rights reserved. // // Use of this source code is governed by a source available license that can be found in the -// LICENSE file or at: +// LICENSE.polyform file or at: // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt use super::{WorkflowLogRoute, WorkflowLogRouter}; From 3b3834bd63381e42ca31d80c013b583a7c3434ae Mon Sep 17 00:00:00 2001 From: Snow Pettersen Date: Tue, 11 Aug 2026 11:22:32 -0400 Subject: [PATCH 07/23] comments --- bd-workflows/src/workflow.rs | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/bd-workflows/src/workflow.rs b/bd-workflows/src/workflow.rs index 173e4913d..ec202bfc7 100644 --- a/bd-workflows/src/workflow.rs +++ b/bd-workflows/src/workflow.rs @@ -379,7 +379,9 @@ impl Workflow { /// Returns the log types that can advance the workflow's current state. /// /// This deliberately keeps debug workflows, active timeouts, and initial delivery state on the - /// fallback route so their behavior remains unchanged while a debug session is active. + /// fallback route so their behavior remains unchanged while a debug session is active. A + /// consequence of this is that we effectively only apply this to workflows in the initial + /// condition, as in general all workflows have a total timeout. pub(crate) fn log_route(&self, config: &Config) -> WorkflowLogRoute { if self.needs_start_metric || config.mode() != WorkflowDebugMode::None { return WorkflowLogRoute::Fallback; @@ -394,9 +396,9 @@ impl Workflow { } for run in &self.runs { - // TODO: Decouple duration expiry from inbound log processing so a progressed workflow can - // remain on its type-specific route. This needs an expiry mechanism that preserves run - // termination and tracing updates without evaluating every log. + // TODO(snowp): Decouple duration expiry from inbound log processing so a progressed workflow + // can remain on its type-specific route. This needs an expiry mechanism that preserves + // run termination and tracing updates without evaluating every log. if run.first_progress_occurred_at.is_some() && config.inner().duration_limit().is_some() { return WorkflowLogRoute::Fallback; } From cd1acbbd3aefaaa7bdc4718ca3061e6ca3cf92e7 Mon Sep 17 00:00:00 2001 From: Snow Pettersen Date: Tue, 11 Aug 2026 11:29:18 -0400 Subject: [PATCH 08/23] Use fixed log type routing buckets Co-Authored-By: GPT-5 --- bd-workflows/src/config.rs | 16 --------- bd-workflows/src/engine.rs | 6 +--- bd-workflows/src/engine_test.rs | 41 ++++++++++++++++++++++ bd-workflows/src/routing.rs | 59 +++++++------------------------- bd-workflows/src/routing_test.rs | 5 ++- 5 files changed, 57 insertions(+), 70 deletions(-) diff --git a/bd-workflows/src/config.rs b/bd-workflows/src/config.rs index cc2622bcb..75cf39370 100644 --- a/bd-workflows/src/config.rs +++ b/bd-workflows/src/config.rs @@ -217,10 +217,6 @@ impl Config { &self.inner } - pub(crate) fn possible_log_types(&self) -> bd_log_matcher::matcher::LogTypeSet { - self.inner.possible_log_types() - } - pub(crate) const fn mode(&self) -> WorkflowDebugMode { self.mode } @@ -331,18 +327,6 @@ impl InnerConfig { .map(|state| state.transitions.as_slice()) } - pub(crate) fn possible_log_types(&self) -> bd_log_matcher::matcher::LogTypeSet { - let mut log_types = bd_log_matcher::matcher::LogTypeSet::default(); - for transition in self.states.iter().flat_map(State::transitions) { - if let Predicate::LogMatch { matcher, .. } = transition.rule() - && let Some(matcher_log_types) = matcher.possible_log_types() - { - log_types.union(matcher_log_types); - } - } - log_types - } - pub(crate) fn actions_for_traversal( &self, traversal: &Traversal, diff --git a/bd-workflows/src/engine.rs b/bd-workflows/src/engine.rs index 7c3778d32..6dac55838 100644 --- a/bd-workflows/src/engine.rs +++ b/bd-workflows/src/engine.rs @@ -645,11 +645,7 @@ impl WorkflowsEngine { workflows: &[Workflow], configs: &[Config], ) { - let mut known_log_types = bd_log_matcher::matcher::LogTypeSet::default(); - for config in configs { - known_log_types.union(config.possible_log_types()); - } - log_router.prepare(workflows.len(), known_log_types); + log_router.prepare(workflows.len()); for (index, (workflow, config)) in workflows.iter().zip(configs).enumerate() { log_router.insert(index, workflow.log_route(config)); diff --git a/bd-workflows/src/engine_test.rs b/bd-workflows/src/engine_test.rs index 234842894..f56749883 100644 --- a/bd-workflows/src/engine_test.rs +++ b/bd-workflows/src/engine_test.rs @@ -3526,6 +3526,47 @@ async fn log_type_router_updates_after_workflow_advances() { .assert_counter_eq(2, "workflows:matched_logs_total", labels! {}); } +#[tokio::test] +async fn log_type_router_recreates_initial_run_after_completion() { + let b = state("B"); + let a = state("A").declare_transition(&b, rule!(log_type_equals(LogType::LIFECYCLE))); + let workflow = WorkflowBuilder::new("workflow", &[&a, &b]).make_config(); + + let setup = Setup::new(); + let mut engine = setup + .make_workflows_engine(WorkflowsEngineConfig::new_with_workflow_configurations( + vec![workflow], + )) + .await; + + // Consume the initial delivery fallback, leaving an initial run whose only matching type is + // LIFECYCLE. + engine.process_log(TestLog::new("normal").with_log_type(LogType::NORMAL)); + engine_assert_active_runs!(engine; 0; "A"); + assert_eq!( + &[] as &[usize], + engine + .engine + .log_router + .select_candidates_for_log_type(LogType::NORMAL) + ); + + // Completing the run leaves no initial run. The router must fall back so the next log can + // recreate one. + engine.process_log(TestLog::new("lifecycle").with_log_type(LogType::LIFECYCLE)); + assert!(engine.engine.state.workflows[0].runs().is_empty()); + assert_eq!( + &[0], + engine + .engine + .log_router + .select_candidates_for_log_type(LogType::NORMAL) + ); + + engine.process_log(TestLog::new("normal").with_log_type(LogType::NORMAL)); + engine_assert_active_runs!(engine; 0; "A"); +} + #[tokio::test] async fn log_type_router_keeps_debug_workflows_on_the_fallback_route() { let b = state("B"); diff --git a/bd-workflows/src/routing.rs b/bd-workflows/src/routing.rs index 42d0e17c0..c9ad70234 100644 --- a/bd-workflows/src/routing.rs +++ b/bd-workflows/src/routing.rs @@ -11,6 +11,12 @@ mod routing_test; use bd_log_matcher::matcher::LogTypeSet; use bd_proto::protos::logging::payload::LogType; +use protobuf::Enum; + +// The routing table indexes directly by the generated enum discriminant. Keep this in lockstep +// with the SDK enum so a sparse enum or a new value after SPAN cannot silently misroute logs. +const LOG_TYPE_BUCKET_COUNT: usize = LogType::VALUES.len(); +const _: () = assert!(LOG_TYPE_BUCKET_COUNT == LogType::SPAN as usize + 1); // // WorkflowLogRoute @@ -33,16 +39,6 @@ pub enum WorkflowLogRoute { Types(LogTypeSet), } -// -// LogTypeBucket -// - -#[derive(Debug)] -struct LogTypeBucket { - log_type: LogType, - workflow_indices: Vec, -} - // // WorkflowLogRouter // @@ -55,12 +51,12 @@ struct LogTypeBucket { pub struct WorkflowLogRouter { routes: Vec, fallback_workflow_indices: Vec, - type_buckets: Vec, + type_buckets: [Vec; LOG_TYPE_BUCKET_COUNT], candidate_indices: Vec, } impl WorkflowLogRouter { - pub(crate) fn prepare(&mut self, workflow_count: usize, known_log_types: LogTypeSet) { + pub(crate) fn prepare(&mut self, workflow_count: usize) { self.routes.clear(); self.routes.reserve(workflow_count); self.fallback_workflow_indices.clear(); @@ -68,22 +64,9 @@ impl WorkflowLogRouter { self.candidate_indices.clear(); reserve_to(&mut self.candidate_indices, workflow_count); - for log_type in known_log_types.iter() { - if self - .type_buckets - .iter() - .all(|bucket| bucket.log_type != log_type) - { - self.type_buckets.push(LogTypeBucket { - log_type, - workflow_indices: Vec::with_capacity(workflow_count), - }); - } - } - for bucket in &mut self.type_buckets { - bucket.workflow_indices.clear(); - reserve_to(&mut bucket.workflow_indices, workflow_count); + bucket.clear(); + reserve_to(bucket, workflow_count); } } @@ -98,11 +81,7 @@ impl WorkflowLogRouter { /// The returned indices remain valid until the next selection or route refresh. pub(crate) fn select_candidates_for_log_type(&mut self, log_type: LogType) -> &[usize] { self.candidate_indices.clear(); - let type_indices = self - .type_buckets - .iter() - .find(|bucket| bucket.log_type == log_type) - .map_or(&[][..], |bucket| bucket.workflow_indices.as_slice()); + let type_indices = &self.type_buckets[log_type as usize]; merge_sorted_indices( &self.fallback_workflow_indices, @@ -150,13 +129,7 @@ impl WorkflowLogRouter { }, WorkflowLogRoute::Types(log_types) => { for log_type in log_types.iter() { - if let Some(bucket) = self - .type_buckets - .iter_mut() - .find(|bucket| bucket.log_type == log_type) - { - insert_sorted(&mut bucket.workflow_indices, workflow_index); - } + insert_sorted(&mut self.type_buckets[log_type as usize], workflow_index); } }, } @@ -170,13 +143,7 @@ impl WorkflowLogRouter { }, WorkflowLogRoute::Types(log_types) => { for log_type in log_types.iter() { - if let Some(bucket) = self - .type_buckets - .iter_mut() - .find(|bucket| bucket.log_type == log_type) - { - remove_sorted(&mut bucket.workflow_indices, workflow_index); - } + remove_sorted(&mut self.type_buckets[log_type as usize], workflow_index); } }, } diff --git a/bd-workflows/src/routing_test.rs b/bd-workflows/src/routing_test.rs index e64c1438f..a7f37ae4b 100644 --- a/bd-workflows/src/routing_test.rs +++ b/bd-workflows/src/routing_test.rs @@ -23,7 +23,7 @@ fn selects_only_matching_and_fallback_workflows_in_index_order() { let lifecycle = log_types(&[LogType::LIFECYCLE]); let lifecycle_or_resource = log_types(&[LogType::LIFECYCLE, LogType::RESOURCE]); let mut router = WorkflowLogRouter::default(); - router.prepare(4, lifecycle_or_resource); + router.prepare(4); router.insert(0, WorkflowLogRoute::Types(lifecycle)); router.insert(1, WorkflowLogRoute::Fallback); router.insert(2, WorkflowLogRoute::Types(lifecycle_or_resource)); @@ -44,9 +44,8 @@ fn selects_only_matching_and_fallback_workflows_in_index_order() { fn refreshes_only_the_workflows_selected_for_a_log() { let lifecycle = log_types(&[LogType::LIFECYCLE]); let resource = log_types(&[LogType::RESOURCE]); - let lifecycle_or_resource = log_types(&[LogType::LIFECYCLE, LogType::RESOURCE]); let mut router = WorkflowLogRouter::default(); - router.prepare(3, lifecycle_or_resource); + router.prepare(3); router.insert(0, WorkflowLogRoute::Types(lifecycle)); router.insert(1, WorkflowLogRoute::Fallback); router.insert(2, WorkflowLogRoute::Types(resource)); From 55fdd209f699d765c5ae4ee1db20b3555c3f0f31 Mon Sep 17 00:00:00 2001 From: Snow Pettersen Date: Tue, 11 Aug 2026 11:31:25 -0400 Subject: [PATCH 09/23] Test log type bucket indices Co-Authored-By: GPT-5 --- bd-workflows/src/routing.rs | 3 --- bd-workflows/src/routing_test.rs | 8 ++++++++ 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/bd-workflows/src/routing.rs b/bd-workflows/src/routing.rs index c9ad70234..b2fa8e809 100644 --- a/bd-workflows/src/routing.rs +++ b/bd-workflows/src/routing.rs @@ -13,10 +13,7 @@ use bd_log_matcher::matcher::LogTypeSet; use bd_proto::protos::logging::payload::LogType; use protobuf::Enum; -// The routing table indexes directly by the generated enum discriminant. Keep this in lockstep -// with the SDK enum so a sparse enum or a new value after SPAN cannot silently misroute logs. const LOG_TYPE_BUCKET_COUNT: usize = LogType::VALUES.len(); -const _: () = assert!(LOG_TYPE_BUCKET_COUNT == LogType::SPAN as usize + 1); // // WorkflowLogRoute diff --git a/bd-workflows/src/routing_test.rs b/bd-workflows/src/routing_test.rs index a7f37ae4b..022082dfa 100644 --- a/bd-workflows/src/routing_test.rs +++ b/bd-workflows/src/routing_test.rs @@ -9,6 +9,7 @@ use super::{WorkflowLogRoute, WorkflowLogRouter}; use bd_log_matcher::matcher::LogTypeSet; use bd_proto::protos::logging::payload::LogType; use pretty_assertions::assert_eq; +use protobuf::Enum; fn log_types(types: &[LogType]) -> LogTypeSet { let mut log_types = LogTypeSet::default(); @@ -18,6 +19,13 @@ fn log_types(types: &[LogType]) -> LogTypeSet { log_types } +#[test] +fn log_type_values_are_dense_bucket_indices() { + for (index, log_type) in LogType::VALUES.iter().enumerate() { + assert_eq!(index, *log_type as usize); + } +} + #[test] fn selects_only_matching_and_fallback_workflows_in_index_order() { let lifecycle = log_types(&[LogType::LIFECYCLE]); From 61f70c3e58dc09cedd01324a49e80ccf7dccaec5 Mon Sep 17 00:00:00 2001 From: Snow Pettersen Date: Tue, 11 Aug 2026 11:34:08 -0400 Subject: [PATCH 10/23] Document workflow router index invariants Co-Authored-By: GPT-5 --- bd-workflows/src/engine.rs | 7 +++---- bd-workflows/src/routing.rs | 7 +++++-- 2 files changed, 8 insertions(+), 6 deletions(-) diff --git a/bd-workflows/src/engine.rs b/bd-workflows/src/engine.rs index 6dac55838..265039229 100644 --- a/bd-workflows/src/engine.rs +++ b/bd-workflows/src/engine.rs @@ -115,10 +115,9 @@ impl ProcessLocalPendingFlushState { /// Orchestrates the execution and management of workflows. It is also responsible for /// persisting and restoring its state in disk when any workflow has changed. pub struct WorkflowsEngine { - // Number of elements in configs and - // state.workflows should be always the same. - // A config at index `i` corresponds to a workflow (state.workflows list) - // at index `i`. + // `configs` and `state.workflows` must remain index-aligned. `log_router` layers an + // index-based routing view over `state.workflows`, so it must be rebuilt whenever either + // primary list is added to, removed from, or reordered. configs: Vec, state: WorkflowsState, log_router: WorkflowLogRouter, diff --git a/bd-workflows/src/routing.rs b/bd-workflows/src/routing.rs index b2fa8e809..dd068ad1b 100644 --- a/bd-workflows/src/routing.rs +++ b/bd-workflows/src/routing.rs @@ -42,8 +42,10 @@ pub enum WorkflowLogRoute { /// Maintains sorted workflow indices for efficient, allocation-free log routing. /// -/// Workflow indices remain stable between configuration updates. The router is rebuilt at those -/// boundaries and otherwise only updates the workflows which processed the current log. +/// This is a derived view over the engine's primary workflow vector: every stored index refers to +/// the workflow at that same position. The caller must rebuild it after adding, removing, or +/// reordering workflows; between those boundaries, the router updates only workflows which +/// processed the current log. #[derive(Debug, Default)] pub struct WorkflowLogRouter { routes: Vec, @@ -53,6 +55,7 @@ pub struct WorkflowLogRouter { } impl WorkflowLogRouter { + /// Clears the derived routes before repopulating them in primary workflow-vector order. pub(crate) fn prepare(&mut self, workflow_count: usize) { self.routes.clear(); self.routes.reserve(workflow_count); From a5b30c70560869adb372dcee06ee6cb738ddf81c Mon Sep 17 00:00:00 2001 From: Snow Pettersen Date: Tue, 11 Aug 2026 11:35:16 -0400 Subject: [PATCH 11/23] Document log routing allocation tradeoff Co-Authored-By: GPT-5 --- bd-workflows/src/routing.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/bd-workflows/src/routing.rs b/bd-workflows/src/routing.rs index dd068ad1b..8a6c04abd 100644 --- a/bd-workflows/src/routing.rs +++ b/bd-workflows/src/routing.rs @@ -46,6 +46,10 @@ pub enum WorkflowLogRoute { /// the workflow at that same position. The caller must rebuild it after adding, removing, or /// reordering workflows; between those boundaries, the router updates only workflows which /// processed the current log. +/// +/// Non-matching logs are expected to be common, so selection and evaluation do not allocate. +/// Rebuilds may reserve storage after configuration or workflow-state changes; that infrequent +/// cost keeps the log-processing hot path allocation-free. #[derive(Debug, Default)] pub struct WorkflowLogRouter { routes: Vec, From 67961ae55ced37e55dae6d1c340cbd9956450363 Mon Sep 17 00:00:00 2001 From: Snow Pettersen Date: Tue, 11 Aug 2026 11:39:22 -0400 Subject: [PATCH 12/23] Clarify workflow router route insertion Co-Authored-By: GPT-5 --- bd-workflows/src/engine.rs | 2 +- bd-workflows/src/routing.rs | 12 ++++++------ bd-workflows/src/routing_test.rs | 14 +++++++------- 3 files changed, 14 insertions(+), 14 deletions(-) diff --git a/bd-workflows/src/engine.rs b/bd-workflows/src/engine.rs index 265039229..f80373bad 100644 --- a/bd-workflows/src/engine.rs +++ b/bd-workflows/src/engine.rs @@ -647,7 +647,7 @@ impl WorkflowsEngine { log_router.prepare(workflows.len()); for (index, (workflow, config)) in workflows.iter().zip(configs).enumerate() { - log_router.insert(index, workflow.log_route(config)); + log_router.append_workflow_route(index, workflow.log_route(config)); } } diff --git a/bd-workflows/src/routing.rs b/bd-workflows/src/routing.rs index 8a6c04abd..88c8c6fca 100644 --- a/bd-workflows/src/routing.rs +++ b/bd-workflows/src/routing.rs @@ -74,10 +74,10 @@ impl WorkflowLogRouter { } } - pub(crate) fn insert(&mut self, workflow_index: usize, route: WorkflowLogRoute) { + pub(crate) fn append_workflow_route(&mut self, workflow_index: usize, route: WorkflowLogRoute) { debug_assert_eq!(workflow_index, self.routes.len()); self.routes.push(route); - self.insert_route(workflow_index, route); + self.add_route_to_buckets(workflow_index, route); } /// Selects the workflows which need to evaluate a log with this type. @@ -118,14 +118,14 @@ impl WorkflowLogRouter { return; } - self.remove_route(workflow_index, previous_route); + self.remove_route_from_buckets(workflow_index, previous_route); if let Some(previous_route) = self.routes.get_mut(workflow_index) { *previous_route = route; } - self.insert_route(workflow_index, route); + self.add_route_to_buckets(workflow_index, route); } - fn insert_route(&mut self, workflow_index: usize, route: WorkflowLogRoute) { + fn add_route_to_buckets(&mut self, workflow_index: usize, route: WorkflowLogRoute) { match route { WorkflowLogRoute::None => {}, WorkflowLogRoute::Fallback => { @@ -139,7 +139,7 @@ impl WorkflowLogRouter { } } - fn remove_route(&mut self, workflow_index: usize, route: WorkflowLogRoute) { + fn remove_route_from_buckets(&mut self, workflow_index: usize, route: WorkflowLogRoute) { match route { WorkflowLogRoute::None => {}, WorkflowLogRoute::Fallback => { diff --git a/bd-workflows/src/routing_test.rs b/bd-workflows/src/routing_test.rs index 022082dfa..f9420d27c 100644 --- a/bd-workflows/src/routing_test.rs +++ b/bd-workflows/src/routing_test.rs @@ -32,10 +32,10 @@ fn selects_only_matching_and_fallback_workflows_in_index_order() { let lifecycle_or_resource = log_types(&[LogType::LIFECYCLE, LogType::RESOURCE]); let mut router = WorkflowLogRouter::default(); router.prepare(4); - router.insert(0, WorkflowLogRoute::Types(lifecycle)); - router.insert(1, WorkflowLogRoute::Fallback); - router.insert(2, WorkflowLogRoute::Types(lifecycle_or_resource)); - router.insert(3, WorkflowLogRoute::None); + router.append_workflow_route(0, WorkflowLogRoute::Types(lifecycle)); + router.append_workflow_route(1, WorkflowLogRoute::Fallback); + router.append_workflow_route(2, WorkflowLogRoute::Types(lifecycle_or_resource)); + router.append_workflow_route(3, WorkflowLogRoute::None); assert_eq!( &[0, 1, 2], @@ -54,9 +54,9 @@ fn refreshes_only_the_workflows_selected_for_a_log() { let resource = log_types(&[LogType::RESOURCE]); let mut router = WorkflowLogRouter::default(); router.prepare(3); - router.insert(0, WorkflowLogRoute::Types(lifecycle)); - router.insert(1, WorkflowLogRoute::Fallback); - router.insert(2, WorkflowLogRoute::Types(resource)); + router.append_workflow_route(0, WorkflowLogRoute::Types(lifecycle)); + router.append_workflow_route(1, WorkflowLogRoute::Fallback); + router.append_workflow_route(2, WorkflowLogRoute::Types(resource)); assert_eq!( &[0, 1], From 3ab75fbd0a6a7b9dfcfa7794b992951532053ab7 Mon Sep 17 00:00:00 2001 From: Snow Pettersen Date: Tue, 11 Aug 2026 11:39:47 -0400 Subject: [PATCH 13/23] comment --- bd-workflows/src/routing.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/bd-workflows/src/routing.rs b/bd-workflows/src/routing.rs index 88c8c6fca..b622f3afc 100644 --- a/bd-workflows/src/routing.rs +++ b/bd-workflows/src/routing.rs @@ -49,7 +49,8 @@ pub enum WorkflowLogRoute { /// /// Non-matching logs are expected to be common, so selection and evaluation do not allocate. /// Rebuilds may reserve storage after configuration or workflow-state changes; that infrequent -/// cost keeps the log-processing hot path allocation-free. +/// cost keeps this additional routing layer allocation-free during the common case of log +/// processing. #[derive(Debug, Default)] pub struct WorkflowLogRouter { routes: Vec, From 4a757f5f63f02a34fa9104416228ee94600a134b Mon Sep 17 00:00:00 2001 From: Snow Pettersen Date: Tue, 11 Aug 2026 11:53:39 -0400 Subject: [PATCH 14/23] optimize replacement, remove manual sort --- bd-log-matcher/src/matcher.rs | 5 +++ bd-workflows/src/routing.rs | 54 ++++++++++++++++++-------------- bd-workflows/src/routing_test.rs | 14 ++++++--- 3 files changed, 44 insertions(+), 29 deletions(-) diff --git a/bd-log-matcher/src/matcher.rs b/bd-log-matcher/src/matcher.rs index 67b723062..6ddd72171 100644 --- a/bd-log-matcher/src/matcher.rs +++ b/bd-log-matcher/src/matcher.rs @@ -95,6 +95,11 @@ impl LogTypeSet { self.0 &= other.0; } + #[must_use] + pub fn difference(self, other: Self) -> Self { + Self(self.0 & !other.0) + } + pub fn iter(self) -> impl Iterator { let mut bits = self.0; std::iter::from_fn(move || { diff --git a/bd-workflows/src/routing.rs b/bd-workflows/src/routing.rs index b622f3afc..0cd0c4cc8 100644 --- a/bd-workflows/src/routing.rs +++ b/bd-workflows/src/routing.rs @@ -11,6 +11,7 @@ mod routing_test; use bd_log_matcher::matcher::LogTypeSet; use bd_proto::protos::logging::payload::LogType; +use itertools::Itertools; use protobuf::Enum; const LOG_TYPE_BUCKET_COUNT: usize = LogType::VALUES.len(); @@ -83,15 +84,24 @@ impl WorkflowLogRouter { /// Selects the workflows which need to evaluate a log with this type. /// + /// The candidate set is the ordered union of type-specific workflows and fallback workflows, + /// which must inspect every log. Keeping the result in workflow-index order preserves the + /// original engine evaluation order. + /// + /// `prepare` reserves enough space for every workflow, so clearing and merging this scratch + /// vector reuses its capacity without allocating while processing a log. + /// /// The returned indices remain valid until the next selection or route refresh. pub(crate) fn select_candidates_for_log_type(&mut self, log_type: LogType) -> &[usize] { self.candidate_indices.clear(); let type_indices = &self.type_buckets[log_type as usize]; - merge_sorted_indices( - &self.fallback_workflow_indices, - type_indices, - &mut self.candidate_indices, + self.candidate_indices.extend( + self + .fallback_workflow_indices + .iter() + .copied() + .merge(type_indices.iter().copied()), ); &self.candidate_indices } @@ -119,11 +129,25 @@ impl WorkflowLogRouter { return; } - self.remove_route_from_buckets(workflow_index, previous_route); + // An active workflow frequently advances from one type-constrained state to another. Keep + // membership in the shared buckets and update only the types that changed. + if let (WorkflowLogRoute::Types(previous_types), WorkflowLogRoute::Types(next_types)) = + (previous_route, route) + { + for log_type in previous_types.difference(next_types).iter() { + remove_sorted(&mut self.type_buckets[log_type as usize], workflow_index); + } + for log_type in next_types.difference(previous_types).iter() { + insert_sorted(&mut self.type_buckets[log_type as usize], workflow_index); + } + } else { + self.remove_route_from_buckets(workflow_index, previous_route); + self.add_route_to_buckets(workflow_index, route); + } + if let Some(previous_route) = self.routes.get_mut(workflow_index) { *previous_route = route; } - self.add_route_to_buckets(workflow_index, route); } fn add_route_to_buckets(&mut self, workflow_index: usize, route: WorkflowLogRoute) { @@ -173,21 +197,3 @@ fn remove_sorted(values: &mut Vec, value: usize) { values.remove(position); } } - -fn merge_sorted_indices(first: &[usize], second: &[usize], output: &mut Vec) { - let mut first = first.iter().copied().peekable(); - let mut second = second.iter().copied().peekable(); - - while let (Some(first_index), Some(second_index)) = (first.peek(), second.peek()) { - if first_index < second_index { - if let Some(index) = first.next() { - output.push(index); - } - } else if let Some(index) = second.next() { - output.push(index); - } - } - - output.extend(first); - output.extend(second); -} diff --git a/bd-workflows/src/routing_test.rs b/bd-workflows/src/routing_test.rs index f9420d27c..219914310 100644 --- a/bd-workflows/src/routing_test.rs +++ b/bd-workflows/src/routing_test.rs @@ -50,19 +50,19 @@ fn selects_only_matching_and_fallback_workflows_in_index_order() { #[test] fn refreshes_only_the_workflows_selected_for_a_log() { - let lifecycle = log_types(&[LogType::LIFECYCLE]); - let resource = log_types(&[LogType::RESOURCE]); + let lifecycle_or_resource = log_types(&[LogType::LIFECYCLE, LogType::RESOURCE]); + let resource_or_normal = log_types(&[LogType::RESOURCE, LogType::NORMAL]); let mut router = WorkflowLogRouter::default(); router.prepare(3); - router.append_workflow_route(0, WorkflowLogRoute::Types(lifecycle)); + router.append_workflow_route(0, WorkflowLogRoute::Types(lifecycle_or_resource)); router.append_workflow_route(1, WorkflowLogRoute::Fallback); - router.append_workflow_route(2, WorkflowLogRoute::Types(resource)); + router.append_workflow_route(2, WorkflowLogRoute::Types(resource_or_normal)); assert_eq!( &[0, 1], router.select_candidates_for_log_type(LogType::LIFECYCLE) ); - let refreshed_routes = [WorkflowLogRoute::Types(resource), WorkflowLogRoute::None]; + let refreshed_routes = [WorkflowLogRoute::Types(resource_or_normal), WorkflowLogRoute::None]; router.refresh_selected_routes(|index| refreshed_routes[index]); assert_eq!( @@ -73,4 +73,8 @@ fn refreshes_only_the_workflows_selected_for_a_log() { &[0, 2], router.select_candidates_for_log_type(LogType::RESOURCE) ); + assert_eq!( + &[0, 2], + router.select_candidates_for_log_type(LogType::NORMAL) + ); } From cfff39208518e0fc3f2ffc5498142469122d44f3 Mon Sep 17 00:00:00 2001 From: Snow Pettersen Date: Tue, 11 Aug 2026 12:10:38 -0400 Subject: [PATCH 15/23] fmt --- bd-workflows/src/routing_test.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/bd-workflows/src/routing_test.rs b/bd-workflows/src/routing_test.rs index 219914310..fba2aa7bd 100644 --- a/bd-workflows/src/routing_test.rs +++ b/bd-workflows/src/routing_test.rs @@ -62,7 +62,10 @@ fn refreshes_only_the_workflows_selected_for_a_log() { &[0, 1], router.select_candidates_for_log_type(LogType::LIFECYCLE) ); - let refreshed_routes = [WorkflowLogRoute::Types(resource_or_normal), WorkflowLogRoute::None]; + let refreshed_routes = [ + WorkflowLogRoute::Types(resource_or_normal), + WorkflowLogRoute::None, + ]; router.refresh_selected_routes(|index| refreshed_routes[index]); assert_eq!( From 1babc5923f070ac63f474b096e3ba902dbf3877f Mon Sep 17 00:00:00 2001 From: Snow Pettersen Date: Tue, 11 Aug 2026 12:31:01 -0400 Subject: [PATCH 16/23] Avoid unchecked workflow route bucket access Co-Authored-By: GPT-5 --- bd-workflows/src/routing.rs | 37 ++++++++++++++++++++++++------------- 1 file changed, 24 insertions(+), 13 deletions(-) diff --git a/bd-workflows/src/routing.rs b/bd-workflows/src/routing.rs index 0cd0c4cc8..093e4ccdb 100644 --- a/bd-workflows/src/routing.rs +++ b/bd-workflows/src/routing.rs @@ -94,7 +94,10 @@ impl WorkflowLogRouter { /// The returned indices remain valid until the next selection or route refresh. pub(crate) fn select_candidates_for_log_type(&mut self, log_type: LogType) -> &[usize] { self.candidate_indices.clear(); - let type_indices = &self.type_buckets[log_type as usize]; + let type_indices = self + .type_buckets + .get(log_type as usize) + .map_or(&[] as &[usize], Vec::as_slice); self.candidate_indices.extend( self @@ -134,12 +137,8 @@ impl WorkflowLogRouter { if let (WorkflowLogRoute::Types(previous_types), WorkflowLogRoute::Types(next_types)) = (previous_route, route) { - for log_type in previous_types.difference(next_types).iter() { - remove_sorted(&mut self.type_buckets[log_type as usize], workflow_index); - } - for log_type in next_types.difference(previous_types).iter() { - insert_sorted(&mut self.type_buckets[log_type as usize], workflow_index); - } + self.remove_types_from_buckets(workflow_index, previous_types.difference(next_types)); + self.add_types_to_buckets(workflow_index, next_types.difference(previous_types)); } else { self.remove_route_from_buckets(workflow_index, previous_route); self.add_route_to_buckets(workflow_index, route); @@ -157,9 +156,7 @@ impl WorkflowLogRouter { insert_sorted(&mut self.fallback_workflow_indices, workflow_index); }, WorkflowLogRoute::Types(log_types) => { - for log_type in log_types.iter() { - insert_sorted(&mut self.type_buckets[log_type as usize], workflow_index); - } + self.add_types_to_buckets(workflow_index, log_types); }, } } @@ -171,12 +168,26 @@ impl WorkflowLogRouter { remove_sorted(&mut self.fallback_workflow_indices, workflow_index); }, WorkflowLogRoute::Types(log_types) => { - for log_type in log_types.iter() { - remove_sorted(&mut self.type_buckets[log_type as usize], workflow_index); - } + self.remove_types_from_buckets(workflow_index, log_types); }, } } + + fn add_types_to_buckets(&mut self, workflow_index: usize, log_types: LogTypeSet) { + for log_type in log_types.iter() { + if let Some(bucket) = self.type_buckets.get_mut(log_type as usize) { + insert_sorted(bucket, workflow_index); + } + } + } + + fn remove_types_from_buckets(&mut self, workflow_index: usize, log_types: LogTypeSet) { + for log_type in log_types.iter() { + if let Some(bucket) = self.type_buckets.get_mut(log_type as usize) { + remove_sorted(bucket, workflow_index); + } + } + } } fn reserve_to(values: &mut Vec, capacity: usize) { From 5398a0e3ccb4bd131e9c166b0aef469a4964d007 Mon Sep 17 00:00:00 2001 From: Snow Pettersen Date: Tue, 11 Aug 2026 12:34:42 -0400 Subject: [PATCH 17/23] clippy --- bd-logger/src/builder.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/bd-logger/src/builder.rs b/bd-logger/src/builder.rs index 1df4894f3..9f25e387a 100644 --- a/bd-logger/src/builder.rs +++ b/bd-logger/src/builder.rs @@ -619,7 +619,7 @@ impl LoggerBuilder { Ok(()) }, async move { - async_log_buffer.run(state_store, crash_monitor).await; + Box::pin(async_log_buffer.run(state_store, crash_monitor)).await; Ok(()) }, async move { From 768e2865bb0c455a4b10c3acb0f65a642b6fb493 Mon Sep 17 00:00:00 2001 From: Snow Pettersen Date: Tue, 11 Aug 2026 15:36:03 -0400 Subject: [PATCH 18/23] Route workflow events through candidate buckets Route logs by inferred log type and route state changes and session starts to workflows whose active transitions can consume them. Keep the index-based router synchronized with the workflow vector, refresh only selected routes, and cover bucket updates and fallback behavior. Co-Authored-By: GPT-5 --- bd-workflows/src/engine.rs | 98 ++++++------- bd-workflows/src/engine_test.rs | 78 +++++----- bd-workflows/src/routing.rs | 243 ++++++++++++++++++++++++++----- bd-workflows/src/routing_test.rs | 157 ++++++++++++++------ bd-workflows/src/workflow.rs | 65 +++++---- 5 files changed, 438 insertions(+), 203 deletions(-) diff --git a/bd-workflows/src/engine.rs b/bd-workflows/src/engine.rs index f80373bad..0c155dd0b 100644 --- a/bd-workflows/src/engine.rs +++ b/bd-workflows/src/engine.rs @@ -36,7 +36,7 @@ use crate::config::{ WorkflowsConfiguration, }; use crate::metrics::MetricsCollector; -use crate::routing::WorkflowLogRouter; +use crate::routing::{SelectedWorkflows, WorkflowEventKind, WorkflowLogRouter}; use crate::sankey_diagram::{self, PendingSankeyPathUpload}; use crate::workflow::{ SankeyPath, @@ -341,7 +341,7 @@ impl WorkflowsEngine { } self.state.refresh_tracing_counts_from_state(); - Self::rebuild_log_routes(&mut self.log_router, &self.state.workflows, &self.configs); + Self::rebuild_event_routes(&mut self.log_router, &self.state.workflows, &self.configs); log::debug!( "started workflows engine with {} workflow(s); {} pending processing action(s); {} pending \ @@ -470,7 +470,7 @@ impl WorkflowsEngine { .flush_buffers_actions_resolver .standardize_streaming_buffers(self.state.streaming_actions.clone()); // Regenerate routes whenever the workflow configuration is reloaded. - Self::rebuild_log_routes(&mut self.log_router, &self.state.workflows, &self.configs); + Self::rebuild_event_routes(&mut self.log_router, &self.state.workflows, &self.configs); log::debug!( "consumed received workflows config update; workflows engine contains {} workflow(s)", @@ -638,8 +638,8 @@ impl WorkflowsEngine { .extend(workflow_tracing_carryover_flush_action_ids); } - /// Rebuilds routes after an event or configuration change that can affect every workflow. - fn rebuild_log_routes( + /// Rebuilds routes after a configuration change that can affect every workflow. + fn rebuild_event_routes( log_router: &mut WorkflowLogRouter, workflows: &[Workflow], configs: &[Config], @@ -647,10 +647,24 @@ impl WorkflowsEngine { log_router.prepare(workflows.len()); for (index, (workflow, config)) in workflows.iter().zip(configs).enumerate() { - log_router.append_workflow_route(index, workflow.log_route(config)); + log_router.append_workflow_route(index, workflow.event_route(config)); } } + fn refresh_selected_event_routes( + selected_workflows: SelectedWorkflows<'_>, + workflows: &[Workflow], + configs: &[Config], + ) { + selected_workflows.refresh_routes(|index| { + workflows + .get(index) + .zip(configs.get(index)) + .map(|(workflow, config)| workflow.event_route(config)) + .unwrap_or_default() + }); + } + /// Attempts to persist the client-side state of the workflows to disk while ignoring any errors. /// Does nothing if state has not changed since the last time it was saved. /// @@ -913,60 +927,34 @@ impl WorkflowsEngine { .. } = self; let workflows = &mut state.workflows; - let workflow_count = workflows.len(); let active_run_tracing_count = &mut state.active_run_tracing_count; - let routed_log = matches!(event, WorkflowEvent::Log(_)); - - // A routed log only evaluates the indices selected by the router. Other event kinds are - // broadcast because they can advance any workflow or change every route. - match event { - WorkflowEvent::Log(log) => { - for &index in log_router.select_candidates_for_log_type(log.log_type) { - Self::process_workflow_event( - configs, - workflows, - stats, - needs_state_persistence, - active_run_tracing_count, - index, - event, - state_reader, - now, - sampled_roll, - &mut workflow_event_output, - ); - } + + let selected_workflows = match event { + WorkflowEvent::Log(log) => log_router.select_candidates_for_log_type(log.log_type), + WorkflowEvent::StateChange(..) => { + log_router.select_event_candidates(WorkflowEventKind::StateChange) }, - WorkflowEvent::SessionStart(_) | WorkflowEvent::StateChange(..) => { - for index in 0 .. workflow_count { - Self::process_workflow_event( - configs, - workflows, - stats, - needs_state_persistence, - active_run_tracing_count, - index, - event, - state_reader, - now, - sampled_roll, - &mut workflow_event_output, - ); - } + WorkflowEvent::SessionStart(_) => { + log_router.select_event_candidates(WorkflowEventKind::SessionStart) }, - } + }; - if routed_log { - log_router.refresh_selected_routes(|index| { - workflows - .get(index) - .zip(configs.get(index)) - .map(|(workflow, config)| workflow.log_route(config)) - .unwrap_or_default() - }); - } else { - Self::rebuild_log_routes(log_router, workflows, configs); + for &index in selected_workflows.indices() { + Self::process_workflow_event( + configs, + workflows, + stats, + needs_state_persistence, + active_run_tracing_count, + index, + event, + state_reader, + now, + sampled_roll, + &mut workflow_event_output, + ); } + Self::refresh_selected_event_routes(selected_workflows, workflows, configs); let WorkflowEventOutput { prepared_actions, diff --git a/bd-workflows/src/engine_test.rs b/bd-workflows/src/engine_test.rs index f56749883..99ddd52f5 100644 --- a/bd-workflows/src/engine_test.rs +++ b/bd-workflows/src/engine_test.rs @@ -3504,23 +3504,21 @@ async fn log_type_router_updates_after_workflow_advances() { engine.process_log(TestLog::new("lifecycle").with_log_type(LogType::LIFECYCLE)); engine_assert_active_runs!(engine; 0; "B"); - assert_eq!( - &[0], - engine - .engine - .log_router - .select_candidates_for_log_type(LogType::NORMAL) - ); + let selected = engine + .engine + .log_router + .select_candidates_for_log_type(LogType::NORMAL); + assert_eq!(&[0], selected.indices()); + selected.finish_without_route_refresh(); engine.process_log(TestLog::new("resource").with_log_type(LogType::RESOURCE)); engine_assert_active_runs!(engine; 0; "A"); - assert_eq!( - &[] as &[usize], - engine - .engine - .log_router - .select_candidates_for_log_type(LogType::NORMAL) - ); + let selected = engine + .engine + .log_router + .select_candidates_for_log_type(LogType::NORMAL); + assert_eq!(&[] as &[usize], selected.indices()); + selected.finish_without_route_refresh(); setup .collector .assert_counter_eq(2, "workflows:matched_logs_total", labels! {}); @@ -3543,25 +3541,23 @@ async fn log_type_router_recreates_initial_run_after_completion() { // LIFECYCLE. engine.process_log(TestLog::new("normal").with_log_type(LogType::NORMAL)); engine_assert_active_runs!(engine; 0; "A"); - assert_eq!( - &[] as &[usize], - engine - .engine - .log_router - .select_candidates_for_log_type(LogType::NORMAL) - ); + let selected = engine + .engine + .log_router + .select_candidates_for_log_type(LogType::NORMAL); + assert_eq!(&[] as &[usize], selected.indices()); + selected.finish_without_route_refresh(); // Completing the run leaves no initial run. The router must fall back so the next log can // recreate one. engine.process_log(TestLog::new("lifecycle").with_log_type(LogType::LIFECYCLE)); assert!(engine.engine.state.workflows[0].runs().is_empty()); - assert_eq!( - &[0], - engine - .engine - .log_router - .select_candidates_for_log_type(LogType::NORMAL) - ); + let selected = engine + .engine + .log_router + .select_candidates_for_log_type(LogType::NORMAL); + assert_eq!(&[0], selected.indices()); + selected.finish_without_route_refresh(); engine.process_log(TestLog::new("normal").with_log_type(LogType::NORMAL)); engine_assert_active_runs!(engine; 0; "A"); @@ -3589,13 +3585,12 @@ async fn log_type_router_keeps_debug_workflows_on_the_fallback_route() { .has_debug_workflows ); - assert_eq!( - &[0], - engine - .engine - .log_router - .select_candidates_for_log_type(LogType::NORMAL) - ); + let selected = engine + .engine + .log_router + .select_candidates_for_log_type(LogType::NORMAL); + assert_eq!(&[0], selected.indices()); + selected.finish_without_route_refresh(); } #[tokio::test] @@ -3614,11 +3609,10 @@ async fn log_type_router_keeps_active_timeouts_on_the_fallback_route() { engine.process_log(TestLog::new("lifecycle").with_log_type(LogType::LIFECYCLE)); - assert_eq!( - &[0], - engine - .engine - .log_router - .select_candidates_for_log_type(LogType::NORMAL) - ); + let selected = engine + .engine + .log_router + .select_candidates_for_log_type(LogType::NORMAL); + assert_eq!(&[0], selected.indices()); + selected.finish_without_route_refresh(); } diff --git a/bd-workflows/src/routing.rs b/bd-workflows/src/routing.rs index 093e4ccdb..e6adf291c 100644 --- a/bd-workflows/src/routing.rs +++ b/bd-workflows/src/routing.rs @@ -15,28 +15,87 @@ use itertools::Itertools; use protobuf::Enum; const LOG_TYPE_BUCKET_COUNT: usize = LogType::VALUES.len(); +const EVENT_BUCKET_COUNT: usize = 2; // -// WorkflowLogRoute +// WorkflowEventKind // -// TODO(snowp): We should be able to apply this routing logic to other event types to dramatically -// reduce the number of workflows attempted for state events. For now state events are rare so we -// limit this to logs. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum WorkflowEventKind { + StateChange, + SessionStart, +} + +impl WorkflowEventKind { + const ALL: [Self; EVENT_BUCKET_COUNT] = [Self::StateChange, Self::SessionStart]; + + const fn index(self) -> usize { + match self { + Self::StateChange => 0, + Self::SessionStart => 1, + } + } +} + +// +// WorkflowEventRoute +// + +/// Describes which events can require a workflow to process an event. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct WorkflowEventRoute { + log: WorkflowLogRoute, + event_mask: u8, +} -/// Describes which log types can require a workflow to process a log. #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] pub enum WorkflowLogRoute { - /// No active traversal can react to a log. This can happen if the workflow is matching on a state - /// change. #[default] None, - /// The workflow must inspect every log. Fallback, - /// The workflow only needs logs whose type is included in the set. Types(LogTypeSet), } +impl WorkflowEventRoute { + #[must_use] + pub(crate) const fn fallback() -> Self { + Self { + log: WorkflowLogRoute::Fallback, + event_mask: Self::all_event_mask(), + } + } + + #[cfg(test)] + #[must_use] + pub(crate) const fn from_log_types(log_types: LogTypeSet) -> Self { + Self { + log: WorkflowLogRoute::Types(log_types), + event_mask: 0, + } + } + + pub(crate) fn add_event(&mut self, event: WorkflowEventKind) { + self.event_mask |= Self::event_mask(event); + } + + pub(crate) const fn set_log_route(&mut self, log: WorkflowLogRoute) { + self.log = log; + } + + const fn needs_event(self, event: WorkflowEventKind) -> bool { + self.event_mask & Self::event_mask(event) != 0 + } + + const fn event_mask(event: WorkflowEventKind) -> u8 { + 1 << event.index() + } + + const fn all_event_mask() -> u8 { + (1 << EVENT_BUCKET_COUNT) - 1 + } +} + // // WorkflowLogRouter // @@ -45,8 +104,8 @@ pub enum WorkflowLogRoute { /// /// This is a derived view over the engine's primary workflow vector: every stored index refers to /// the workflow at that same position. The caller must rebuild it after adding, removing, or -/// reordering workflows; between those boundaries, the router updates only workflows which -/// processed the current log. +/// reordering workflows; between those boundaries, it refreshes only the routes for workflows +/// selected to process the current event. /// /// Non-matching logs are expected to be common, so selection and evaluation do not allocate. /// Rebuilds may reserve storage after configuration or workflow-state changes; that infrequent @@ -54,12 +113,70 @@ pub enum WorkflowLogRoute { /// processing. #[derive(Debug, Default)] pub struct WorkflowLogRouter { - routes: Vec, + routes: Vec, fallback_workflow_indices: Vec, type_buckets: [Vec; LOG_TYPE_BUCKET_COUNT], + event_buckets: [Vec; EVENT_BUCKET_COUNT], candidate_indices: Vec, } +// +// SelectedWorkflows +// + +/// A selected candidate set that must be finalized before processing the next event. +/// +/// Holding this guard prevents the router from being updated while the engine is evaluating the +/// selected workflows. Its debug-only drop assertion catches future early returns that would leave +/// the derived routes stale. +#[must_use = "selected workflow candidates must be finalized after processing"] +pub struct SelectedWorkflows<'a> { + router: &'a mut WorkflowLogRouter, + finalized: bool, +} + +impl SelectedWorkflows<'_> { + #[must_use] + pub(crate) fn indices(&self) -> &[usize] { + &self.router.candidate_indices + } + + /// Finalizes a selection without recalculating any routes when processing left them unchanged. + #[cfg(test)] + pub(crate) fn finish_without_route_refresh(mut self) { + self.finalized = true; + } + + /// Recalculates routes for exactly the workflows selected for the preceding event. + pub(crate) fn refresh_routes(mut self, mut route_for_index: F) + where + F: FnMut(usize) -> WorkflowEventRoute, + { + for candidate_position in 0 .. self.router.candidate_indices.len() { + let Some(workflow_index) = self + .router + .candidate_indices + .get(candidate_position) + .copied() + else { + continue; + }; + let route = route_for_index(workflow_index); + self.router.replace(workflow_index, route); + } + self.finalized = true; + } +} + +impl Drop for SelectedWorkflows<'_> { + fn drop(&mut self) { + debug_assert!( + self.finalized, + "selected workflow candidates were not finalized" + ); + } +} + impl WorkflowLogRouter { /// Clears the derived routes before repopulating them in primary workflow-vector order. pub(crate) fn prepare(&mut self, workflow_count: usize) { @@ -74,9 +191,13 @@ impl WorkflowLogRouter { bucket.clear(); reserve_to(bucket, workflow_count); } + for bucket in &mut self.event_buckets { + bucket.clear(); + reserve_to(bucket, workflow_count); + } } - pub(crate) fn append_workflow_route(&mut self, workflow_index: usize, route: WorkflowLogRoute) { + pub(crate) fn append_workflow_route(&mut self, workflow_index: usize, route: WorkflowEventRoute) { debug_assert_eq!(workflow_index, self.routes.len()); self.routes.push(route); self.add_route_to_buckets(workflow_index, route); @@ -90,9 +211,10 @@ impl WorkflowLogRouter { /// /// `prepare` reserves enough space for every workflow, so clearing and merging this scratch /// vector reuses its capacity without allocating while processing a log. - /// - /// The returned indices remain valid until the next selection or route refresh. - pub(crate) fn select_candidates_for_log_type(&mut self, log_type: LogType) -> &[usize] { + pub(crate) fn select_candidates_for_log_type( + &mut self, + log_type: LogType, + ) -> SelectedWorkflows<'_> { self.candidate_indices.clear(); let type_indices = self .type_buckets @@ -106,24 +228,28 @@ impl WorkflowLogRouter { .copied() .merge(type_indices.iter().copied()), ); - &self.candidate_indices + SelectedWorkflows { + router: self, + finalized: false, + } } - /// Updates routes for the workflows selected by the preceding log selection. - pub(crate) fn refresh_selected_routes(&mut self, mut route_for_index: F) - where - F: FnMut(usize) -> WorkflowLogRoute, - { - for candidate_position in 0 .. self.candidate_indices.len() { - let Some(workflow_index) = self.candidate_indices.get(candidate_position).copied() else { - continue; - }; - let route = route_for_index(workflow_index); - self.replace(workflow_index, route); + /// Selects only the workflows which need to evaluate this non-log event. + pub(crate) fn select_event_candidates( + &mut self, + event: WorkflowEventKind, + ) -> SelectedWorkflows<'_> { + self.candidate_indices.clear(); + if let Some(indices) = self.event_buckets.get(event.index()) { + self.candidate_indices.extend(indices); + } + SelectedWorkflows { + router: self, + finalized: false, } } - fn replace(&mut self, workflow_index: usize, route: WorkflowLogRoute) { + fn replace(&mut self, workflow_index: usize, route: WorkflowEventRoute) { let Some(previous_route) = self.routes.get(workflow_index).copied() else { return; }; @@ -132,24 +258,63 @@ impl WorkflowLogRouter { return; } + self.replace_log_route(workflow_index, previous_route.log, route.log); + for event in WorkflowEventKind::ALL { + match (previous_route.needs_event(event), route.needs_event(event)) { + (false, true) => { + if let Some(bucket) = self.event_buckets.get_mut(event.index()) { + insert_sorted(bucket, workflow_index); + } + }, + (true, false) => { + if let Some(bucket) = self.event_buckets.get_mut(event.index()) { + remove_sorted(bucket, workflow_index); + } + }, + (false, false) | (true, true) => {}, + } + } + + if let Some(previous_route) = self.routes.get_mut(workflow_index) { + *previous_route = route; + } + } + + fn add_route_to_buckets(&mut self, workflow_index: usize, route: WorkflowEventRoute) { + self.add_log_route_to_buckets(workflow_index, route.log); + for event in WorkflowEventKind::ALL { + if route.needs_event(event) + && let Some(bucket) = self.event_buckets.get_mut(event.index()) + { + insert_sorted(bucket, workflow_index); + } + } + } + + fn replace_log_route( + &mut self, + workflow_index: usize, + previous_route: WorkflowLogRoute, + route: WorkflowLogRoute, + ) { + if previous_route == route { + return; + } + // An active workflow frequently advances from one type-constrained state to another. Keep - // membership in the shared buckets and update only the types that changed. + // membership in shared log-type buckets and update only the types that changed. if let (WorkflowLogRoute::Types(previous_types), WorkflowLogRoute::Types(next_types)) = (previous_route, route) { self.remove_types_from_buckets(workflow_index, previous_types.difference(next_types)); self.add_types_to_buckets(workflow_index, next_types.difference(previous_types)); } else { - self.remove_route_from_buckets(workflow_index, previous_route); - self.add_route_to_buckets(workflow_index, route); - } - - if let Some(previous_route) = self.routes.get_mut(workflow_index) { - *previous_route = route; + self.remove_log_route_from_buckets(workflow_index, previous_route); + self.add_log_route_to_buckets(workflow_index, route); } } - fn add_route_to_buckets(&mut self, workflow_index: usize, route: WorkflowLogRoute) { + fn add_log_route_to_buckets(&mut self, workflow_index: usize, route: WorkflowLogRoute) { match route { WorkflowLogRoute::None => {}, WorkflowLogRoute::Fallback => { @@ -161,7 +326,7 @@ impl WorkflowLogRouter { } } - fn remove_route_from_buckets(&mut self, workflow_index: usize, route: WorkflowLogRoute) { + fn remove_log_route_from_buckets(&mut self, workflow_index: usize, route: WorkflowLogRoute) { match route { WorkflowLogRoute::None => {}, WorkflowLogRoute::Fallback => { @@ -192,7 +357,7 @@ impl WorkflowLogRouter { fn reserve_to(values: &mut Vec, capacity: usize) { if values.capacity() < capacity { - values.reserve(capacity - values.capacity()); + values.reserve(capacity - values.len()); } } diff --git a/bd-workflows/src/routing_test.rs b/bd-workflows/src/routing_test.rs index fba2aa7bd..e3edbcf53 100644 --- a/bd-workflows/src/routing_test.rs +++ b/bd-workflows/src/routing_test.rs @@ -5,7 +5,7 @@ // LICENSE.polyform file or at: // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt -use super::{WorkflowLogRoute, WorkflowLogRouter}; +use super::{WorkflowEventKind, WorkflowEventRoute, WorkflowLogRouter}; use bd_log_matcher::matcher::LogTypeSet; use bd_proto::protos::logging::payload::LogType; use pretty_assertions::assert_eq; @@ -19,6 +19,14 @@ fn log_types(types: &[LogType]) -> LogTypeSet { log_types } +fn log_route(log_types: LogTypeSet) -> WorkflowEventRoute { + WorkflowEventRoute::from_log_types(log_types) +} + +fn no_route() -> WorkflowEventRoute { + WorkflowEventRoute::default() +} + #[test] fn log_type_values_are_dense_bucket_indices() { for (index, log_type) in LogType::VALUES.iter().enumerate() { @@ -32,20 +40,22 @@ fn selects_only_matching_and_fallback_workflows_in_index_order() { let lifecycle_or_resource = log_types(&[LogType::LIFECYCLE, LogType::RESOURCE]); let mut router = WorkflowLogRouter::default(); router.prepare(4); - router.append_workflow_route(0, WorkflowLogRoute::Types(lifecycle)); - router.append_workflow_route(1, WorkflowLogRoute::Fallback); - router.append_workflow_route(2, WorkflowLogRoute::Types(lifecycle_or_resource)); - router.append_workflow_route(3, WorkflowLogRoute::None); - - assert_eq!( - &[0, 1, 2], - router.select_candidates_for_log_type(LogType::LIFECYCLE) - ); - assert_eq!( - &[1, 2], - router.select_candidates_for_log_type(LogType::RESOURCE) - ); - assert_eq!(&[1], router.select_candidates_for_log_type(LogType::NORMAL)); + router.append_workflow_route(0, log_route(lifecycle)); + router.append_workflow_route(1, WorkflowEventRoute::fallback()); + router.append_workflow_route(2, log_route(lifecycle_or_resource)); + router.append_workflow_route(3, no_route()); + + let selected = router.select_candidates_for_log_type(LogType::LIFECYCLE); + assert_eq!(&[0, 1, 2], selected.indices()); + selected.finish_without_route_refresh(); + + let selected = router.select_candidates_for_log_type(LogType::RESOURCE); + assert_eq!(&[1, 2], selected.indices()); + selected.finish_without_route_refresh(); + + let selected = router.select_candidates_for_log_type(LogType::NORMAL); + assert_eq!(&[1], selected.indices()); + selected.finish_without_route_refresh(); } #[test] @@ -54,30 +64,95 @@ fn refreshes_only_the_workflows_selected_for_a_log() { let resource_or_normal = log_types(&[LogType::RESOURCE, LogType::NORMAL]); let mut router = WorkflowLogRouter::default(); router.prepare(3); - router.append_workflow_route(0, WorkflowLogRoute::Types(lifecycle_or_resource)); - router.append_workflow_route(1, WorkflowLogRoute::Fallback); - router.append_workflow_route(2, WorkflowLogRoute::Types(resource_or_normal)); - - assert_eq!( - &[0, 1], - router.select_candidates_for_log_type(LogType::LIFECYCLE) - ); - let refreshed_routes = [ - WorkflowLogRoute::Types(resource_or_normal), - WorkflowLogRoute::None, - ]; - router.refresh_selected_routes(|index| refreshed_routes[index]); - - assert_eq!( - &[] as &[usize], - router.select_candidates_for_log_type(LogType::LIFECYCLE) - ); - assert_eq!( - &[0, 2], - router.select_candidates_for_log_type(LogType::RESOURCE) - ); - assert_eq!( - &[0, 2], - router.select_candidates_for_log_type(LogType::NORMAL) - ); + router.append_workflow_route(0, log_route(lifecycle_or_resource)); + router.append_workflow_route(1, WorkflowEventRoute::fallback()); + router.append_workflow_route(2, log_route(resource_or_normal)); + + let selected = router.select_candidates_for_log_type(LogType::LIFECYCLE); + assert_eq!(&[0, 1], selected.indices()); + let refreshed_routes = [log_route(resource_or_normal), no_route()]; + selected.refresh_routes(|index| refreshed_routes[index]); + + let selected = router.select_candidates_for_log_type(LogType::LIFECYCLE); + assert_eq!(&[] as &[usize], selected.indices()); + selected.finish_without_route_refresh(); + + let selected = router.select_candidates_for_log_type(LogType::RESOURCE); + assert_eq!(&[0, 2], selected.indices()); + selected.finish_without_route_refresh(); + + let selected = router.select_candidates_for_log_type(LogType::NORMAL); + assert_eq!(&[0, 2], selected.indices()); + selected.finish_without_route_refresh(); +} + +#[test] +fn selects_only_event_candidates() { + let mut state_change = no_route(); + state_change.add_event(WorkflowEventKind::StateChange); + let mut session_start = no_route(); + session_start.add_event(WorkflowEventKind::SessionStart); + let mut router = WorkflowLogRouter::default(); + router.prepare(3); + router.append_workflow_route(0, state_change); + router.append_workflow_route(1, session_start); + router.append_workflow_route(2, WorkflowEventRoute::fallback()); + + let selected = router.select_event_candidates(WorkflowEventKind::StateChange); + assert_eq!(&[0, 2], selected.indices()); + selected.finish_without_route_refresh(); + + let selected = router.select_event_candidates(WorkflowEventKind::SessionStart); + assert_eq!(&[1, 2], selected.indices()); + selected.finish_without_route_refresh(); +} + +#[test] +fn refreshing_an_event_route_moves_it_between_event_buckets() { + let mut state_change = no_route(); + state_change.add_event(WorkflowEventKind::StateChange); + let mut session_start = no_route(); + session_start.add_event(WorkflowEventKind::SessionStart); + let mut router = WorkflowLogRouter::default(); + router.prepare(1); + router.append_workflow_route(0, state_change); + + let selected = router.select_event_candidates(WorkflowEventKind::StateChange); + assert_eq!(&[0], selected.indices()); + selected.refresh_routes(|_| session_start); + + let selected = router.select_event_candidates(WorkflowEventKind::StateChange); + assert_eq!(&[] as &[usize], selected.indices()); + selected.finish_without_route_refresh(); + + let selected = router.select_event_candidates(WorkflowEventKind::SessionStart); + assert_eq!(&[0], selected.indices()); + selected.finish_without_route_refresh(); +} + +#[test] +fn finish_without_route_refresh_leaves_routes_unchanged() { + let lifecycle = log_types(&[LogType::LIFECYCLE]); + let mut router = WorkflowLogRouter::default(); + router.prepare(1); + router.append_workflow_route(0, log_route(lifecycle)); + + let selected = router.select_candidates_for_log_type(LogType::LIFECYCLE); + selected.finish_without_route_refresh(); + + let selected = router.select_candidates_for_log_type(LogType::LIFECYCLE); + assert_eq!(&[0], selected.indices()); + selected.finish_without_route_refresh(); +} + +#[test] +fn reserve_to_reaches_the_requested_capacity_after_clear() { + let mut values = Vec::with_capacity(4); + values.extend([0, 1, 2, 3]); + values.clear(); + let requested_capacity = values.capacity() + 1; + + super::reserve_to(&mut values, requested_capacity); + + assert!(values.capacity() >= requested_capacity); } diff --git a/bd-workflows/src/workflow.rs b/bd-workflows/src/workflow.rs index ec202bfc7..7be6a1339 100644 --- a/bd-workflows/src/workflow.rs +++ b/bd-workflows/src/workflow.rs @@ -22,7 +22,7 @@ use crate::config::{ WorkflowDebugMode, }; use crate::generate_log::generate_log_action; -use crate::routing::WorkflowLogRoute; +use crate::routing::{WorkflowEventKind, WorkflowEventRoute, WorkflowLogRoute}; use bd_log_primitives::tiny_set::{TinyMap, TinySet}; use bd_log_primitives::{FieldsRef, Log, log_level}; use bd_proto::protos::logging::payload::LogType; @@ -247,48 +247,62 @@ impl WorkflowEvent<'_> { } // -// WorkflowLogRouteBuilder +// WorkflowEventRouteBuilder // #[derive(Default)] -struct WorkflowLogRouteBuilder { +struct WorkflowEventRouteBuilder { log_types: bd_log_matcher::matcher::LogTypeSet, needs_fallback: bool, + event_route: WorkflowEventRoute, } -impl WorkflowLogRouteBuilder { +impl WorkflowEventRouteBuilder { fn add_state(&mut self, config: &Config, state_index: usize) { let Some(state) = config.inner().states().get(state_index) else { return; }; - // A state timeout is checked when any event arrives, not only when a transition's matcher - // accepts that event. Keep such states on the fallback path. + // A timeout can expire while processing any supported event. Keep the log route on fallback + // and include both non-log event buckets. if state.timeout().is_some() { self.needs_fallback = true; + self.event_route.add_event(WorkflowEventKind::StateChange); + self.event_route.add_event(WorkflowEventKind::SessionStart); } for transition in state.transitions() { - let Predicate::LogMatch { matcher, .. } = transition.rule() else { - continue; - }; - - if let Some(log_types) = matcher.possible_log_types() { - self.log_types.union(log_types); - } else { - self.needs_fallback = true; + match transition.rule() { + Predicate::LogMatch { matcher, .. } => { + if let Some(log_types) = matcher.possible_log_types() { + self.log_types.union(log_types); + } else { + self.needs_fallback = true; + } + }, + // State changes use a dedicated candidate bucket so log-only workflows are skipped. + Predicate::StateChangeMatch { .. } => { + self.event_route.add_event(WorkflowEventKind::StateChange); + }, + Predicate::OnNewSession => { + self.event_route.add_event(WorkflowEventKind::SessionStart); + }, + Predicate::OnReport => {}, } } } - fn finish(self) -> WorkflowLogRoute { + fn finish(mut self) -> WorkflowEventRoute { if self.needs_fallback { - WorkflowLogRoute::Fallback + self.event_route.set_log_route(WorkflowLogRoute::Fallback); } else if self.log_types.is_empty() { - WorkflowLogRoute::None + self.event_route.set_log_route(WorkflowLogRoute::None); } else { - WorkflowLogRoute::Types(self.log_types) + self + .event_route + .set_log_route(WorkflowLogRoute::Types(self.log_types)); } + self.event_route } } @@ -376,23 +390,23 @@ impl Workflow { self.workflow_debug_state = None; } - /// Returns the log types that can advance the workflow's current state. + /// Returns the events that can advance the workflow's current state. /// /// This deliberately keeps debug workflows, active timeouts, and initial delivery state on the /// fallback route so their behavior remains unchanged while a debug session is active. A /// consequence of this is that we effectively only apply this to workflows in the initial /// condition, as in general all workflows have a total timeout. - pub(crate) fn log_route(&self, config: &Config) -> WorkflowLogRoute { + pub(crate) fn event_route(&self, config: &Config) -> WorkflowEventRoute { if self.needs_start_metric || config.mode() != WorkflowDebugMode::None { - return WorkflowLogRoute::Fallback; + return WorkflowEventRoute::fallback(); } - let mut route = WorkflowLogRouteBuilder::default(); + let mut route = WorkflowEventRouteBuilder::default(); // Workflow execution lazily creates an initial run while processing every log. It must remain // on the fallback path until that run exists, even when the initial state has no log matcher. if self.needs_new_run() { - return WorkflowLogRoute::Fallback; + return WorkflowEventRoute::fallback(); } for run in &self.runs { @@ -400,12 +414,12 @@ impl Workflow { // can remain on its type-specific route. This needs an expiry mechanism that preserves // run termination and tracing updates without evaluating every log. if run.first_progress_occurred_at.is_some() && config.inner().duration_limit().is_some() { - return WorkflowLogRoute::Fallback; + return WorkflowEventRoute::fallback(); } for traversal in &run.traversals { if traversal.timeout_unix_ms.is_some() { - return WorkflowLogRoute::Fallback; + return WorkflowEventRoute::fallback(); } route.add_state(config, traversal.state_index); } @@ -497,7 +511,6 @@ impl Workflow { } let run_did_make_progress = run_result.did_make_progress(); - (run_result, run_did_make_progress) }; From ebafe3b74dae644d922d104bc054b981cb25150c Mon Sep 17 00:00:00 2001 From: Snow Pettersen Date: Wed, 12 Aug 2026 14:59:21 -0400 Subject: [PATCH 19/23] Fix async log buffer test future lint Pin the oversized async log buffer future in the test task to satisfy Clippy large_futures. Co-Authored-By: GPT-5 --- bd-logger/src/async_log_buffer_test.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/bd-logger/src/async_log_buffer_test.rs b/bd-logger/src/async_log_buffer_test.rs index 19aac4755..8edf21c86 100644 --- a/bd-logger/src/async_log_buffer_test.rs +++ b/bd-logger/src/async_log_buffer_test.rs @@ -474,7 +474,7 @@ async fn logs_are_replayed_in_order() { let test_store = TestStore::new().await; let state_store = (*test_store).clone(); let run_buffer_task = tokio::task::spawn(async move { - _ = buffer.run(state_store, ()).await; + _ = Box::pin(buffer.run(state_store, ())).await; }); shutdown.store(true, Ordering::SeqCst); From 5ce2a258553e30948ad46a81e2de41cfc336dc8b Mon Sep 17 00:00:00 2001 From: Snow Pettersen Date: Wed, 12 Aug 2026 15:01:54 -0400 Subject: [PATCH 20/23] Assert workflow event routes are finalized once Co-Authored-By: GPT-5 --- bd-workflows/src/routing.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/bd-workflows/src/routing.rs b/bd-workflows/src/routing.rs index e6adf291c..86b10245f 100644 --- a/bd-workflows/src/routing.rs +++ b/bd-workflows/src/routing.rs @@ -79,7 +79,11 @@ impl WorkflowEventRoute { self.event_mask |= Self::event_mask(event); } - pub(crate) const fn set_log_route(&mut self, log: WorkflowLogRoute) { + pub(crate) fn set_log_route(&mut self, log: WorkflowLogRoute) { + debug_assert!( + matches!(self.log, WorkflowLogRoute::None), + "workflow event route log routing must be finalized exactly once" + ); self.log = log; } From fb4f61e2099d2ee852dda9f77204ea615445ef88 Mon Sep 17 00:00:00 2001 From: Snow Pettersen Date: Wed, 12 Aug 2026 15:13:48 -0400 Subject: [PATCH 21/23] Assert workflow route builders finish Co-Authored-By: GPT-5 --- bd-workflows/src/workflow.rs | 23 ++++++++++++++++++++--- 1 file changed, 20 insertions(+), 3 deletions(-) diff --git a/bd-workflows/src/workflow.rs b/bd-workflows/src/workflow.rs index 7be6a1339..a7224abd8 100644 --- a/bd-workflows/src/workflow.rs +++ b/bd-workflows/src/workflow.rs @@ -255,6 +255,7 @@ struct WorkflowEventRouteBuilder { log_types: bd_log_matcher::matcher::LogTypeSet, needs_fallback: bool, event_route: WorkflowEventRoute, + finished: bool, } impl WorkflowEventRouteBuilder { @@ -302,7 +303,17 @@ impl WorkflowEventRouteBuilder { .event_route .set_log_route(WorkflowLogRoute::Types(self.log_types)); } - self.event_route + self.finished = true; + std::mem::take(&mut self.event_route) + } +} + +impl Drop for WorkflowEventRouteBuilder { + fn drop(&mut self) { + debug_assert!( + self.finished, + "workflow event route builder must be finalized before it is dropped" + ); } } @@ -401,14 +412,14 @@ impl Workflow { return WorkflowEventRoute::fallback(); } - let mut route = WorkflowEventRouteBuilder::default(); - // Workflow execution lazily creates an initial run while processing every log. It must remain // on the fallback path until that run exists, even when the initial state has no log matcher. if self.needs_new_run() { return WorkflowEventRoute::fallback(); } + // Check the fallback conditions before constructing the builder. Its Drop implementation + // asserts that any builder which is created is finalized exactly once. for run in &self.runs { // TODO(snowp): Decouple duration expiry from inbound log processing so a progressed workflow // can remain on its type-specific route. This needs an expiry mechanism that preserves @@ -421,6 +432,12 @@ impl Workflow { if traversal.timeout_unix_ms.is_some() { return WorkflowEventRoute::fallback(); } + } + } + + let mut route = WorkflowEventRouteBuilder::default(); + for run in &self.runs { + for traversal in &run.traversals { route.add_state(config, traversal.state_index); } } From 1cbb6ee48e8ee1a9a99ea4f1de6e77795c6dd75e Mon Sep 17 00:00:00 2001 From: Snow Pettersen Date: Wed, 12 Aug 2026 15:18:24 -0400 Subject: [PATCH 22/23] Simplify workflow route builder finalization Co-Authored-By: GPT-5 --- bd-workflows/src/workflow.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/bd-workflows/src/workflow.rs b/bd-workflows/src/workflow.rs index a7224abd8..ff81169b2 100644 --- a/bd-workflows/src/workflow.rs +++ b/bd-workflows/src/workflow.rs @@ -304,7 +304,7 @@ impl WorkflowEventRouteBuilder { .set_log_route(WorkflowLogRoute::Types(self.log_types)); } self.finished = true; - std::mem::take(&mut self.event_route) + self.event_route } } From 8f91cabeba6f20791ccbeb098a42251cb4757258 Mon Sep 17 00:00:00 2001 From: Snow Pettersen Date: Wed, 12 Aug 2026 15:20:59 -0400 Subject: [PATCH 23/23] Clarify workflow route finalization assertion Co-Authored-By: GPT-5 --- bd-workflows/src/routing.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/bd-workflows/src/routing.rs b/bd-workflows/src/routing.rs index 86b10245f..466c99b2c 100644 --- a/bd-workflows/src/routing.rs +++ b/bd-workflows/src/routing.rs @@ -82,7 +82,7 @@ impl WorkflowEventRoute { pub(crate) fn set_log_route(&mut self, log: WorkflowLogRoute) { debug_assert!( matches!(self.log, WorkflowLogRoute::None), - "workflow event route log routing must be finalized exactly once" + "workflow event route log routing must set at most once" ); self.log = log; }