bd-workflows: implement type-based routing within the workflow engine

bd-log-matcher/src/matcher.rs

@@ -47,13 +47,77 @@ 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;
 
 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<Item = LogType> {
+    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<LogTypeSet> {
+    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,

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::<Vec<_>>()
+  );
+  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()
+  );
+}

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);

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 {