diff --git a/bd-log-matcher/src/matcher.rs b/bd-log-matcher/src/matcher.rs index ed35778e3..6ddd72171 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,69 @@ 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. 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); + +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; + } + + #[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 || { + 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 +239,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-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); 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 { diff --git a/bd-workflows/src/engine.rs b/bd-workflows/src/engine.rs index a549dd819..0c155dd0b 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::{SelectedWorkflows, WorkflowEventKind, WorkflowLogRouter}; use crate::sankey_diagram::{self, PendingSankeyPathUpload}; use crate::workflow::{ SankeyPath, @@ -114,12 +115,12 @@ 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, // 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 +220,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 +341,7 @@ impl WorkflowsEngine { } self.state.refresh_tracing_counts_from_state(); + 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 \ @@ -466,6 +469,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_event_routes(&mut self.log_router, &self.state.workflows, &self.configs); log::debug!( "consumed received workflows config update; workflows engine contains {} workflow(s)", @@ -548,6 +553,118 @@ 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); + output.has_debug_workflows |= config.mode() != WorkflowDebugMode::None; + + 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 a configuration change that can affect every workflow. + fn rebuild_event_routes( + log_router: &mut WorkflowLogRouter, + workflows: &[Workflow], + configs: &[Config], + ) { + log_router.prepare(workflows.len()); + + for (index, (workflow, config)) in workflows.iter().zip(configs).enumerate() { + 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. /// @@ -796,88 +913,57 @@ 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); - } - 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 active_run_tracing_count = &mut state.active_run_tracing_count; + + 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(_) => { + log_router.select_event_candidates(WorkflowEventKind::SessionStart) + }, + }; - // 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, - }), + 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, ); - tracing_carryover_flush_action_ids.extend(workflow_tracing_carryover_flush_action_ids); } - + Self::refresh_selected_event_routes(selected_workflows, workflows, configs); + + let WorkflowEventOutput { + prepared_actions, + logs_to_inject, + 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, emit_metric_action_counts, @@ -1054,6 +1140,16 @@ 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, + has_debug_workflows: bool, +} + #[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..99ddd52f5 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,132 @@ 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"); + 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"); + 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! {}); +} + +#[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"); + 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()); + 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"); +} + +#[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. + assert!( + engine + .process_log(TestLog::new("normal").with_log_type(LogType::NORMAL)) + .has_debug_workflows + ); + + 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] +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)); + + 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/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..466c99b2c --- /dev/null +++ b/bd-workflows/src/routing.rs @@ -0,0 +1,379 @@ +// 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.polyform 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; + +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(); +const EVENT_BUCKET_COUNT: usize = 2; + +// +// WorkflowEventKind +// + +#[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, +} + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub enum WorkflowLogRoute { + #[default] + None, + Fallback, + 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) fn set_log_route(&mut self, log: WorkflowLogRoute) { + debug_assert!( + matches!(self.log, WorkflowLogRoute::None), + "workflow event route log routing must set at most once" + ); + 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 +// + +/// Maintains sorted workflow indices for efficient, allocation-free log routing. +/// +/// 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, 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 +/// cost keeps this additional routing layer allocation-free during the common case of log +/// processing. +#[derive(Debug, Default)] +pub struct WorkflowLogRouter { + 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) { + 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 bucket in &mut self.type_buckets { + 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: WorkflowEventRoute) { + debug_assert_eq!(workflow_index, self.routes.len()); + self.routes.push(route); + self.add_route_to_buckets(workflow_index, route); + } + + /// 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. + pub(crate) fn select_candidates_for_log_type( + &mut self, + log_type: LogType, + ) -> SelectedWorkflows<'_> { + self.candidate_indices.clear(); + let type_indices = self + .type_buckets + .get(log_type as usize) + .map_or(&[] as &[usize], Vec::as_slice); + + self.candidate_indices.extend( + self + .fallback_workflow_indices + .iter() + .copied() + .merge(type_indices.iter().copied()), + ); + SelectedWorkflows { + router: self, + finalized: false, + } + } + + /// 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: WorkflowEventRoute) { + let Some(previous_route) = self.routes.get(workflow_index).copied() else { + return; + }; + + if previous_route == route { + 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 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_log_route_from_buckets(workflow_index, previous_route); + self.add_log_route_to_buckets(workflow_index, route); + } + } + + fn add_log_route_to_buckets(&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) => { + self.add_types_to_buckets(workflow_index, log_types); + }, + } + } + + fn remove_log_route_from_buckets(&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) => { + 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) { + if values.capacity() < capacity { + values.reserve(capacity - values.len()); + } +} + +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); + } +} diff --git a/bd-workflows/src/routing_test.rs b/bd-workflows/src/routing_test.rs new file mode 100644 index 000000000..e3edbcf53 --- /dev/null +++ b/bd-workflows/src/routing_test.rs @@ -0,0 +1,158 @@ +// 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.polyform file or at: +// https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt + +use super::{WorkflowEventKind, WorkflowEventRoute, 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(); + for log_type in types { + log_types.union(LogTypeSet::from_log_type(*log_type)); + } + 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() { + assert_eq!(index, *log_type as usize); + } +} + +#[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); + 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] +fn refreshes_only_the_workflows_selected_for_a_log() { + 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, 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/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..ff81169b2 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::{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; @@ -245,6 +246,77 @@ impl WorkflowEvent<'_> { } } +// +// WorkflowEventRouteBuilder +// + +#[derive(Default)] +struct WorkflowEventRouteBuilder { + log_types: bd_log_matcher::matcher::LogTypeSet, + needs_fallback: bool, + event_route: WorkflowEventRoute, + finished: bool, +} + +impl WorkflowEventRouteBuilder { + fn add_state(&mut self, config: &Config, state_index: usize) { + let Some(state) = config.inner().states().get(state_index) else { + return; + }; + + // 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() { + 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(mut self) -> WorkflowEventRoute { + if self.needs_fallback { + self.event_route.set_log_route(WorkflowLogRoute::Fallback); + } else if self.log_types.is_empty() { + self.event_route.set_log_route(WorkflowLogRoute::None); + } else { + self + .event_route + .set_log_route(WorkflowLogRoute::Types(self.log_types)); + } + self.finished = true; + 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" + ); + } +} + // // Workflow // @@ -329,6 +401,50 @@ impl Workflow { self.workflow_debug_state = None; } + /// 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 event_route(&self, config: &Config) -> WorkflowEventRoute { + if self.needs_start_metric || config.mode() != WorkflowDebugMode::None { + return WorkflowEventRoute::fallback(); + } + + // 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 + // run termination and tracing updates without evaluating every log. + if run.first_progress_occurred_at.is_some() && config.inner().duration_limit().is_some() { + return WorkflowEventRoute::fallback(); + } + + for traversal in &run.traversals { + 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); + } + } + + route.finish() + } + pub(crate) fn process_event<'a>( &mut self, config: &'a Config, @@ -412,7 +528,6 @@ impl Workflow { } let run_did_make_progress = run_result.did_make_progress(); - (run_result, run_did_make_progress) };