diff --git a/bd-api/src/api.rs b/bd-api/src/api.rs index 1608225be..e9ef0d38f 100644 --- a/bd-api/src/api.rs +++ b/bd-api/src/api.rs @@ -587,6 +587,8 @@ impl Api { StateScope::FEATURE_FLAG => Some(bd_state::Scope::FeatureFlagExposure), StateScope::GLOBAL_STATE => Some(bd_state::Scope::GlobalState), StateScope::SYSTEM => Some(bd_state::Scope::System), + // Log-field scopes are owned by the SDK metadata collector. Server-pushed state updates + // must not overwrite their virtual matcher/extractor view independently of live metadata. StateScope::CUSTOM_FIELDS | StateScope::OOTB_FIELDS | StateScope::UNSPECIFIED => None, } } diff --git a/bd-api/src/api_test.rs b/bd-api/src/api_test.rs index 2aee71556..533644b00 100644 --- a/bd-api/src/api_test.rs +++ b/bd-api/src/api_test.rs @@ -55,6 +55,7 @@ use bd_proto::protos::client::api::{ }; use bd_proto::protos::logging::payload::LogType; use bd_proto::protos::logging::payload::data::Data_type; +use bd_proto::protos::state::scope::StateScope; use bd_proto::protos::workflow::workflow::workflow::action::action_flush_buffers; use bd_runtime::runtime::{ConfigLoader, FeatureFlag}; use bd_state::StateReader; @@ -725,6 +726,12 @@ fn make_server_pushed_feature_flag_update(flag: &str, value: &str) -> ClientStat update } +#[test] +fn server_state_updates_exclude_sdk_owned_log_field_scopes() { + assert_eq!(Api::map_state_scope(StateScope::CUSTOM_FIELDS.into()), None); + assert_eq!(Api::map_state_scope(StateScope::OOTB_FIELDS.into()), None); +} + fn make_server_pushed_feature_flag_clear(flag: &str) -> ClientStateUpdate { let mut update = ClientStateUpdate::new(); update.set_clear_client_state(client_state_update::ClearClientState { diff --git a/bd-crash-handler/src/lib.rs b/bd-crash-handler/src/lib.rs index 7847bb510..320ad3e95 100644 --- a/bd-crash-handler/src/lib.rs +++ b/bd-crash-handler/src/lib.rs @@ -140,7 +140,7 @@ pub trait CrashReportHook: Send + Sync { #[derive(Clone)] pub struct Monitor { report_directory: PathBuf, - previous_run_state: bd_versioned_kv::ScopedMaps, + previous_run_state: Arc, state: bd_state::Store, artifact_client: Arc, @@ -158,7 +158,7 @@ impl Monitor { session: Arc, init_lifecycle: &InitLifecycleState, state: bd_state::Store, - previous_run_state: bd_versioned_kv::ScopedMaps, + previous_run_state: Arc, emit_log: impl Fn(CrashLog) -> anyhow::Result<()> + Send + Sync + 'static, crash_report_hook: Option>, ) -> Self { diff --git a/bd-crash-handler/src/monitor_test.rs b/bd-crash-handler/src/monitor_test.rs index 9c01307b3..6aaac3135 100644 --- a/bd-crash-handler/src/monitor_test.rs +++ b/bd-crash-handler/src/monitor_test.rs @@ -322,7 +322,7 @@ impl Setup { session, &InitLifecycleState::new(), (*state).clone(), - previous_run_state, + Arc::new(previous_run_state), emit_log, crash_report_hook, ); diff --git a/bd-log-filter/src/lib.rs b/bd-log-filter/src/lib.rs index 94c93d0fb..9b2010ea7 100644 --- a/bd-log-filter/src/lib.rs +++ b/bd-log-filter/src/lib.rs @@ -88,6 +88,17 @@ impl FilterChain { (Self { filters }, failures_count) } + /// Applies matching filters and their inline-log transforms. + /// + /// Matchers may read state-backed custom and OOTB fields through `state`, but transforms never + /// materialize, redact, replace, or remove those values. Later state-aware consumers resolve + /// the same state again, so transforms apply only to fields concretely present on `log`. + /// + /// TODO(snowp): Before eliding these fields from logs, preserve filter semantics with a per-log + /// state overlay. The overlay must carry rewritten and removed state values through every + /// downstream consumer, including buffer/workflow matching and tailing, without mutating the + /// global state. Snapshot uploads need the corresponding resolved values, not the current + /// state, because a snapshot can outlive the filter configuration that transformed its log. pub fn process(&self, log: &mut Log, state: &dyn bd_state::StateReader) { for filter in &self.filters { let fields_ref = FieldsRef::new(&log.fields, &log.matching_fields); diff --git a/bd-log-matcher/src/matcher.rs b/bd-log-matcher/src/matcher.rs index fabd76dee..f13a6263d 100644 --- a/bd-log-matcher/src/matcher.rs +++ b/bd-log-matcher/src/matcher.rs @@ -37,7 +37,7 @@ use base_log_matcher::tag_match::Value_match::{ StringValueMatch, }; use bd_log_primitives::tiny_set::TinyMap; -use bd_log_primitives::{DataValue, FieldsRef, LogLevel, LogMessage}; +use bd_log_primitives::{DataValue, FieldsRef, LogLevel, LogMessage, data_to_string_value}; use bd_proto::protos::config::v1::config::log_matcher::base_log_matcher::StringMatchType; use bd_proto::protos::config::v1::config::log_matcher::{ BaseLogMatcher as LegacyBaseLogMatcher, @@ -48,14 +48,15 @@ use bd_proto::protos::config::v1::config::{ log_matcher as legacy_log_matcher, }; use bd_proto::protos::log_matcher::log_matcher; -use bd_proto::protos::logging::payload::LogType; +use bd_proto::protos::logging::payload::data::Data_type; +use bd_proto::protos::logging::payload::{Data, LogType}; use bd_proto::protos::state::scope::StateScope; use bd_proto::protos::value_matcher::value_matcher::Operator; use bd_proto::protos::value_matcher::value_matcher::json_path_value_match::{ KeyOrIndex, key_or_index, }; -use bd_state::{Scope, state_value_as_cow}; +use bd_state::{Scope, Value_type, state_value_as_cow}; use log_matcher::LogMatcher; use log_matcher::log_matcher::{BaseLogMatcher, Matcher, base_log_matcher}; use rand::RngExt; @@ -289,21 +290,23 @@ impl Tree { Leaf::VersionValue(input, criteria) => input .get(message, fields, state) .is_some_and(|input| criteria.evaluate(input.as_ref())), - Leaf::IsSetValue(input) => input.get(message, fields, state).is_some(), + Leaf::IsSetValue(input) => input.is_set(message, fields, state), Leaf::JsonPathValue { field_key, path, matcher, } => { - let Some(value) = fields.field(field_key) else { + let Some(value) = resolved_field_value_with_state(fields, state, field_key) else { return MatchResult::NotMatched; }; // TODO: Fold Disabled into the planned general matcher evaluation context/cache work. - if !context.json_path_string_matching_enabled && value.as_str().is_some() { + if !context.json_path_string_matching_enabled && value.is_json_string() { return MatchResult::Disabled; } - resolve_json_path(value, path) - .is_some_and(|input| matcher.evaluate(input.as_ref(), extracted_fields)) + let Some(input) = value.resolve_json_path(path) else { + return MatchResult::NotMatched; + }; + matcher.evaluate(input.as_ref(), extracted_fields) }, Leaf::Sampled(sample_rate) => sample_matches_with_roll(*sample_rate, sampled_roll), Leaf::Any => true, @@ -396,7 +399,219 @@ pub enum InputType { State(Scope, String), } +/// A log-field value resolved from either the current log or its persistent state overlay. +/// +/// The state representation is deliberately borrowed as protobuf `Data`; converting it to a +/// `DataValue` here would recursively allocate for maps and arrays on every matcher evaluation. +#[derive(Clone, Copy)] +enum ResolvedFieldValue<'a> { + Log(&'a DataValue), + State(&'a bd_state::Value), +} + +impl<'a> ResolvedFieldValue<'a> { + fn as_cow(self) -> Option> { + match self { + Self::Log(value) => value.to_string_value(), + Self::State(value) => state_value_as_cow(value), + } + } + + fn as_i32(self) -> Option { + match self { + Self::Log(value) => log_field_as_i32(value), + Self::State(value) => state_value_as_i32(value), + } + } + + fn as_f64(self) -> Option { + match self { + Self::Log(value) => log_field_as_f64(value), + Self::State(value) => state_value_as_f64(value), + } + } + + fn resolve_json_path(self, path: &[JsonPathToken]) -> Option> { + match self { + Self::Log(value) => resolve_json_path(value, path), + Self::State(value) => resolve_json_path_from_state(value, path), + } + } + + fn is_json_string(self) -> bool { + match self { + Self::Log(value) => value.as_str().is_some(), + Self::State(value) => matches!( + value.value_type.as_ref(), + Some(Value_type::Data(data)) + if matches!(data.data_type.as_ref(), Some(Data_type::StringData(_))) + ), + } + } +} + +/// Views a logging `Data` value from a state entry with `DataValue` string semantics. +/// +/// Virtual fields are only written as `Value_type::Data`, and must preserve the behavior of the +/// equivalent inline `DataValue`. In particular, booleans are not string-matchable log fields. +fn persisted_log_field_as_cow(value: &bd_state::Value) -> Option> { + let Value_type::Data(data) = value.value_type.as_ref()? else { + return None; + }; + + data_to_string_value(data) +} + +/// Resolves a field using the metadata collector's persistent-field precedence. +/// +/// OOTB SDK state fields have the highest priority. Concrete log fields retain their existing +/// provider and per-log precedence and take priority over custom SDK state fields, which are the +/// lowest virtual layer. This lets callers read state-backed fields exactly like regular fields +/// without materializing them in the log's captured field map. +fn resolved_field_value_with_state<'a>( + fields: FieldsRef<'a>, + state: &'a dyn bd_state::StateReader, + field_key: &str, +) -> Option> { + state + .get(Scope::OotbFields, field_key) + .map(ResolvedFieldValue::State) + .or_else(|| fields.field(field_key).map(ResolvedFieldValue::Log)) + .or_else(|| { + state + .get(Scope::CustomFields, field_key) + .map(ResolvedFieldValue::State) + }) +} + +/// Resolves a field using the metadata collector's persistent-field precedence. +/// +/// OOTB SDK state fields have the highest priority. Concrete log fields retain their existing +/// provider and per-log precedence and take priority over custom SDK state fields, which are the +/// lowest virtual layer. This lets callers read state-backed fields exactly like regular fields +/// without materializing them in the log's captured field map. +#[must_use] +pub fn field_value_with_state<'a>( + fields: FieldsRef<'a>, + state: &'a dyn bd_state::StateReader, + field_key: &str, +) -> Option> { + // An OOTB state entry is authoritative even when it cannot be represented as a string. After + // that, retain FieldsRef::field_value's captured-to-matching-only fallback before considering + // the lowest-priority custom state layer. + if let Some(value) = state.get(Scope::OotbFields, field_key) { + return persisted_log_field_as_cow(value); + } + + fields.field_value(field_key).or_else(|| { + state + .get(Scope::CustomFields, field_key) + .and_then(persisted_log_field_as_cow) + }) +} + +/// Views an integer-compatible log-field value persisted in state without cloning its protobuf. +#[allow(clippy::cast_possible_truncation)] +fn persisted_log_field_as_i32(data: &Data) -> Option { + match data.data_type.as_ref()? { + Data_type::IntData(value) => i32::try_from(*value).ok(), + Data_type::SintData(value) => i32::try_from(*value).ok(), + Data_type::DoubleData(value) if !value.is_nan() => Some(*value as i32), + Data_type::StringData(value) => value.parse::().ok().map(|value| value as i32), + Data_type::BinaryData(_) + | Data_type::BoolData(_) + | Data_type::MapData(_) + | Data_type::ArrayData(_) + | Data_type::DoubleData(_) => None, + } +} + +/// Views a double-compatible log-field value persisted in state without cloning its protobuf. +#[allow(clippy::cast_precision_loss)] +fn persisted_log_field_as_f64(data: &Data) -> Option { + match data.data_type.as_ref()? { + Data_type::DoubleData(value) if !value.is_nan() => Some(*value), + Data_type::SintData(value) => Some(*value as f64), + Data_type::IntData(value) => Some(*value as f64), + Data_type::StringData(value) => value.parse().ok(), + Data_type::BinaryData(_) + | Data_type::BoolData(_) + | Data_type::MapData(_) + | Data_type::ArrayData(_) + | Data_type::DoubleData(_) => None, + } +} + +fn state_value_as_i32(value: &bd_state::Value) -> Option { + use bd_state::Value_type; + + match value.value_type.as_ref() { + Some(Value_type::IntValue(value)) => i32::try_from(*value).ok(), + #[allow(clippy::cast_possible_truncation)] + Some(Value_type::DoubleValue(value)) => Some(*value as i32), + Some(Value_type::StringValue(value)) => value.parse().ok(), + Some(Value_type::Data(value)) => persisted_log_field_as_i32(value), + Some(Value_type::BoolValue(_)) | None => None, + } +} + +fn state_value_as_f64(value: &bd_state::Value) -> Option { + use bd_state::Value_type; + + match value.value_type.as_ref() { + Some(Value_type::DoubleValue(value)) => Some(*value), + #[allow(clippy::cast_precision_loss)] + Some(Value_type::IntValue(value)) => Some(*value as f64), + Some(Value_type::StringValue(value)) => value.parse().ok(), + Some(Value_type::Data(value)) => persisted_log_field_as_f64(value), + Some(Value_type::BoolValue(_)) | None => None, + } +} + +#[allow(clippy::cast_possible_truncation)] +fn log_field_as_i32(field: &DataValue) -> Option { + match field { + DataValue::I64(value) => i32::try_from(*value).ok(), + DataValue::U64(value) => i32::try_from(*value).ok(), + DataValue::Double(value) => Some(**value as i32), + DataValue::String(_) | DataValue::SharedString(_) | DataValue::StaticString(_) => { + // Parse as f64 first then truncate to preserve backward compatibility with strings like + // "13.0" that were previously accepted. + Some(field.as_str()?.parse::().ok()? as i32) + }, + DataValue::Bytes(_) | DataValue::Boolean(_) | DataValue::Map(_) | DataValue::Array(_) => None, + } +} + +#[allow(clippy::cast_precision_loss)] +fn log_field_as_f64(field: &DataValue) -> Option { + match field { + DataValue::Double(value) => Some(**value), + DataValue::I64(value) => Some(*value as f64), + DataValue::U64(value) => Some(*value as f64), + DataValue::String(_) | DataValue::SharedString(_) | DataValue::StaticString(_) => { + field.as_str()?.parse().ok() + }, + DataValue::Bytes(_) | DataValue::Boolean(_) | DataValue::Map(_) | DataValue::Array(_) => None, + } +} + impl InputType { + fn is_set( + &self, + message: &LogMessage, + fields: FieldsRef<'_>, + state: &dyn bd_state::StateReader, + ) -> bool { + match self { + Self::Message => message.as_str().is_some(), + // Preserve log-field string semantics: binary values are not a matchable field value. + Self::Field(field_key) => field_value_with_state(fields, state, field_key).is_some(), + // State presence is independent of whether a value can be represented as a matcher string. + Self::State(scope, key) => state.get(*scope, key).is_some(), + } + } + fn get<'a>( &self, message: &'a LogMessage, @@ -405,12 +620,12 @@ impl InputType { ) -> Option> { match self { Self::Message => message.as_str().map(Cow::Borrowed), - Self::Field(field_key) => fields.field_value(field_key), + Self::Field(field_key) => field_value_with_state(fields, state, field_key), Self::State(scope, flag_key) => state.get(*scope, flag_key).and_then(|value| { if value.value_type.is_none() { Some(Cow::Borrowed("")) } else { - state_value_as_cow(value) + ResolvedFieldValue::State(value).as_cow() } }), } @@ -427,35 +642,11 @@ impl InputType { ) -> Option { match self { Self::Message => message.as_str().and_then(|s| s.parse().ok()), - Self::Field(field_key) => { - let field = fields.field(field_key)?; - match field { - DataValue::I64(v) => i32::try_from(*v).ok(), - DataValue::U64(v) => i32::try_from(*v).ok(), - DataValue::Double(v) => Some(**v as i32), - DataValue::String(_) | DataValue::SharedString(_) | DataValue::StaticString(_) => { - // Parse as f64 first then truncate to preserve backward compatibility with strings - // like "13.0" that were previously accepted. - Some(field.as_str()?.parse::().ok()? as i32) - }, - DataValue::Bytes(_) | DataValue::Boolean(_) | DataValue::Map(_) | DataValue::Array(_) => { - None - }, - } - }, - Self::State(scope, flag_key) => { - use bd_state::Value_type; - let v = state.get(*scope, flag_key)?; - match v.value_type { - Some(Value_type::IntValue(i)) => i32::try_from(i).ok(), - Some(Value_type::DoubleValue(d)) => Some(d as i32), - Some(Value_type::StringValue(ref s)) => s.parse().ok(), - Some(Value_type::Data(_)) => { - state_value_as_cow(v).and_then(|value| value.parse::().ok().map(|v| v as i32)) - }, - Some(Value_type::BoolValue(_)) | None => None, - } - }, + Self::Field(field_key) => resolved_field_value_with_state(fields, state, field_key) + .and_then(ResolvedFieldValue::as_i32), + Self::State(scope, flag_key) => state + .get(*scope, flag_key) + .and_then(|value| ResolvedFieldValue::State(value).as_i32()), } } @@ -470,31 +661,11 @@ impl InputType { ) -> Option { match self { Self::Message => message.as_str().and_then(|s| s.parse().ok()), - Self::Field(field_key) => { - let field = fields.field(field_key)?; - match field { - DataValue::Double(v) => Some(**v), - DataValue::I64(v) => Some(*v as f64), - DataValue::U64(v) => Some(*v as f64), - DataValue::String(_) | DataValue::SharedString(_) | DataValue::StaticString(_) => { - field.as_str()?.parse().ok() - }, - DataValue::Bytes(_) | DataValue::Boolean(_) | DataValue::Map(_) | DataValue::Array(_) => { - None - }, - } - }, - Self::State(scope, flag_key) => { - use bd_state::Value_type; - let v = state.get(*scope, flag_key)?; - match v.value_type { - Some(Value_type::DoubleValue(d)) => Some(d), - Some(Value_type::IntValue(i)) => Some(i as f64), - Some(Value_type::StringValue(ref s)) => s.parse().ok(), - Some(Value_type::Data(_)) => state_value_as_cow(v).and_then(|value| value.parse().ok()), - Some(Value_type::BoolValue(_)) | None => None, - } - }, + Self::Field(field_key) => resolved_field_value_with_state(fields, state, field_key) + .and_then(ResolvedFieldValue::as_f64), + Self::State(scope, flag_key) => state + .get(*scope, flag_key) + .and_then(|value| ResolvedFieldValue::State(value).as_f64()), } } } @@ -646,8 +817,9 @@ impl Leaf { StateScope::FEATURE_FLAG => Scope::FeatureFlagExposure, StateScope::GLOBAL_STATE => Scope::GlobalState, StateScope::SYSTEM => Scope::System, + // Custom and SDK-owned fields are virtual log fields. They are intentionally not + // general state-matcher inputs until their state-change semantics are introduced. StateScope::CUSTOM_FIELDS | StateScope::OOTB_FIELDS | StateScope::UNSPECIFIED => { - // Custom and SDK-owned fields are not stored in the bd_state namespaces. return Err(anyhow!("Unsupported state scope")); }, }; @@ -815,3 +987,49 @@ fn resolve_structured_json_path<'a>( | DataValue::Array(_) => None, } } + +fn resolve_json_path_from_state<'a>( + value: &'a bd_state::Value, + path: &[JsonPathToken], +) -> Option> { + let Value_type::Data(value) = value.value_type.as_ref()? else { + return None; + }; + + // JSON-string fields retain their existing parsing behavior after moving into state. + if let Some(Data_type::StringData(value)) = value.data_type.as_ref() { + return resolve_json_string_path(value, path); + } + + let mut current = value; + for token in path { + match token { + JsonPathToken::Key(key) => { + let Data_type::MapData(map_data) = current.data_type.as_ref()? else { + return None; + }; + current = map_data.entries.get(key)?; + }, + JsonPathToken::Index(index) => { + let Data_type::ArrayData(array_data) = current.data_type.as_ref()? else { + return None; + }; + let len = i32::try_from(array_data.items.len()).ok()?; + let index = if *index < 0 { len + *index } else { *index }; + let index: usize = index.try_into().ok()?; + current = array_data.items.get(index)?; + }, + } + } + + match current.data_type.as_ref()? { + Data_type::StringData(value) => Some(Cow::Borrowed(value)), + Data_type::BinaryData(_) + | Data_type::BoolData(_) + | Data_type::IntData(_) + | Data_type::SintData(_) + | Data_type::DoubleData(_) + | Data_type::MapData(_) + | Data_type::ArrayData(_) => None, + } +} diff --git a/bd-log-matcher/src/matcher_test.rs b/bd-log-matcher/src/matcher_test.rs index 917b5dda0..e10e95f51 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::{MatchContext, RandomNumberGenerator, Tree}; +use crate::matcher::{MatchContext, RandomNumberGenerator, Tree, field_value_with_state}; use crate::test::TestMatcher; use ahash::AHashMap; use bd_log_primitives::tiny_set::TinyMap; @@ -22,11 +22,8 @@ use bd_log_primitives::{ log_level, }; use bd_proto::protos::log_matcher::log_matcher::{LogMatcher, log_matcher}; -use bd_proto::protos::logging::payload::data::Data_type; -use bd_proto::protos::logging::payload::{Data, LogType, MapData}; +use bd_proto::protos::logging::payload::LogType; use bd_proto::protos::state::matcher::state_value_match; -use bd_proto::protos::state::payload::StateValue; -use bd_proto::protos::state::payload::state_value::Value_type; use bd_proto::protos::state::scope::StateScope; use bd_proto::protos::value_matcher::value_matcher::double_value_match::Double_value_match_type; use bd_proto::protos::value_matcher::value_matcher::int_value_match::Int_value_match_type; @@ -1470,61 +1467,10 @@ fn state_match_double_values() { } } -#[test] -fn state_match_data_values() { - let mut state = bd_state::InMemoryStateReader::default(); - for (key, data_type) in [ - ("string", Data_type::StringData("value".to_string())), - ("unsigned", Data_type::IntData(42)), - ("signed", Data_type::SintData(-42)), - ("double", Data_type::DoubleData(98.6)), - ("boolean", Data_type::BoolData(true)), - ("map", Data_type::MapData(MapData::default())), - ] { - state.insert( - bd_state::Scope::FeatureFlagExposure, - key, - StateValue { - value_type: Some(Value_type::Data(Data { - data_type: Some(data_type), - ..Default::default() - })), - ..Default::default() - }, - ); - } - - for (matcher, matches) in [ - ( - make_string_feature_flag_matcher("string", Operator::OPERATOR_EQUALS, "value"), - true, - ), - ( - make_int_state_matcher("unsigned", Operator::OPERATOR_EQUALS, 42), - true, - ), - ( - make_int_state_matcher("signed", Operator::OPERATOR_EQUALS, -42), - true, - ), - ( - make_double_state_matcher("double", Operator::OPERATOR_EQUALS, 98.6), - true, - ), - ( - make_string_feature_flag_matcher("boolean", Operator::OPERATOR_EQUALS, "true"), - true, - ), - ( - make_string_feature_flag_matcher("map", Operator::OPERATOR_EQUALS, ""), - false, - ), - ] { - let matcher = TestMatcher::new(&matcher).unwrap(); - assert_eq!( - matches, - matcher.match_log_with_state(TypedLogLevel::Debug, LogType::NORMAL, "foo", [], &state), - ); +fn persisted_log_field_state_value(value: DataValue) -> bd_state::Value { + bd_state::Value { + value_type: bd_state::Value_type::Data(value.into_proto()).into(), + ..Default::default() } } @@ -1549,6 +1495,212 @@ fn custom_field_state_scopes_are_unsupported() { } } +#[test] +fn virtual_state_fields_follow_log_field_precedence() { + let mut state = bd_state::InMemoryStateReader::default(); + state.insert( + bd_state::Scope::CustomFields, + "custom_only", + persisted_log_field_state_value(DataValue::String("custom".to_string())), + ); + state.insert( + bd_state::Scope::CustomFields, + "overridden_custom", + persisted_log_field_state_value(DataValue::String("custom".to_string())), + ); + state.insert( + bd_state::Scope::OotbFields, + "overridden_ootb", + persisted_log_field_state_value(DataValue::String("ootb".to_string())), + ); + state.insert( + bd_state::Scope::OotbFields, + "typed_ootb", + persisted_log_field_state_value(DataValue::Double( + NotNan::new(42.5).expect("test value must not be NaN"), + )), + ); + + let fields = [ + ("overridden_custom", "log"), + ("overridden_ootb", "log"), + ("typed_ootb", "13.0"), + ]; + + for matcher in [ + builder::field_equals("custom_only", "custom"), + builder::field_equals("overridden_custom", "log"), + builder::field_equals("overridden_ootb", "ootb"), + builder::field_double_equals("typed_ootb", 42.5), + ] { + assert!( + TestMatcher::new(&matcher) + .expect("field matcher should be valid") + .match_log_with_state(TypedLogLevel::Debug, LogType::NORMAL, "foo", fields, &state) + ); + } +} + +#[test] +fn virtual_state_fields_preserve_matching_field_fallback() { + let captured_fields: LogFields = [("shared".into(), DataValue::Bytes(vec![0].into()))].into(); + let matching_fields: LogFields = [( + "shared".into(), + DataValue::String("matching-only".to_string()), + )] + .into(); + let mut state = bd_state::InMemoryStateReader::default(); + + assert_eq!( + field_value_with_state( + FieldsRef::new(&captured_fields, &matching_fields), + &state, + "shared", + ) + .as_deref(), + Some("matching-only") + ); + + state.insert( + bd_state::Scope::OotbFields, + "shared", + persisted_log_field_state_value(DataValue::Bytes(vec![1].into())), + ); + assert_eq!( + field_value_with_state( + FieldsRef::new(&captured_fields, &matching_fields), + &state, + "shared", + ) + .as_deref(), + None + ); +} + +#[test] +fn virtual_state_fields_preserve_inline_boolean_string_semantics() { + let mut state = bd_state::InMemoryStateReader::default(); + state.insert( + bd_state::Scope::OotbFields, + "enabled", + persisted_log_field_state_value(DataValue::Boolean(true)), + ); + + assert!( + field_value_with_state( + FieldsRef::new(&LogFields::default(), &LogFields::default()), + &state, + "enabled", + ) + .is_none() + ); +} + +#[test] +fn virtual_state_fields_support_json_path_matching() { + let matcher = simple_log_matcher(TagMatch(base_log_matcher::TagMatch { + tag_key: "payload".to_string(), + value_match: Some( + log_matcher::base_log_matcher::tag_match::Value_match::JsonValueMatch(JsonPathValueMatch { + operator: Operator::OPERATOR_EQUALS.into(), + match_value: "state-value".to_string(), + key_or_index: vec![KeyOrIndex { + key_or_index: Some(key_or_index::Key_or_index::Key("key".to_string())), + ..Default::default() + }], + ..Default::default() + }), + ), + ..Default::default() + })); + + let mut state = bd_state::InMemoryStateReader::default(); + state.insert( + bd_state::Scope::CustomFields, + "payload", + persisted_log_field_state_value(DataValue::from(AHashMap::from_iter([( + "key".to_string(), + DataValue::String("state-value".to_string()), + )]))), + ); + + assert!( + TestMatcher::new(&matcher) + .expect("JSON matcher should be valid") + .match_log_with_state(TypedLogLevel::Debug, LogType::NORMAL, "foo", [], &state) + ); +} + +#[test] +fn virtual_state_json_strings_match_and_honor_the_runtime_gate() { + let matcher = simple_log_matcher(TagMatch(base_log_matcher::TagMatch { + tag_key: "payload".to_string(), + value_match: Some( + log_matcher::base_log_matcher::tag_match::Value_match::JsonValueMatch(JsonPathValueMatch { + operator: Operator::OPERATOR_EQUALS.into(), + match_value: "state-value".to_string(), + key_or_index: vec![KeyOrIndex { + key_or_index: Some(key_or_index::Key_or_index::Key("key".to_string())), + ..Default::default() + }], + ..Default::default() + }), + ), + ..Default::default() + })); + let tree = Tree::new(&matcher).expect("JSON matcher should be valid"); + let mut state = bd_state::InMemoryStateReader::default(); + state.insert( + bd_state::Scope::CustomFields, + "payload", + persisted_log_field_state_value(DataValue::String(r#"{"key":"state-value"}"#.to_string())), + ); + let fields: LogFields = [].into(); + let message = LogMessage::String("foo".to_string()); + + assert!(tree.do_match( + log_level::DEBUG, + LogType::NORMAL, + &message, + FieldsRef::new(&fields, &EMPTY_FIELDS), + &state, + &TinyMap::default(), + 0, + MatchContext::default(), + )); + assert!(!tree.do_match( + log_level::DEBUG, + LogType::NORMAL, + &message, + FieldsRef::new(&fields, &EMPTY_FIELDS), + &state, + &TinyMap::default(), + 0, + MatchContext { + json_path_string_matching_enabled: false, + }, + )); + + let negated_tree = Tree::new(&builder::not(matcher)).expect("JSON matcher should be valid"); + state.insert( + bd_state::Scope::CustomFields, + "payload", + persisted_log_field_state_value(DataValue::String(r#"{"other":"value"}"#.to_string())), + ); + assert!(!negated_tree.do_match( + log_level::DEBUG, + LogType::NORMAL, + &message, + FieldsRef::new(&fields, &EMPTY_FIELDS), + &state, + &TinyMap::default(), + 0, + MatchContext { + json_path_string_matching_enabled: false, + }, + )); +} + #[test] fn state_match_is_set() { struct Input { @@ -1597,6 +1749,24 @@ fn state_match_is_set() { idx, input.matches, actual ); } + + let mut untyped_state = bd_state::InMemoryStateReader::default(); + untyped_state.insert( + bd_state::Scope::FeatureFlagExposure, + "untyped", + bd_state::Value::default(), + ); + assert!( + TestMatcher::new(&make_state_is_set_matcher("untyped")) + .unwrap() + .match_log_with_state( + TypedLogLevel::Debug, + LogType::NORMAL, + "foo", + [], + &untyped_state, + ) + ); } fn simple_log_matcher(match_type: base_log_matcher::Match_type) -> LogMatcher { diff --git a/bd-log-primitives/src/lib.rs b/bd-log-primitives/src/lib.rs index 20639c08a..1453a9af1 100644 --- a/bd-log-primitives/src/lib.rs +++ b/bd-log-primitives/src/lib.rs @@ -317,6 +317,25 @@ pub enum DataValue { Array(LogArrayData), } +/// Views a protobuf log value as the string representation supported by string matchers. +/// +/// This is the borrowed counterpart to `DataValue::to_string_value`. It avoids constructing a +/// `DataValue` when callers already hold a decoded protobuf `Data`. +#[must_use] +pub fn data_to_string_value(data: &Data) -> Option> { + match data.data_type.as_ref()? { + Data_type::StringData(value) => Some(Cow::Borrowed(value)), + Data_type::IntData(value) => Some(Cow::Owned(value.to_string())), + Data_type::SintData(value) => Some(Cow::Owned(value.to_string())), + Data_type::DoubleData(value) if !value.is_nan() => Some(Cow::Owned(value.to_string())), + Data_type::BinaryData(_) + | Data_type::BoolData(_) + | Data_type::MapData(_) + | Data_type::ArrayData(_) + | Data_type::DoubleData(_) => None, + } +} + impl DataValue { /// Creates a new `DataValue` instance from a static string slice. This is slightly more /// efficient than using `SharedString` as it avoids heap allocation. diff --git a/bd-log-primitives/src/lib_test.rs b/bd-log-primitives/src/lib_test.rs index 464f170c9..0a05f633d 100644 --- a/bd-log-primitives/src/lib_test.rs +++ b/bd-log-primitives/src/lib_test.rs @@ -7,7 +7,16 @@ #![allow(clippy::cast_possible_truncation, clippy::unwrap_used)] -use crate::{DataValue, EncodableLog, Log, LogFieldValue, LogType, TypedLogLevel, log_level}; +use crate::{ + DataValue, + EncodableLog, + Log, + LogFieldValue, + LogType, + TypedLogLevel, + data_to_string_value, + log_level, +}; use ahash::AHashMap; use bd_proto::protos::logging::payload::data::Data_type; use bd_proto::protos::logging::payload::log::Field; @@ -343,6 +352,29 @@ fn to_string_value_converts_numeric_types() { ); } +#[test] +fn data_to_string_value_matches_data_value_semantics_without_conversion() { + for value in [ + DataValue::String("hello".to_string()), + DataValue::I64(-42), + DataValue::U64(42), + DataValue::Double(NotNan::new(1.5).unwrap()), + DataValue::Boolean(true), + DataValue::Bytes(vec![1, 2, 3].into()), + ] { + let expected = value.to_string_value().map(Cow::into_owned); + let proto = value.into_proto(); + + assert_eq!(data_to_string_value(&proto).map(Cow::into_owned), expected); + } + + let nan = Data { + data_type: Data_type::DoubleData(f64::NAN).into(), + ..Default::default() + }; + assert!(data_to_string_value(&nan).is_none()); +} + #[test] fn field_value_returns_numeric_types_as_strings() { use crate::FieldsRef; diff --git a/bd-logger/src/async_log_buffer.rs b/bd-logger/src/async_log_buffer.rs index 7cc6e917c..ec98d4f0a 100644 --- a/bd-logger/src/async_log_buffer.rs +++ b/bd-logger/src/async_log_buffer.rs @@ -18,7 +18,7 @@ use crate::logger::{ with_thread_local_logger_guard, }; use crate::logging_state::{ConfigUpdate, LoggingState, UninitializedLoggingContext}; -use crate::metadata::MetadataCollector; +use crate::metadata::{MetadataCollector, verify_custom_field_name}; use crate::network::{NetworkQualityInterceptor, SystemTimeProvider}; use crate::{Block, battery, internal_report, network}; use anyhow::anyhow; @@ -48,6 +48,7 @@ use bd_log_metadata::MetadataProvider; use bd_log_primitives::{ AnnotatedLogField, AnnotatedLogFields, + DataValue, Log, LogFieldValue, LogFields, @@ -71,6 +72,9 @@ use bd_state::{ MEMORY_PRESSURE_LEVEL_KEY, SYSTEM_SESSION_ID_KEY, Scope, + StateReader, + Value, + Value_type, string_value, }; use bd_stats_common::{Counter as _, Histogram as _, labels}; @@ -78,7 +82,7 @@ use bd_time::{OffsetDateTimeExt, TimeDurationExt, TimeProvider}; use bd_workflow_stats::workflow::{WorkflowDebugStateKey, WorkflowDebugTransitionType}; use bd_workflows::workflow::WorkflowDebugStateMap; use debug_data_request::workflow_transition_debug_data::Transition_type; -use std::collections::{HashMap, VecDeque}; +use std::collections::{BTreeMap, HashMap, VecDeque}; use std::future::{Future, ready}; use std::pin::Pin; use std::sync::Arc; @@ -161,6 +165,53 @@ impl AdmissionCounters { } } +/// Stores a log field without changing its type in the persistent state journal. +fn persistent_field_value(value: DataValue) -> Value { + Value { + value_type: Value_type::Data(value.into_proto()).into(), + ..Default::default() + } +} + +/// Returns the minimal state mutations that reconcile startup fields with persisted field state. +/// +/// Fields historically lived only for one SDK process. Reconcile rather than clear-and-reseed so +/// unchanged values retain their original timestamps while fields absent at startup receive a +/// tombstone that bounds their prior value's lifetime. +fn initial_field_state_updates( + initial_ootb_fields: LogFields, + initial_custom_fields: LogFields, + state: &dyn StateReader, +) -> Vec<(Scope, String, Option)> { + let mut updates = Vec::new(); + for (scope, fields) in [ + (Scope::OotbFields, initial_ootb_fields), + (Scope::CustomFields, initial_custom_fields), + ] { + let desired_fields: BTreeMap<_, _> = fields + .into_iter() + .map(|(key, value)| (key.to_string(), persistent_field_value(value))) + .collect(); + + updates.extend( + desired_fields + .iter() + .filter(|(key, value)| state.get(scope, key) != Some(*value)) + .map(|(key, value)| (scope, key.clone(), Some(value.clone()))), + ); + updates.extend( + state + .as_scoped_maps() + .iter_scope(scope) + .map(|(key, _)| key) + .filter(|key| !desired_fields.contains_key(key.as_str())) + .map(|key| (scope, key.clone(), None)), + ); + } + + updates +} + #[derive(Clone)] pub struct Sender { inner: SenderInner, @@ -814,6 +865,7 @@ impl AsyncLogBuffer { &mut self, log: LogLine, state_store: &bd_state::Store, + previous_run_state: &bd_versioned_kv::ScopedMaps, context: Option, ) -> anyhow::Result<()> { let mut logs = VecDeque::new(); @@ -821,7 +873,9 @@ impl AsyncLogBuffer { while let Some((log, context)) = logs.pop_front() { let source_context = context.clone(); let source_attributes_overrides = log.attributes_overrides.clone(); - let log_replay_result = self.process_log(log, state_store, context).await?; + let log_replay_result = self + .process_log(log, state_store, previous_run_state, context) + .await?; logs.extend(log_replay_result.logs_to_inject.into_iter().map(|log| { workflow_generated_log( log, @@ -852,9 +906,15 @@ impl AsyncLogBuffer { &mut self, log: LogLine, state_store: &bd_state::Store, + previous_run_state: &bd_versioned_kv::ScopedMaps, context: Option, ) -> anyhow::Result { // Prevent re-entrancy when we are evaluating the log metadata. + let previous_process = matches!(&context, Some(EventContext::PreviousProcess { .. })) + || matches!( + &log.attributes_overrides, + Some(LogAttributesOverrides::PreviousRunSessionID(_)) + ); let result = with_thread_local_logger_guard(|| { match context { Some(EventContext::CurrentProcess(context)) => Ok(( @@ -878,12 +938,7 @@ impl AsyncLogBuffer { ), None, )), - None - if matches!( - &log.attributes_overrides, - Some(LogAttributesOverrides::PreviousRunSessionID(_)) - ) => - { + None if previous_process => { // Since we're mimicing a log from the previous app start we want to use the previous // global state instead of calling into the providers at this point. Ok(( @@ -971,7 +1026,14 @@ impl AsyncLogBuffer { capture_session: log.capture_session, }; - self.write_log(processed_log, state_store).await + if previous_process { + // A previous-process log is evaluated against the complete state snapshot captured at + // startup, including feature flags and system state from that process. + self.write_log(processed_log, previous_run_state).await + } else { + let state = state_store.read().await; + self.write_log(processed_log, &state).await + } }, Err(e) => { // TODO(Augustyniak): Consider logging as error so that SDK customers can see these @@ -985,7 +1047,7 @@ impl AsyncLogBuffer { async fn write_log( &mut self, log: Log, - state_store: &bd_state::Store, + state: &dyn StateReader, ) -> anyhow::Result { let log_replay_result = match &mut self.logging_state { LoggingState::Uninitialized(_) => { @@ -996,7 +1058,7 @@ impl AsyncLogBuffer { .replay_log( log, &mut initialized_logging_context.processing_pipeline, - state_store, + state, self.time_provider.now(), ) .await @@ -1029,6 +1091,41 @@ impl AsyncLogBuffer { } } + async fn persist_initial_log_fields( + &mut self, + initial_ootb_fields: LogFields, + initial_custom_fields: LogFields, + state_store: &bd_state::Store, + ) { + let updates = { + let state = state_store.read().await; + initial_field_state_updates(initial_ootb_fields, initial_custom_fields, &state) + }; + + for (scope, key, value) in updates { + match value { + Some(value) => { + if let Err(e) = state_store.insert(scope, key.clone(), value).await { + log::warn!("state rejected initial {scope:?} log field {key:?}; dropping it: {e}"); + Self::clear_virtual_log_field(state_store, scope, &key).await; + if scope == Scope::CustomFields { + self.metadata_collector.remove_field(key.into()); + } else { + self.metadata_collector.remove_ootb_field(key.into()); + } + } + }, + None => Self::clear_virtual_log_field(state_store, scope, &key).await, + } + } + } + + async fn clear_virtual_log_field(state_store: &bd_state::Store, scope: Scope, key: &str) { + if let Err(e) = state_store.remove(scope, key).await { + log::warn!("failed to clear {scope:?} log field {key:?}: {e}"); + } + } + async fn update(mut self, config: ConfigUpdate) -> Self { let initialized_logging_context = match self.logging_state { LoggingState::Uninitialized(uninitialized_logging_context) => { @@ -1049,27 +1146,28 @@ impl AsyncLogBuffer { self } - pub async fn run( + pub async fn run_with_previous_state( self, state_store: bd_state::Store, report_processor: impl ReportProcessor, + previous_run_state: Arc, ) -> Self { let shutdown_trigger = ComponentShutdownTrigger::default(); self - .run_with_shutdown( + .run_with_shutdown_and_previous_state( state_store, report_processor, + previous_run_state, shutdown_trigger.make_shutdown(), ) .await } - // TODO(mattklein123): This seems to only be used for tests. Figure out how to clean this up - // so we don't need this just for tests. - pub async fn run_with_shutdown( + async fn run_with_shutdown_and_previous_state( mut self, state_store: bd_state::Store, report_processor: impl ReportProcessor, + previous_run_state: Arc, mut shutdown: ComponentShutdown, ) -> Self { // EventBuffer protects ingress behind its startup gate while configuration is applied. Once @@ -1079,6 +1177,12 @@ impl AsyncLogBuffer { .event_buffer .start_startup_gate(self.startup_replay_delay.take()); + let (initial_ootb_fields, initial_custom_fields) = + self.metadata_collector.initial_persistent_fields(); + self + .persist_initial_log_fields(initial_ootb_fields, initial_custom_fields, &state_store) + .await; + let local_shutdown = shutdown.cancelled(); tokio::pin!(local_shutdown); let mut self_shutdown = self.shutdown_trigger_handle.make_shutdown(); @@ -1139,7 +1243,10 @@ impl AsyncLogBuffer { ); } - if let Err(e) = self.process_all_logs(log, &state_store, Some(context)).await { + if let Err(e) = self + .process_all_logs(log, &state_store, &previous_run_state, Some(context)) + .await + { log::debug!("failed to process all logs: {e}"); } }, @@ -1292,14 +1399,76 @@ impl AsyncLogBuffer { ) { match async_log_buffer_message { LoggerControl::AddLogField(key, value) => { + if self.metadata_collector.is_ootb_field(&key) + || state_store + .read() + .await + .get(Scope::OotbFields, &key) + .is_some() + { + log::debug!("ignoring custom log field {key:?} because an OOTB field owns it"); + return; + } + + if let Err(e) = verify_custom_field_name(&key) { + log::warn!("failed to add log field ({key:?}): {e}"); + return; + } + + if let Err(e) = state_store + .insert( + Scope::CustomFields, + key.clone(), + persistent_field_value(value.clone()), + ) + .await + { + log::warn!("state rejected custom log field ({key:?}); leaving it unchanged: {e}"); + return; + } if let Err(e) = self.metadata_collector.add_field(key.clone().into(), value) { log::warn!("failed to add log field ({key:?}): {e}"); } }, LoggerControl::UpdateOotbLogField(key, value) => { - self.metadata_collector.update_ootb_field(key.into(), value); + if let Err(e) = state_store + .insert( + Scope::OotbFields, + key.clone(), + persistent_field_value(value.clone()), + ) + .await + { + log::warn!("state rejected OOTB log field ({key:?}); leaving it unchanged: {e}"); + return; + } + + self + .metadata_collector + .update_ootb_field(key.clone().into(), value); + + if let Err(e) = state_store.remove(Scope::CustomFields, &key).await { + log::warn!("failed to remove shadowed custom log field ({key:?}): {e}"); + } }, LoggerControl::RemoveLogField(field_name) => { + if self.metadata_collector.is_ootb_field(&field_name) + || state_store + .read() + .await + .get(Scope::OotbFields, &field_name) + .is_some() + { + log::debug!( + "ignoring removal of custom log field {field_name:?} because an OOTB field owns it" + ); + return; + } + + if let Err(e) = state_store.remove(Scope::CustomFields, &field_name).await { + log::warn!("failed to remove custom log field ({field_name:?}): {e}"); + return; + } self.metadata_collector.remove_field(field_name.into()); }, LoggerControl::SetMemoryPressureLevel { level } => { diff --git a/bd-logger/src/async_log_buffer_test.rs b/bd-logger/src/async_log_buffer_test.rs index f9078f77c..08ddc3185 100644 --- a/bd-logger/src/async_log_buffer_test.rs +++ b/bd-logger/src/async_log_buffer_test.rs @@ -56,9 +56,16 @@ use bd_proto::protos::logging::payload::LogType; use bd_runtime::runtime::{ConfigLoader, FeatureFlag}; use bd_session::Strategy; use bd_session::test::no_timeout; -use bd_shutdown::ComponentShutdownTrigger; +use bd_shutdown::{ComponentShutdown, ComponentShutdownTrigger}; use bd_state::test::TestStore; -use bd_state::{MEMORY_PRESSURE_LEVEL_KEY, SYSTEM_SESSION_ID_KEY, Scope, StateReader}; +use bd_state::{ + InMemoryStateReader, + MEMORY_PRESSURE_LEVEL_KEY, + PersistentStoreConfig, + SYSTEM_SESSION_ID_KEY, + Scope, + StateReader, +}; use bd_stats_common::labels; use bd_test_helpers::events::NoOpListenerTarget; use bd_test_helpers::metadata_provider::LogMetadata; @@ -72,7 +79,7 @@ use bd_workflows::config::WorkflowsConfiguration; use bd_workflows::engine::ProcessLocalPendingFlushState; use bd_workflows::test::MakeConfig; use futures_util::poll; -use std::future; +use std::future::{self, Future}; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; use time::OffsetDateTime; @@ -80,6 +87,20 @@ use time::ext::{NumericalDuration, NumericalStdDuration}; use tokio::sync::{Notify, mpsc}; use tokio_test::assert_ok; +fn run_with_shutdown( + buffer: AsyncLogBuffer, + state_store: bd_state::Store, + report_processor: impl ReportProcessor, + shutdown: ComponentShutdown, +) -> impl Future> { + buffer.run_with_shutdown_and_previous_state( + state_store, + report_processor, + Arc::new(bd_versioned_kv::ScopedMaps::default()), + shutdown, + ) +} + // // StartupGateReady // @@ -176,6 +197,7 @@ struct Setup { replayer_log_notify: Arc, replayer_logs: Arc>>, replayer_fields: Arc>>, + replayer_feature_flags: Arc>>>, shutdown: Option, store: Arc, session_strategy: Arc, @@ -213,6 +235,7 @@ impl Setup { replayer_log_notify: Arc::new(Notify::new()), replayer_logs: Arc::default(), replayer_fields: Arc::default(), + replayer_feature_flags: Arc::default(), shutdown: Some(ComponentShutdownTrigger::default()), _data_upload_rx: data_upload_rx, data_upload_tx, @@ -254,6 +277,7 @@ impl Setup { self.replayer_log_notify = replayer.logs_notify.clone(); self.replayer_logs = replayer.logs.clone(); self.replayer_fields = replayer.fields.clone(); + self.replayer_feature_flags = replayer.feature_flags.clone(); let (_, report_rx) = tokio::sync::mpsc::channel(1); @@ -446,7 +470,8 @@ async fn startup_gate_holds_preconfiguration_logs_until_the_replay_timer() { let state_store = TestStore::new().await; let shutdown_trigger = ComponentShutdownTrigger::default(); - let handle = tokio::task::spawn(buffer.run_with_shutdown( + let handle = tokio::task::spawn(run_with_shutdown( + buffer, state_store.take_inner(), (), shutdown_trigger.make_shutdown(), @@ -480,7 +505,8 @@ async fn empty_startup_gate_opening_marks_log_processing_running() { ); let state_store = TestStore::new().await; let shutdown_trigger = ComponentShutdownTrigger::default(); - let handle = tokio::spawn(buffer.run_with_shutdown( + let handle = tokio::spawn(run_with_shutdown( + buffer, state_store.take_inner(), (), shutdown_trigger.make_shutdown(), @@ -522,7 +548,8 @@ async fn shutdown_before_configuration_does_not_open_the_startup_gate() { let event_buffer = buffer.event_buffer.clone(); let state_store = TestStore::new().await; let shutdown_trigger = ComponentShutdownTrigger::default(); - let handle = tokio::spawn(buffer.run_with_shutdown( + let handle = tokio::spawn(run_with_shutdown( + buffer, state_store.take_inner(), (), shutdown_trigger.make_shutdown(), @@ -567,7 +594,8 @@ async fn runtime_startup_replay_delay_extension_rearms_the_running_gate() { let state_store = TestStore::new().await; let shutdown_trigger = ComponentShutdownTrigger::default(); - let handle = tokio::task::spawn(buffer.run_with_shutdown( + let handle = tokio::task::spawn(run_with_shutdown( + buffer, state_store.take_inner(), (), shutdown_trigger.make_shutdown(), @@ -625,7 +653,8 @@ async fn report_processing_does_not_change_the_selected_startup_delay() { let processed = Arc::new(Notify::new()); let state_store = TestStore::new().await; let shutdown_trigger = ComponentShutdownTrigger::default(); - let handle = tokio::spawn(buffer.run_with_shutdown( + let handle = tokio::spawn(run_with_shutdown( + buffer, state_store.take_inner(), ReportProcessingSignal(processed.clone()), shutdown_trigger.make_shutdown(), @@ -681,7 +710,8 @@ async fn flush_during_report_discovery_preserves_previous_process_replay_priorit let resume = Arc::new(Notify::new()); let state_store = TestStore::new().await; let shutdown = ComponentShutdownTrigger::default(); - let handle = tokio::spawn(buffer.run_with_shutdown( + let handle = tokio::spawn(run_with_shutdown( + buffer, state_store.take_inner(), PausedReportProcessor { entered: entered.clone(), @@ -772,7 +802,8 @@ async fn startup_gate_ready_blocking_flush_releases_after_older_work() { let state_store = TestStore::new().await; let shutdown_trigger = ComponentShutdownTrigger::default(); - let handle = tokio::task::spawn(buffer.run_with_shutdown( + let handle = tokio::task::spawn(run_with_shutdown( + buffer, state_store.take_inner(), (), shutdown_trigger.make_shutdown(), @@ -823,7 +854,8 @@ async fn startup_gate_ready_nonblocking_flush_does_not_release() { let event_buffer = buffer.event_buffer.clone(); let state_store = TestStore::new().await; let shutdown_trigger = ComponentShutdownTrigger::default(); - let handle = tokio::task::spawn(buffer.run_with_shutdown( + let handle = tokio::task::spawn(run_with_shutdown( + buffer, state_store.take_inner(), (), shutdown_trigger.make_shutdown(), @@ -868,7 +900,8 @@ async fn startup_gate_releases_when_loaded_runtime_limits_expose_existing_pressu } let state_store = TestStore::new().await; let shutdown_trigger = ComponentShutdownTrigger::default(); - let handle = tokio::task::spawn(buffer.run_with_shutdown( + let handle = tokio::task::spawn(run_with_shutdown( + buffer, state_store.take_inner(), (), shutdown_trigger.make_shutdown(), @@ -1043,7 +1076,8 @@ async fn startup_gate_replays_previous_process_entries_before_current_entries() let state_store = TestStore::new().await; let shutdown_trigger = ComponentShutdownTrigger::default(); - let handle = tokio::task::spawn(buffer.run_with_shutdown( + let handle = tokio::task::spawn(run_with_shutdown( + buffer, state_store.take_inner(), (), shutdown_trigger.make_shutdown(), @@ -1061,6 +1095,7 @@ struct TestReplay { logs_notify: Arc, logs: Arc>>, fields: Arc>>, + feature_flags: Arc>>>, } struct StaticReportProcessor(parking_lot::Mutex>); @@ -1153,6 +1188,7 @@ impl TestReplay { logs_notify: Arc::new(Notify::new()), logs: Arc::new(parking_lot::Mutex::new(vec![])), fields: Arc::new(parking_lot::Mutex::new(vec![])), + feature_flags: Arc::new(parking_lot::Mutex::new(vec![])), } } } @@ -1163,13 +1199,19 @@ impl LogReplay for TestReplay { &mut self, log: Log, _processing_pipeline: &mut ProcessingPipeline, - _state: &bd_state::Store, + _state: &dyn StateReader, _now: OffsetDateTime, ) -> anyhow::Result { if let Some(message) = log.message.as_str() { self.logs.lock().push(message.to_string()); } + self.feature_flags.lock().push( + _state + .get(Scope::FeatureFlagExposure, "flag") + .filter(|value| value.has_string_value()) + .map(|value| value.string_value().to_string()), + ); self.fields.lock().push(log.fields); self.logs_count.fetch_add(1, Ordering::SeqCst); self.logs_notify.notify_waiters(); @@ -1674,7 +1716,12 @@ 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_with_previous_state( + state_store, + (), + Arc::new(bd_versioned_kv::ScopedMaps::default()), + )) + .await; }); shutdown.store(true, Ordering::SeqCst); @@ -1735,8 +1782,12 @@ async fn creates_workflows_engine_in_response_to_config_update() { let test_store = TestStore::new().await; let state_store = (*test_store).clone(); let shutdown_trigger = ComponentShutdownTrigger::default(); - let handle = - tokio::task::spawn(buffer.run_with_shutdown(state_store, (), shutdown_trigger.make_shutdown())); + let handle = tokio::task::spawn(run_with_shutdown( + buffer, + state_store, + (), + shutdown_trigger.make_shutdown(), + )); 1.seconds().sleep().await; shutdown_trigger.shutdown().await; buffer = handle.await.unwrap(); @@ -1774,8 +1825,12 @@ async fn updates_workflow_engine_in_response_to_config_update() { let test_store = TestStore::new().await; let state_store = (*test_store).clone(); let shutdown_trigger = ComponentShutdownTrigger::default(); - let handle = - tokio::task::spawn(buffer.run_with_shutdown(state_store, (), shutdown_trigger.make_shutdown())); + let handle = tokio::task::spawn(run_with_shutdown( + buffer, + state_store, + (), + shutdown_trigger.make_shutdown(), + )); 1.seconds().sleep().await; shutdown_trigger.shutdown().await; buffer = handle.await.unwrap(); @@ -1800,7 +1855,8 @@ async fn updates_workflow_engine_in_response_to_config_update() { // Timeout as otherwise buffer's workflows engine continues to try // to periodically flush its state to disk which hold us stuck here. let shutdown_trigger = ComponentShutdownTrigger::default(); - let handle = tokio::task::spawn(buffer.run_with_shutdown( + let handle = tokio::task::spawn(run_with_shutdown( + buffer, state_store.take_inner(), (), shutdown_trigger.make_shutdown(), @@ -1868,7 +1924,8 @@ async fn logs_resource_utilization_log() { // Timeout as otherwise buffer's workflows engine continues to try // to periodically flush its state to disk which hold us stuck here. let shutdown_trigger = ComponentShutdownTrigger::default(); - let handle = tokio::task::spawn(buffer.run_with_shutdown( + let handle = tokio::task::spawn(run_with_shutdown( + buffer, state_store.take_inner(), (), shutdown_trigger.make_shutdown(), @@ -1905,8 +1962,12 @@ async fn updates_system_session_id_for_new_sessions() { let test_store = TestStore::new().await; let state_store = (*test_store).clone(); let shutdown_trigger = ComponentShutdownTrigger::default(); - let handle = - tokio::task::spawn(buffer.run_with_shutdown(state_store, (), shutdown_trigger.make_shutdown())); + let handle = tokio::task::spawn(run_with_shutdown( + buffer, + state_store, + (), + shutdown_trigger.make_shutdown(), + )); wait_for_startup_gate_ready(&setup).await; let first_session_id = setup.session_strategy.session_id().unwrap(); @@ -1965,7 +2026,8 @@ async fn set_memory_pressure_level_writes_to_system_scope() { let test_store = TestStore::new().await; let shutdown_trigger = ComponentShutdownTrigger::default(); - let handle = tokio::task::spawn(buffer.run_with_shutdown( + let handle = tokio::task::spawn(run_with_shutdown( + buffer, (*test_store).clone(), (), shutdown_trigger.make_shutdown(), @@ -2017,8 +2079,12 @@ async fn previous_run_log_does_not_override_system_session_id() { let test_store = TestStore::new().await; let state_store = (*test_store).clone(); let shutdown_trigger = ComponentShutdownTrigger::default(); - let handle = - tokio::task::spawn(buffer.run_with_shutdown(state_store, (), shutdown_trigger.make_shutdown())); + let handle = tokio::task::spawn(run_with_shutdown( + buffer, + state_store, + (), + shutdown_trigger.make_shutdown(), + )); wait_for_startup_gate_ready(&setup).await; let current_session_id = setup.session_strategy.session_id().unwrap(); @@ -2090,6 +2156,311 @@ async fn previous_run_log_does_not_override_system_session_id() { task.join().unwrap(); } +#[test] +fn initial_field_state_updates_skip_unchanged_values() { + let initial_custom_fields: LogFields = [("field".into(), "value".into())].into(); + let mut state = InMemoryStateReader::new(); + state.insert( + Scope::CustomFields, + "field", + super::persistent_field_value(DataValue::String("value".to_string())), + ); + + assert!( + super::initial_field_state_updates(LogFields::default(), initial_custom_fields, &state,) + .is_empty() + ); +} + +#[tokio::test] +async fn ootb_ownership_prevents_custom_state_changes() { + let mut setup = Setup::new(); + let (_config_update_tx, config_update_rx) = mpsc::channel(1); + let (mut buffer, _) = setup.make_test_async_log_buffer(config_update_rx); + let state_store = TestStore::new().await; + let ootb_value = super::persistent_field_value(DataValue::String("ootb".to_string())); + let custom_value = super::persistent_field_value(DataValue::String("custom".to_string())); + + assert_ok!( + state_store + .insert(Scope::OotbFields, "shared".to_string(), ootb_value.clone()) + .await + ); + + buffer + .process_control( + LoggerControl::AddLogField( + "shared".to_string(), + DataValue::String("custom".to_string()), + ), + &state_store, + ) + .await; + + let state = state_store.read().await; + assert_eq!(state.get(Scope::OotbFields, "shared"), Some(&ootb_value)); + assert!(state.get(Scope::CustomFields, "shared").is_none()); + drop(state); + assert!(!buffer.metadata_collector.is_ootb_field("shared")); + + // Preserve a legacy custom value while its OOTB counterpart owns the virtual field. This can + // occur after upgrading from a version that allowed both state entries to coexist. + assert_ok!( + state_store + .insert( + Scope::CustomFields, + "shared".to_string(), + custom_value.clone() + ) + .await + ); + buffer + .process_control( + LoggerControl::RemoveLogField("shared".to_string()), + &state_store, + ) + .await; + + assert_eq!( + state_store.read().await.get(Scope::CustomFields, "shared"), + Some(&custom_value) + ); +} + +#[tokio::test] +async fn metadata_ootb_ownership_prevents_custom_state_changes() { + let mut setup = Setup::new(); + let (_config_update_tx, config_update_rx) = mpsc::channel(1); + let (mut buffer, _) = setup.make_test_async_log_buffer(config_update_rx); + let state_store = TestStore::new().await; + + buffer + .metadata_collector + .update_ootb_field("metadata_only".into(), "ootb".into()); + buffer + .process_control( + LoggerControl::AddLogField( + "metadata_only".to_string(), + DataValue::String("custom".to_string()), + ), + &state_store, + ) + .await; + + assert!( + state_store + .read() + .await + .get(Scope::CustomFields, "metadata_only") + .is_none() + ); + + let custom_value = super::persistent_field_value(DataValue::String("custom".to_string())); + assert_ok!( + state_store + .insert( + Scope::CustomFields, + "metadata_only".to_string(), + custom_value.clone(), + ) + .await + ); + buffer + .process_control( + LoggerControl::RemoveLogField("metadata_only".to_string()), + &state_store, + ) + .await; + + assert_eq!( + state_store + .read() + .await + .get(Scope::CustomFields, "metadata_only"), + Some(&custom_value) + ); +} + +#[tokio::test] +async fn capacity_rejected_field_updates_preserve_state_and_metadata() { + let mut setup = Setup::new(); + let (_config_update_tx, config_update_rx) = mpsc::channel(1); + let (mut buffer, _) = setup.make_test_async_log_buffer(config_update_rx); + let state_store = TestStore::new_with_config(PersistentStoreConfig { + initial_buffer_size: 8 * 1024, + max_capacity_bytes: 8 * 1024, + high_water_mark_ratio: 0.8, + }) + .await; + let custom_initial = DataValue::String("custom_initial".to_string()); + let ootb_initial = DataValue::String("ootb_initial".to_string()); + + buffer + .process_control( + LoggerControl::AddLogField("custom".to_string(), custom_initial.clone()), + &state_store, + ) + .await; + buffer + .process_control( + LoggerControl::UpdateOotbLogField("ootb".to_string(), ootb_initial.clone()), + &state_store, + ) + .await; + assert_ok!( + state_store + .insert( + Scope::System, + "unrelated".to_string(), + bd_state::string_value("x".repeat(6_200)), + ) + .await + ); + + // These writes cannot fit alongside the surviving system state after compaction. Rejection + // must leave both the virtual state and the metadata source of emitted fields unchanged. + let rejected = DataValue::String("rejected".repeat(585)); + buffer + .process_control( + LoggerControl::AddLogField("custom".to_string(), rejected.clone()), + &state_store, + ) + .await; + buffer + .process_control( + LoggerControl::UpdateOotbLogField("ootb".to_string(), rejected), + &state_store, + ) + .await; + + let state = state_store.read().await; + assert_eq!( + state.get(Scope::CustomFields, "custom"), + Some(&super::persistent_field_value(custom_initial.clone())) + ); + assert_eq!( + state.get(Scope::OotbFields, "ootb"), + Some(&super::persistent_field_value(ootb_initial.clone())) + ); + drop(state); + + let (ootb_fields, custom_fields) = buffer.metadata_collector.initial_persistent_fields(); + assert_eq!(custom_fields.get("custom"), Some(&custom_initial)); + assert_eq!(ootb_fields.get("ootb"), Some(&ootb_initial)); +} + +#[tokio::test] +async fn capacity_rejected_initial_fields_are_dropped_from_metadata() { + let mut setup = Setup::new(); + let (_config_update_tx, config_update_rx) = mpsc::channel(1); + let (mut buffer, _) = setup.make_test_async_log_buffer(config_update_rx); + let state_store = TestStore::new_with_config(PersistentStoreConfig { + initial_buffer_size: 8 * 1024, + max_capacity_bytes: 8 * 1024, + high_water_mark_ratio: 0.8, + }) + .await; + assert_ok!( + state_store + .insert( + Scope::System, + "unrelated".to_string(), + bd_state::string_value("x".repeat(6_200)), + ) + .await + ); + let rejected = DataValue::String("rejected".repeat(585)); + assert_ok!( + buffer + .metadata_collector + .add_field("initial_custom".into(), rejected.clone()) + ); + buffer + .metadata_collector + .update_ootb_field("initial_ootb".into(), rejected.clone()); + + // Startup has no previous inline value to retain, so an unpersistable field is dropped rather + // than emitted without a matching virtual state value. + buffer + .persist_initial_log_fields( + [("initial_ootb".into(), rejected.clone())].into(), + [("initial_custom".into(), rejected)].into(), + &state_store, + ) + .await; + + let state = state_store.read().await; + assert!(state.get(Scope::CustomFields, "initial_custom").is_none()); + assert!(state.get(Scope::OotbFields, "initial_ootb").is_none()); + drop(state); + + let (ootb_fields, custom_fields) = buffer.metadata_collector.initial_persistent_fields(); + assert!(!custom_fields.contains_key("initial_custom")); + assert!(!ootb_fields.contains_key("initial_ootb")); +} + +#[tokio::test] +async fn previous_process_logs_use_snapshot_state() { + let mut setup = Setup::new(); + let (config_update_tx, config_update_rx) = mpsc::channel(1); + let (buffer, sender) = setup.make_test_async_log_buffer(config_update_rx); + let state_store = TestStore::new().await; + assert_ok!( + state_store + .insert( + Scope::FeatureFlagExposure, + "flag".to_string(), + bd_state::string_value("current"), + ) + .await + ); + let mut previous_run_state = bd_versioned_kv::ScopedMaps::default(); + previous_run_state.insert( + Scope::FeatureFlagExposure, + "flag".to_string(), + bd_versioned_kv::TimestampedValue { + timestamp: 0, + value: bd_state::string_value("previous"), + }, + ); + let config_update = setup.make_config_update(WorkflowsConfiguration::default()); + let task = std::thread::spawn(move || { + assert_ok!(config_update_tx.blocking_send(config_update)); + }); + let shutdown_trigger = ComponentShutdownTrigger::default(); + let handle = tokio::task::spawn(buffer.run_with_shutdown_and_previous_state( + state_store.take_inner(), + (), + Arc::new(previous_run_state), + shutdown_trigger.make_shutdown(), + )); + wait_for_startup_gate_ready(&setup).await; + + sender + .try_send_log(LogLine { + log_level: log_level::DEBUG, + log_type: LogType::NORMAL, + message: "previous".into(), + fields: AnnotatedLogFields::new(), + matching_fields: AnnotatedLogFields::new(), + attributes_overrides: Some(LogAttributesOverrides::PreviousRunSessionID( + OffsetDateTime::now_utc(), + )), + capture_session: None, + }) + .unwrap(); + wait_for_replayed_logs(&setup, 1).await; + + shutdown_trigger.shutdown().await; + handle.await.unwrap(); + task.join().unwrap(); + + assert_eq!( + &[Some("previous".to_string())], + setup.replayer_feature_flags.lock().as_slice() + ); +} + #[tokio::test] async fn processes_log_with_global_state_in_attributes_overrides() { let mut setup = Setup::new(); @@ -2116,7 +2487,8 @@ async fn processes_log_with_global_state_in_attributes_overrides() { let state_store = TestStore::new().await; let shutdown_trigger = ComponentShutdownTrigger::default(); - let handle = tokio::task::spawn(buffer.run_with_shutdown( + let handle = tokio::task::spawn(run_with_shutdown( + buffer, state_store.take_inner(), (), shutdown_trigger.make_shutdown(), @@ -2183,7 +2555,8 @@ async fn processes_log_with_global_state_in_attributes_overrides() { let shutdown_trigger_2 = ComponentShutdownTrigger::default(); let state_store_2 = TestStore::new().await; - let handle_2 = tokio::task::spawn(buffer_2.run_with_shutdown( + let handle_2 = tokio::task::spawn(run_with_shutdown( + buffer_2, state_store_2.take_inner(), (), shutdown_trigger_2.make_shutdown(), diff --git a/bd-logger/src/builder.rs b/bd-logger/src/builder.rs index d58fc2f3f..b71da6ead 100644 --- a/bd-logger/src/builder.rs +++ b/bd-logger/src/builder.rs @@ -489,6 +489,8 @@ impl LoggerBuilder { result.previous_state, result.retention_registry, ); + // The crash monitor and log buffer only read this immutable startup snapshot. + let previous_run_state = Arc::new(previous_run_state); let pending_entity_id = pending_entity_id.lock().take(); initialize_opaque_entity_updates(&state_store, &opaque_entity_updates_tx, pending_entity_id) @@ -549,6 +551,7 @@ impl LoggerBuilder { (None, None) }; + let previous_run_state_for_log_buffer = previous_run_state.clone(); let crash_monitor = Monitor::new( &self.params.sdk_directory, self.params.store.clone(), @@ -652,7 +655,12 @@ impl LoggerBuilder { Ok(()) }, async move { - async_log_buffer.run(state_store, crash_monitor).await; + Box::pin(async_log_buffer.run_with_previous_state( + state_store, + crash_monitor, + previous_run_state_for_log_buffer, + )) + .await; Ok(()) }, async move { diff --git a/bd-logger/src/log_replay.rs b/bd-logger/src/log_replay.rs index 4e3311f11..711b0ed74 100644 --- a/bd-logger/src/log_replay.rs +++ b/bd-logger/src/log_replay.rs @@ -64,7 +64,7 @@ pub trait LogReplay { &mut self, log: Log, pipeline: &mut ProcessingPipeline, - state: &bd_state::Store, + state: &dyn bd_state::StateReader, now: OffsetDateTime, ) -> anyhow::Result; @@ -93,10 +93,10 @@ impl LogReplay for LoggerReplay { &mut self, log: Log, pipeline: &mut ProcessingPipeline, - state_store: &bd_state::Store, + state: &dyn bd_state::StateReader, now: OffsetDateTime, ) -> anyhow::Result { - pipeline.process_log(log, state_store, now).await + pipeline.process_log(log, state, now).await } async fn replay_state_change( @@ -253,17 +253,16 @@ impl ProcessingPipeline { async fn process_log( &mut self, mut log: Log, - state: &bd_state::Store, + state: &dyn bd_state::StateReader, now: OffsetDateTime, ) -> anyhow::Result { self.stats.logs_received.inc(); - let state_reader = state.read().await; // TODO(Augustyniak): Add a histogram for the time it takes to process a log. - self.filter_chain.process(&mut log, &state_reader); + self.filter_chain.process(&mut log, state); let mut log = EncodableLog::new(log, (*self.min_log_compression_size.read()).into()); - match self.tail_configs.maybe_stream_log(&mut log, &state_reader) { + match self.tail_configs.maybe_stream_log(&mut log, state) { Ok(streamed) => { if streamed { self.stats.streamed_logs.inc(); @@ -279,13 +278,13 @@ impl ProcessingPipeline { log.log.log_level, &log.log.message, FieldsRef::new(&log.log.fields, &log.log.matching_fields), - &state_reader, + state, ); let mut result = self.workflows_engine.process_event( WorkflowEvent::Log(&log.log), &matching_buffers, - &state_reader, + state, now, ); self diff --git a/bd-logger/src/metadata.rs b/bd-logger/src/metadata.rs index 8eb8ed704..da816acdc 100644 --- a/bd-logger/src/metadata.rs +++ b/bd-logger/src/metadata.rs @@ -103,9 +103,10 @@ impl MetadataCollector { } /// Returns log metadata using the last active global state at the end of the last process run. - /// Log fields take precedence over persisted global state fields to allow the caller to - /// override values in global state, e.g. when the crash handler knows that the event happened - /// in the background. + /// + /// The previous-global-state map remains the inline source until field elision is enabled. + /// State-backed fields are passed separately to the replay pipeline for matcher evaluation; + /// this preserves the prior inline payload behavior without decoding persisted values. /// Does *not* invoke the field providers as these would incorrectly reflect the state of the /// current process. pub(crate) fn metadata_from_fields_with_previous_global_state( @@ -114,17 +115,13 @@ impl MetadataCollector { global_state_reader: &global_state::Reader, timestamp: time::OffsetDateTime, ) -> LogMetadata { - let fields = if let Some(previous_global_state_fields) = - global_state_reader.previous_global_state_fields() - { - previous_global_state_fields - .clone() - .into_iter() - .chain(fields.into_iter().map(|(k, v)| (k, v.value))) - .collect() - } else { - fields.into_iter().map(|(k, v)| (k, v.value)).collect() - }; + let fields = global_state_reader + .previous_global_state_fields() + .cloned() + .unwrap_or_default() + .into_iter() + .chain(fields.into_iter().map(|(k, v)| (k, v.value))) + .collect(); LogMetadata { timestamp, @@ -284,6 +281,28 @@ impl MetadataCollector { entry.remove(); } } + + pub(crate) fn remove_ootb_field(&mut self, field_key: LogFieldKey) { + if let Entry::Occupied(entry) = self.fields.entry(field_key) + && entry.get().kind == LogFieldKind::Ootb + { + entry.remove(); + } + } + + /// Returns whether an OOTB field currently owns `key`. + pub(crate) fn is_ootb_field(&self, key: &str) -> bool { + self + .fields + .get(key) + .is_some_and(|field| field.kind == LogFieldKind::Ootb) + } + + /// Returns the persistent fields supplied during logger construction for state-store seeding. + pub(crate) fn initial_persistent_fields(&self) -> (LogFields, LogFields) { + let PartitionedFields { ootb, custom } = partition_fields(self.fields.clone()); + (ootb, custom) + } } fn partition_fields(field: AnnotatedLogFields) -> PartitionedFields { @@ -309,7 +328,7 @@ fn partition_fields(field: AnnotatedLogFields) -> PartitionedFields { PartitionedFields { ootb, custom } } -fn verify_custom_field_name(key: &str) -> anyhow::Result<()> { +pub fn verify_custom_field_name(key: &str) -> anyhow::Result<()> { if RESERVED_FIELD_NAMES.contains(key) { anyhow::bail!( "Custom global field with {key:?} name is not allowed as the name is reserved for SDK \ diff --git a/bd-logger/src/metadata_test.rs b/bd-logger/src/metadata_test.rs index 975d888ed..9e22c73ec 100644 --- a/bd-logger/src/metadata_test.rs +++ b/bd-logger/src/metadata_test.rs @@ -437,13 +437,15 @@ fn expected_field_value(fields: &LogFields, key: &str) -> Option { } #[test] -fn metadata_from_fields_with_previous_global_state_includes_global_fields() { +fn metadata_from_fields_with_previous_global_state_uses_legacy_inline_fields() { let store = in_memory_store(); let mut tracker = global_state::Tracker::new(store.clone(), Watch::new_for_testing(10.seconds())); // Setup global state let global_fields = [ ("global_key".into(), "global_value".into()), + ("custom_key".into(), "legacy_custom_value".into()), + ("ootb_key".into(), "legacy_ootb_value".into()), ("shared_key".into(), "global_value".into()), ] .into(); @@ -463,7 +465,6 @@ fn metadata_from_fields_with_previous_global_state_includes_global_fields() { .into(); let reader = Reader::new(store); - let metadata = MetadataCollector::metadata_from_fields_with_previous_global_state( input_fields, [].into(), @@ -479,6 +480,16 @@ fn metadata_from_fields_with_previous_global_state_includes_global_fields() { expected_field_value(&metadata.fields, "global_key").unwrap() ); + // Previous-process payloads continue to use the crash-global-state fields until elision. + assert_eq!( + "legacy_custom_value", + expected_field_value(&metadata.fields, "custom_key").unwrap() + ); + assert_eq!( + "legacy_ootb_value", + expected_field_value(&metadata.fields, "ootb_key").unwrap() + ); + // Unique local field should be present assert_eq!( "local_value", diff --git a/bd-logger/src/test/stats_integration.rs b/bd-logger/src/test/stats_integration.rs index f9d65971c..5fb28a359 100644 --- a/bd-logger/src/test/stats_integration.rs +++ b/bd-logger/src/test/stats_integration.rs @@ -132,6 +132,11 @@ fn read_index(setup: &Setup) -> PendingAggregationIndex { read_compressed_protobuf(&fs::read(setup.pending_aggregation_index_file_path()).unwrap()).unwrap() } +fn try_read_index(setup: &Setup) -> Option { + let contents = fs::read(setup.pending_aggregation_index_file_path()).ok()?; + read_compressed_protobuf(&contents).ok() +} + #[test] fn inline_startup_upload_success_is_acked_and_reported_on_next_connection() { let directory = TempDir::new().unwrap(); @@ -957,15 +962,14 @@ fn snapshot_rotation_drop_is_reported_on_the_next_handshake() { let _initial_handshake = setup.server.blocking_next_handshake_request().unwrap(); setup.trigger_periodic_stats_flush(); - wait_for!(setup.pending_aggregation_index_file_path().exists()); - wait_for!(read_index(&setup).pending_files.len() == 1); + wait_for!(try_read_index(&setup).is_some_and(|index| index.pending_files.len() == 1)); setup.trigger_periodic_stats_flush(); - wait_for!( - read_index(&setup) + wait_for!(try_read_index(&setup).is_some_and(|index| { + index .unreported_stats_pipeline_analytics .as_ref() .is_some_and(|analytics| analytics.stats_files_dropped_due_to_rotation == 1) - ); + })); setup.restart_stream(false); let (_, handshake) = setup.server.blocking_next_handshake_request().unwrap(); diff --git a/bd-state/src/lib.rs b/bd-state/src/lib.rs index fdac44178..13676f440 100644 --- a/bd-state/src/lib.rs +++ b/bd-state/src/lib.rs @@ -285,7 +285,7 @@ pub struct StateEntry { /// A trait for reading state values. This pattern allows for non-async access to state values while /// the underlying store may be async. -pub trait StateReader { +pub trait StateReader: Sync { /// Gets a reference to the raw state value from the store. fn get(&self, scope: Scope, key: &str) -> Option<&StateValue>; @@ -296,6 +296,35 @@ pub trait StateReader { fn as_scoped_maps(&self) -> &ScopedMaps; } +impl StateReader for ScopedMaps { + fn get(&self, scope: Scope, key: &str) -> Option<&StateValue> { + Self::get(self, scope, key).map(|value| &value.value) + } + + fn iter(&self) -> Box + '_> { + Box::new( + Self::iter(self).filter_map(|(scope, key, timestamped_value)| { + let timestamp = OffsetDateTime::from_unix_timestamp_nanos( + i128::from(timestamped_value.timestamp) * 1_000, + ) + .ok()?; + timestamped_value.value.value_type.as_ref()?; + + Some(StateEntry { + scope, + key: key.clone(), + value: timestamped_value.value.clone(), + timestamp, + }) + }), + ) + } + + fn as_scoped_maps(&self) -> &ScopedMaps { + self + } +} + // // Store // diff --git a/bd-state/src/lib_test.rs b/bd-state/src/lib_test.rs index c4f45d626..93203a16b 100644 --- a/bd-state/src/lib_test.rs +++ b/bd-state/src/lib_test.rs @@ -424,6 +424,69 @@ async fn system_scope_persists_on_restart() { } } +#[tokio::test] +async fn log_field_scopes_persist_on_restart() { + let temp_dir = tempfile::tempdir().unwrap(); + let time_provider = Arc::new(bd_time::TestTimeProvider::new( + datetime!(2024-01-01 00:00:00 UTC), + )); + let runtime_loader = bd_runtime::runtime::ConfigLoader::new(temp_dir.path()); + + { + let store = Store::persistent( + temp_dir.path(), + PersistentStoreConfig::default(), + time_provider.clone(), + &runtime_loader, + &Collector::default().scope("test"), + ) + .await + .unwrap() + .store; + + for (scope, key, value) in [ + (Scope::CustomFields, "custom_key", "custom_value"), + (Scope::OotbFields, "ootb_key", "ootb_value"), + ] { + store + .insert(scope, key.to_string(), crate::string_value(value)) + .await + .unwrap() + .unwrap(); + } + } + + let result = Store::persistent( + temp_dir.path(), + PersistentStoreConfig::default(), + time_provider, + &runtime_loader, + &Collector::default().scope("test"), + ) + .await + .unwrap(); + + for (scope, key, value) in [ + (Scope::CustomFields, "custom_key", "custom_value"), + (Scope::OotbFields, "ootb_key", "ootb_value"), + ] { + assert!( + result + .previous_state + .get(scope, key) + .is_some_and(|entry| entry.value.has_string_value() && entry.value.string_value() == value) + ); + assert!( + result + .store + .read() + .await + .get(scope, key) + .is_some_and(|entry| entry.has_string_value() && entry.string_value() == value) + ); + } +} + #[tokio::test] async fn session_id_persists_while_ephemeral_scopes_clear() { let temp_dir = tempfile::tempdir().unwrap(); diff --git a/bd-state/src/test.rs b/bd-state/src/test.rs index 4081fdfa7..d577e3b9d 100644 --- a/bd-state/src/test.rs +++ b/bd-state/src/test.rs @@ -27,13 +27,18 @@ pub struct TestStore { impl TestStore { #[must_use] pub async fn new() -> Self { + Self::new_with_config(crate::PersistentStoreConfig::default()).await + } + + #[must_use] + pub async fn new_with_config(config: crate::PersistentStoreConfig) -> Self { let temp_dir = tempfile::tempdir().unwrap(); let time_provider = Arc::new(bd_time::TestTimeProvider::new( datetime!(2024-01-01 00:00:00 UTC), )); let store = crate::Store::persistent( temp_dir.path(), - crate::PersistentStoreConfig::default(), + config, time_provider.clone(), &bd_runtime::runtime::ConfigLoader::new(temp_dir.path()), &bd_client_stats_store::Collector::default().scope("test"), diff --git a/bd-versioned-kv/src/lib.rs b/bd-versioned-kv/src/lib.rs index 73f970e8b..a8db02ffe 100644 --- a/bd-versioned-kv/src/lib.rs +++ b/bd-versioned-kv/src/lib.rs @@ -37,8 +37,8 @@ pub const MAX_DECOMPRESSED_STATE_SNAPSHOT_BYTES: usize = 10 * 1024 * 1024; /// limit above. pub const MAX_COMPRESSED_STATE_SNAPSHOT_BYTES: usize = 10 * 1024 * 1024; -pub use bd_proto::protos::state::payload::StateValue; -pub use bd_proto::protos::state::payload::state_value::Value_type; +pub use bd_proto::protos::state::state_payload::StateValue; +pub use bd_proto::protos::state::state_payload::state_value::Value_type; pub use scope::Scope; pub use versioned_kv_journal::filename::SnapshotFilename; pub use versioned_kv_journal::recovery::{ diff --git a/bd-versioned-kv/src/scope.rs b/bd-versioned-kv/src/scope.rs index 9ec4578ab..68b642c9e 100644 --- a/bd-versioned-kv/src/scope.rs +++ b/bd-versioned-kv/src/scope.rs @@ -15,6 +15,8 @@ pub enum Scope { FeatureFlagExposure = 1, GlobalState = 2, System = 3, + CustomFields = 4, + OotbFields = 5, } impl Scope { diff --git a/bd-versioned-kv/src/tests/mod.rs b/bd-versioned-kv/src/tests/mod.rs index 35cc80043..a9fe5448e 100644 --- a/bd-versioned-kv/src/tests/mod.rs +++ b/bd-versioned-kv/src/tests/mod.rs @@ -40,9 +40,9 @@ pub fn decompress_zlib(data: &[u8]) -> anyhow::Result> { Ok(decompressed) } -pub fn make_string_value(s: &str) -> state::payload::StateValue { - state::payload::StateValue { - value_type: Some(state::payload::state_value::Value_type::StringValue( +pub fn make_string_value(s: &str) -> state::state_payload::StateValue { + state::state_payload::StateValue { + value_type: Some(state::state_payload::state_value::Value_type::StringValue( s.to_string(), )), ..Default::default() diff --git a/bd-versioned-kv/src/tests/versioned_kv_store_dynamic_growth_test.rs b/bd-versioned-kv/src/tests/versioned_kv_store_dynamic_growth_test.rs index bf37824b8..7eeb0d423 100644 --- a/bd-versioned-kv/src/tests/versioned_kv_store_dynamic_growth_test.rs +++ b/bd-versioned-kv/src/tests/versioned_kv_store_dynamic_growth_test.rs @@ -517,3 +517,101 @@ async fn compaction_capacity_rejection_keeps_persistent_store() -> anyhow::Resul Ok(()) } + +/// A deletion must remain visible after its tombstone cannot fit in the persistent journal. +#[tokio::test] +async fn capacity_rejected_removal_falls_back_to_in_memory() -> anyhow::Result<()> { + use bd_client_stats_store::test::StatsHelper; + use std::collections::BTreeMap; + + let setup = Setup::new(); + let config = PersistentStoreConfig { + initial_buffer_size: 8 * 1024, + max_capacity_bytes: 8 * 1024, + high_water_mark_ratio: 0.8, + }; + let mut store = setup.open_store(config).await?; + + // The live entries fit in the journal, but replaying the long-key tombstone alongside the + // pre-deletion compacted map would not. The delete must still take effect in memory. + let deleted_key = "deleted".repeat(143); + let deleted_value = make_string_value("value"); + let unrelated_value = make_string_value(&"x".repeat(6_200)); + store + .insert( + Scope::GlobalState, + deleted_key.clone(), + deleted_value.clone(), + ) + .await?; + store + .insert( + Scope::GlobalState, + "unrelated".to_string(), + unrelated_value.clone(), + ) + .await?; + assert_eq!(store.persistence_mode(), PersistenceMode::Persistent); + + let removed = store.remove(Scope::GlobalState, &deleted_key).await?; + + assert!(removed.is_some_and(|(_, value)| value == deleted_value)); + assert!(!store.contains_key(Scope::GlobalState, &deleted_key)); + assert_eq!( + store.get(Scope::GlobalState, "unrelated"), + Some(&unrelated_value) + ); + assert_eq!(store.persistence_mode(), PersistenceMode::InMemory); + assert!(store.journal_path().is_none()); + setup + .collector + .assert_counter_eq(1, "test:kv:persistence_fallbacks", BTreeMap::new()); + + Ok(()) +} + +/// A scope clear must retain its deletion result when the batch tombstone cannot fit. +#[tokio::test] +async fn capacity_rejected_scope_clear_falls_back_to_in_memory() -> anyhow::Result<()> { + use bd_client_stats_store::test::StatsHelper; + use std::collections::BTreeMap; + + let setup = Setup::new(); + let config = PersistentStoreConfig { + initial_buffer_size: 8 * 1024, + max_capacity_bytes: 8 * 1024, + high_water_mark_ratio: 0.8, + }; + let mut store = setup.open_store(config).await?; + + // The system value survives the clear. Replaying it with the long global-state tombstone does + // not fit after compaction, so clear must fall back to bounded in-memory state. + let deleted_key = "deleted".repeat(143); + let deleted_value = make_string_value("value"); + let unrelated_value = make_string_value(&"x".repeat(6_200)); + store + .insert(Scope::GlobalState, deleted_key.clone(), deleted_value) + .await?; + store + .insert( + Scope::System, + "unrelated".to_string(), + unrelated_value.clone(), + ) + .await?; + assert_eq!(store.persistence_mode(), PersistenceMode::Persistent); + + assert!(store.clear(Scope::GlobalState).await?.is_some()); + assert!(!store.contains_key(Scope::GlobalState, &deleted_key)); + assert_eq!( + store.get(Scope::System, "unrelated"), + Some(&unrelated_value) + ); + assert_eq!(store.persistence_mode(), PersistenceMode::InMemory); + assert!(store.journal_path().is_none()); + setup + .collector + .assert_counter_eq(1, "test:kv:persistence_fallbacks", BTreeMap::new()); + + Ok(()) +} diff --git a/bd-versioned-kv/src/tests/versioned_kv_store_test.rs b/bd-versioned-kv/src/tests/versioned_kv_store_test.rs index 4ef8e5014..48c1e2283 100644 --- a/bd-versioned-kv/src/tests/versioned_kv_store_test.rs +++ b/bd-versioned-kv/src/tests/versioned_kv_store_test.rs @@ -12,7 +12,7 @@ use crate::versioned_kv_journal::retention::{RetentionHandle, RetentionRegistry} use crate::versioned_kv_journal::store::PersistentStoreConfig; use crate::versioned_kv_journal::{TimestampedValue, make_string_value}; use crate::{DataLoss, PersistenceMode, Scope, UpdateError, VersionedKVStore}; -use bd_proto::protos::state::payload::StateValue; +use bd_proto::protos::state::state_payload::StateValue; use bd_time::TestTimeProvider; use rstest::rstest; use std::sync::Arc; diff --git a/bd-versioned-kv/src/versioned_kv_journal/framing_test.rs b/bd-versioned-kv/src/versioned_kv_journal/framing_test.rs index 8bef6e931..4a534f5b5 100644 --- a/bd-versioned-kv/src/versioned_kv_journal/framing_test.rs +++ b/bd-versioned-kv/src/versioned_kv_journal/framing_test.rs @@ -10,7 +10,7 @@ use super::*; use crate::Scope; use crate::tests::make_string_value; -use bd_proto::protos::state::payload::StateValue; +use bd_proto::protos::state::state_payload::StateValue; #[test] fn varint_encoding() { diff --git a/bd-versioned-kv/src/versioned_kv_journal/journal.rs b/bd-versioned-kv/src/versioned_kv_journal/journal.rs index 5f8a43595..942361f55 100644 --- a/bd-versioned-kv/src/versioned_kv_journal/journal.rs +++ b/bd-versioned-kv/src/versioned_kv_journal/journal.rs @@ -5,6 +5,10 @@ // LICENSE.polyform file or at: // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt +#[cfg(test)] +#[path = "./journal_test.rs"] +mod tests; + use super::framing::Frame; use crate::{Scope, UpdateError}; use bd_client_common::error::InvariantError; diff --git a/bd-versioned-kv/src/versioned_kv_journal/journal_test.rs b/bd-versioned-kv/src/versioned_kv_journal/journal_test.rs new file mode 100644 index 000000000..6462619e5 --- /dev/null +++ b/bd-versioned-kv/src/versioned_kv_journal/journal_test.rs @@ -0,0 +1,64 @@ +// 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 + +#![allow(clippy::unwrap_used)] + +use super::*; +use crate::tests::make_string_value; +use bd_proto::protos::state::state_payload::StateValue; +use bd_time::TestTimeProvider; +use crc32fast::Hasher; +use std::cell::Cell; +use std::sync::Arc; +use time::macros::datetime; + +#[test] +fn unknown_scope_marks_partial_data_loss_without_startup_failure() { + let time_provider = Arc::new(TestTimeProvider::new(datetime!(2024-01-01 00:00:00 UTC))); + let mut buffer = vec![0_u8; 4096]; + + VersionedJournal::new( + &mut buffer, + 0.8, + time_provider.clone(), + [( + Scope::CustomFields, + "field".to_string(), + make_string_value("value"), + 1, + )], + ) + .unwrap(); + + // Older SDKs do not recognize the CustomFields scope byte. The test models their decoder with + // an otherwise valid future scope frame, which must degrade recovery rather than fail startup. + assert_eq!( + buffer[HEADER_SIZE] & 0x80, + 0, + "test frame length must use one byte" + ); + let frame_start = HEADER_SIZE + 1; + let frame_len = usize::from(buffer[HEADER_SIZE]); + let crc_start = frame_start + frame_len - 4; + buffer[frame_start] = u8::MAX; + + let mut hasher = Hasher::new(); + hasher.update(&buffer[frame_start .. crc_start]); + buffer[crc_start .. crc_start + 4].copy_from_slice(&hasher.finalize().to_le_bytes()); + + let callback_called = Cell::new(false); + let (_journal, data_loss) = VersionedJournal::::from_buffer( + &mut buffer, + 0.8, + time_provider, + |_scope, _key, _value, _timestamp| callback_called.set(true), + ) + .unwrap(); + + assert!(!callback_called.get()); + assert!(matches!(data_loss, PartialDataLoss::Yes)); +} diff --git a/bd-versioned-kv/src/versioned_kv_journal/mod.rs b/bd-versioned-kv/src/versioned_kv_journal/mod.rs index 519ec1a69..ea63513da 100644 --- a/bd-versioned-kv/src/versioned_kv_journal/mod.rs +++ b/bd-versioned-kv/src/versioned_kv_journal/mod.rs @@ -41,7 +41,7 @@ pub enum UpdateError { #[derive(Debug, Clone, PartialEq)] pub struct TimestampedValue { /// The value stored in the key-value store. - pub value: state::payload::StateValue, + pub value: state::state_payload::StateValue, /// The timestamp (in microseconds since UNIX epoch) when this value was last written. pub timestamp: u64, @@ -49,9 +49,9 @@ pub struct TimestampedValue { #[cfg(test)] #[must_use] -pub fn make_string_value(s: &str) -> state::payload::StateValue { - state::payload::StateValue { - value_type: Some(state::payload::state_value::Value_type::StringValue( +pub fn make_string_value(s: &str) -> state::state_payload::StateValue { + state::state_payload::StateValue { + value_type: Some(state::state_payload::state_value::Value_type::StringValue( s.to_string(), )), ..Default::default() diff --git a/bd-versioned-kv/src/versioned_kv_journal/recovery.rs b/bd-versioned-kv/src/versioned_kv_journal/recovery.rs index 0e583baff..c62d464a5 100644 --- a/bd-versioned-kv/src/versioned_kv_journal/recovery.rs +++ b/bd-versioned-kv/src/versioned_kv_journal/recovery.rs @@ -10,7 +10,7 @@ use crate::versioned_kv_journal::framing::{Frame, decode_raw_frame}; use crate::versioned_kv_journal::journal::{HEADER_SIZE, VERSION}; use crate::{MAX_COMPRESSED_STATE_SNAPSHOT_BYTES, Scope}; use ahash::{AHashMap, AHashSet}; -use bd_proto::protos::state::payload::StateValue; +use bd_proto::protos::state::state_payload::StateValue; use flate2::{Decompress, FlushDecompress, Status}; use protobuf::Message; use std::io::{ErrorKind, Read}; @@ -347,8 +347,9 @@ pub fn extract_non_empty_string_values_from_compressed_journal( if frame.scope == scope && frame.key == key { let value = StateValue::parse_from_bytes(frame.payload) .map_err(|error| anyhow::anyhow!("Invalid state value at offset {offset}: {error}"))?; - if let Some(bd_proto::protos::state::payload::state_value::Value_type::StringValue(value)) = - value.value_type + if let Some(bd_proto::protos::state::state_payload::state_value::Value_type::StringValue( + value, + )) = value.value_type && !value.is_empty() && seen_values.insert(value.clone()) { diff --git a/bd-versioned-kv/src/versioned_kv_journal/store.rs b/bd-versioned-kv/src/versioned_kv_journal/store.rs index f5bca7249..bccd40b21 100644 --- a/bd-versioned-kv/src/versioned_kv_journal/store.rs +++ b/bd-versioned-kv/src/versioned_kv_journal/store.rs @@ -13,7 +13,7 @@ use crate::versioned_kv_journal::retention::RetentionRegistry; use crate::{Scope, UpdateError}; use ahash::AHashMap; use bd_error_reporter::reporter::handle_unexpected; -use bd_proto::protos::state::payload::StateValue; +use bd_proto::protos::state::state_payload::StateValue; use bd_runtime::runtime::IntWatch; use bd_stats_common::Counter; use bd_time::TimeProvider; @@ -141,6 +141,8 @@ pub struct ScopedMaps { pub feature_flags: AHashMap, pub global_state: AHashMap, pub system: AHashMap, + pub custom_fields: AHashMap, + pub ootb_fields: AHashMap, } impl ScopedMaps { @@ -150,6 +152,8 @@ impl ScopedMaps { Scope::FeatureFlagExposure => self.feature_flags.get(key), Scope::GlobalState => self.global_state.get(key), Scope::System => self.system.get(key), + Scope::CustomFields => self.custom_fields.get(key), + Scope::OotbFields => self.ootb_fields.get(key), } } @@ -164,6 +168,8 @@ impl ScopedMaps { Scope::FeatureFlagExposure => self.feature_flags.insert(key, value), Scope::GlobalState => self.global_state.insert(key, value), Scope::System => self.system.insert(key, value), + Scope::CustomFields => self.custom_fields.insert(key, value), + Scope::OotbFields => self.ootb_fields.insert(key, value), } } @@ -172,6 +178,8 @@ impl ScopedMaps { Scope::FeatureFlagExposure => self.feature_flags.remove(key), Scope::GlobalState => self.global_state.remove(key), Scope::System => self.system.remove(key), + Scope::CustomFields => self.custom_fields.remove(key), + Scope::OotbFields => self.ootb_fields.remove(key), } } @@ -181,17 +189,27 @@ impl ScopedMaps { Scope::FeatureFlagExposure => self.feature_flags.contains_key(key), Scope::GlobalState => self.global_state.contains_key(key), Scope::System => self.system.contains_key(key), + Scope::CustomFields => self.custom_fields.contains_key(key), + Scope::OotbFields => self.ootb_fields.contains_key(key), } } #[must_use] pub fn len(&self) -> usize { - self.feature_flags.len() + self.global_state.len() + self.system.len() + self.feature_flags.len() + + self.global_state.len() + + self.system.len() + + self.custom_fields.len() + + self.ootb_fields.len() } #[must_use] pub fn is_empty(&self) -> bool { - self.feature_flags.is_empty() && self.global_state.is_empty() && self.system.is_empty() + self.feature_flags.is_empty() + && self.global_state.is_empty() + && self.system.is_empty() + && self.custom_fields.is_empty() + && self.ootb_fields.is_empty() } pub fn iter(&self) -> impl Iterator { @@ -206,6 +224,31 @@ impl ScopedMaps { .map(|(k, v)| (Scope::GlobalState, k, v)), ) .chain(self.system.iter().map(|(k, v)| (Scope::System, k, v))) + .chain( + self + .custom_fields + .iter() + .map(|(k, v)| (Scope::CustomFields, k, v)), + ) + .chain( + self + .ootb_fields + .iter() + .map(|(k, v)| (Scope::OotbFields, k, v)), + ) + } + + /// Returns the entries in one scope without visiting the other scoped maps. + pub fn iter_scope(&self, scope: Scope) -> impl Iterator { + let map = match scope { + Scope::FeatureFlagExposure => &self.feature_flags, + Scope::GlobalState => &self.global_state, + Scope::System => &self.system, + Scope::CustomFields => &self.custom_fields, + Scope::OotbFields => &self.ootb_fields, + }; + + map.iter() } fn values(&self) -> impl Iterator { @@ -214,6 +257,8 @@ impl ScopedMaps { .values() .chain(self.global_state.values()) .chain(self.system.values()) + .chain(self.custom_fields.values()) + .chain(self.ootb_fields.values()) } /// Get a mutable entry for the given scope and key, allowing efficient insert/update operations. @@ -226,6 +271,8 @@ impl ScopedMaps { Scope::FeatureFlagExposure => self.feature_flags.entry(key), Scope::GlobalState => self.global_state.entry(key), Scope::System => self.system.entry(key), + Scope::CustomFields => self.custom_fields.entry(key), + Scope::OotbFields => self.ootb_fields.entry(key), } } } @@ -235,6 +282,8 @@ struct PendingScopedValues { feature_flags: AHashMap>, global_state: AHashMap>, system: AHashMap>, + custom_fields: AHashMap>, + ootb_fields: AHashMap>, } impl PendingScopedValues { @@ -243,6 +292,8 @@ impl PendingScopedValues { Scope::FeatureFlagExposure => self.feature_flags.get(key), Scope::GlobalState => self.global_state.get(key), Scope::System => self.system.get(key), + Scope::CustomFields => self.custom_fields.get(key), + Scope::OotbFields => self.ootb_fields.get(key), } } @@ -257,6 +308,12 @@ impl PendingScopedValues { Scope::System => { self.system.insert(key, value); }, + Scope::CustomFields => { + self.custom_fields.insert(key, value); + }, + Scope::OotbFields => { + self.ootb_fields.insert(key, value); + }, } } } @@ -1380,7 +1437,8 @@ impl VersionedKVStore { /// /// All entries are written with the same timestamp. If a persistent journal encounters a system /// error, the store transitions to bounded in-memory mode and applies the entire batch there - /// when it fits. Capacity rejections leave the persistent store unchanged. + /// when it fits. Capacity rejections leave the persistent store unchanged, except for batches + /// containing only deletions: those also fall back so stale state cannot remain live. /// /// For persistent stores, this operation handles rotation and retries automatically if needed. /// If empty, this is a no-op that returns the current timestamp. @@ -1394,6 +1452,10 @@ impl VersionedKVStore { &mut self, entries: Vec<(Scope, String, StateValue)>, ) -> Result { + let deletion_only = entries + .iter() + .all(|(_, _, value)| value.value_type.is_none()); + if let StoreBackend::Persistent(store) = &mut self.backend { match store.extend_entries(entries.clone()).await { Ok(PersistentOperation::Persisted(timestamp)) => return Ok(timestamp), @@ -1405,6 +1467,12 @@ impl VersionedKVStore { return Ok(timestamp); }, Err(UpdateError::System(error)) => self.fallback_to_in_memory(&error), + Err(UpdateError::CapacityExceeded) if deletion_only => { + let error = anyhow::anyhow!( + "journal capacity prevented recording state deletions; retaining state in memory" + ); + self.fallback_to_in_memory(&error); + }, Err(error) => return Err(error), } } @@ -1423,8 +1491,10 @@ impl VersionedKVStore { /// Returns `None` if the key didn't exist, otherwise returns the timestamp and old value. /// /// # Errors - /// If the persistent journal encounters a system error, the store transitions to bounded - /// in-memory mode and removes the value from the live state. + /// If the persistent journal cannot record the deletion due to a system error or exhausted + /// journal capacity, the store transitions to bounded in-memory mode and removes the value + /// from live state. A removal reduces live state, so leaving a stale value visible is worse + /// than losing durability for the rest of the process. pub async fn remove( &mut self, scope: Scope, @@ -1438,7 +1508,12 @@ impl VersionedKVStore { return Ok(result); }, Err(UpdateError::System(error)) => self.fallback_to_in_memory(&error), - Err(error) => return Err(error), + Err(UpdateError::CapacityExceeded) => { + let error = anyhow::anyhow!( + "journal capacity prevented recording a state deletion; retaining state in memory" + ); + self.fallback_to_in_memory(&error); + }, } } diff --git a/bd-workflows/src/config.rs b/bd-workflows/src/config.rs index a3902addb..bd37ec753 100644 --- a/bd-workflows/src/config.rs +++ b/bd-workflows/src/config.rs @@ -8,7 +8,7 @@ use crate::workflow::Traversal; use anyhow::{anyhow, bail}; use bd_api::TriggerUploadStreaming; -use bd_log_matcher::matcher::Tree; +use bd_log_matcher::matcher::{Tree, field_value_with_state}; use bd_log_primitives::{FieldsRef, LogMessage}; use bd_proto::protos::workflow::save_field::SaveField; use bd_proto::protos::workflow::save_field::save_field::Save_field_type; @@ -1161,7 +1161,7 @@ impl TagValue { state_reader: &'a dyn bd_state::StateReader, ) -> Option> { match self { - Self::FieldExtract(field_key) => fields.field_value(field_key), + Self::FieldExtract(field_key) => field_value_with_state(fields, state_reader, field_key), Self::StateExtract(scope, key) => state_reader.get(*scope, key).and_then(|value| { if value.value_type.is_none() { Some(Cow::Borrowed("")) diff --git a/bd-workflows/src/engine_test_helpers.rs b/bd-workflows/src/engine_test_helpers.rs index 69f136035..7ad51a950 100644 --- a/bd-workflows/src/engine_test_helpers.rs +++ b/bd-workflows/src/engine_test_helpers.rs @@ -477,6 +477,11 @@ pub fn make_state_change_rule( bd_state::Scope::FeatureFlagExposure => StateScope::FEATURE_FLAG.into(), bd_state::Scope::GlobalState => StateScope::GLOBAL_STATE.into(), bd_state::Scope::System => StateScope::SYSTEM.into(), + // Virtual log fields never produce state changes. Leave the generated test config + // invalid when one is requested so normal configuration validation rejects it. + bd_state::Scope::CustomFields | bd_state::Scope::OotbFields => { + StateScope::UNSPECIFIED.into() + }, }, key: key.to_string(), previous_value: protobuf::MessageField::none(), diff --git a/bd-workflows/src/generate_log.rs b/bd-workflows/src/generate_log.rs index a236258a1..8a2c26c21 100644 --- a/bd-workflows/src/generate_log.rs +++ b/bd-workflows/src/generate_log.rs @@ -14,10 +14,12 @@ use action::ActionGenerateLog; use action::action_generate_log::ValueReference; use action::action_generate_log::generated_field::Generated_field_value_type; use action::action_generate_log::value_reference::Value_reference_type; +use bd_log_matcher::matcher::field_value_with_state; use bd_log_primitives::{DataValue, FieldsRef, Log, LogFields, log_level}; use bd_proto::protos::logging::payload::LogType; use bd_proto::protos::workflow::workflow::workflow::action; use bd_proto::protos::workflow::workflow::workflow::action::action_generate_log::ValueReferencePair; +use bd_state::StateReader; use protobuf::Enum; use std::borrow::Cow; use std::fmt::Display; @@ -49,12 +51,14 @@ fn resolve_reference<'a>( extractions: &'a TraversalExtractions, reference: &'a ValueReference, current_log_fields: FieldsRef<'a>, + state_reader: &'a dyn StateReader, ) -> Option> { match reference.value_reference_type.as_ref()? { Value_reference_type::Fixed(value) => Some(StringOrFloat::String(value.into())), - Value_reference_type::FieldFromCurrentLog(field_name) => current_log_fields - .field_value(field_name) - .map(StringOrFloat::String), + Value_reference_type::FieldFromCurrentLog(field_name) => { + field_value_with_state(current_log_fields, state_reader, field_name) + .map(StringOrFloat::String) + }, Value_reference_type::SavedFieldId(saved_field_id) => extractions .fields .get(saved_field_id) @@ -72,6 +76,7 @@ fn pair_to_floats( extractions: &TraversalExtractions, pair: &ValueReferencePair, current_log_fields: FieldsRef<'_>, + state_reader: &dyn StateReader, ) -> (f64, f64) { fn to_float(string_or_float: Option>) -> f64 { match string_or_float { @@ -85,11 +90,13 @@ fn pair_to_floats( extractions, &pair.lhs, current_log_fields, + state_reader, )); let rhs = to_float(resolve_reference( extractions, &pair.rhs, current_log_fields, + state_reader, )); (lhs, rhs) } @@ -98,28 +105,29 @@ pub fn generate_log_action( extractions: &TraversalExtractions, action: &ActionGenerateLog, current_log_fields: FieldsRef<'_>, + state_reader: &dyn StateReader, ) -> Option { let message = action.message.clone(); let mut fields = LogFields::default(); for field in &action.fields { let value = match field.generated_field_value_type.as_ref()? { Generated_field_value_type::Single(reference) => { - resolve_reference(extractions, reference, current_log_fields) + resolve_reference(extractions, reference, current_log_fields, state_reader) }, Generated_field_value_type::Subtract(pair) => { - let (lhs, rhs) = pair_to_floats(extractions, pair, current_log_fields); + let (lhs, rhs) = pair_to_floats(extractions, pair, current_log_fields, state_reader); Some(StringOrFloat::Float(lhs - rhs)) }, Generated_field_value_type::Add(pair) => { - let (lhs, rhs) = pair_to_floats(extractions, pair, current_log_fields); + let (lhs, rhs) = pair_to_floats(extractions, pair, current_log_fields, state_reader); Some(StringOrFloat::Float(lhs + rhs)) }, Generated_field_value_type::Multiply(pair) => { - let (lhs, rhs) = pair_to_floats(extractions, pair, current_log_fields); + let (lhs, rhs) = pair_to_floats(extractions, pair, current_log_fields, state_reader); Some(StringOrFloat::Float(lhs * rhs)) }, Generated_field_value_type::Divide(pair) => { - let (lhs, rhs) = pair_to_floats(extractions, pair, current_log_fields); + let (lhs, rhs) = pair_to_floats(extractions, pair, current_log_fields, state_reader); Some(StringOrFloat::Float(lhs / rhs)) }, }; diff --git a/bd-workflows/src/generate_log_test.rs b/bd-workflows/src/generate_log_test.rs index 21ea25f73..c477a2283 100644 --- a/bd-workflows/src/generate_log_test.rs +++ b/bd-workflows/src/generate_log_test.rs @@ -7,7 +7,7 @@ use crate::generate_log::generate_log_action; use crate::workflow::TraversalExtractions; -use bd_log_primitives::{FieldsRef, Log, LogFieldKey, LogFields, log_level}; +use bd_log_primitives::{DataValue, FieldsRef, Log, LogFieldKey, LogFields, log_level}; use bd_proto::protos::logging::payload::LogType; use bd_proto::protos::workflow::workflow::workflow::action::ActionGenerateLog; use bd_proto_util::serialization::TimestampMicros; @@ -25,6 +25,7 @@ struct Helper { extractions: TraversalExtractions, captured_fields: LogFields, matching_fields: LogFields, + state_reader: bd_state::InMemoryStateReader, } impl Helper { @@ -36,6 +37,7 @@ impl Helper { extractions, captured_fields, matching_fields, + state_reader: bd_state::InMemoryStateReader::default(), } } @@ -64,6 +66,7 @@ impl Helper { &self.extractions, action, FieldsRef::new(&self.captured_fields, &self.matching_fields), + &self.state_reader, ) ); } @@ -85,6 +88,18 @@ impl Helper { fn add_field(&mut self, key: LogFieldKey, value: &str) { self.captured_fields.insert(key, value.into()); } + + fn add_ootb_field(&mut self, key: &str, value: &str) { + self.state_reader.insert( + bd_state::Scope::OotbFields, + key, + bd_state::Value { + value_type: bd_state::Value_type::Data(DataValue::String(value.to_string()).into_proto()) + .into(), + ..Default::default() + }, + ); + } } #[test] @@ -203,6 +218,14 @@ fn generate_log_with_field_from_current_log() { LogType::NORMAL, &action, ); + + helper.add_ootb_field("id2", "20"); + helper.expect_log( + "hello world", + &[("add_both_bad", "NaN"), ("add_1_bad", "NaN"), ("add", "22")], + LogType::NORMAL, + &action, + ); } #[test] @@ -255,6 +278,7 @@ fn generate_log_with_uuid() { &helper.extractions, &action, FieldsRef::new(&helper.captured_fields, &helper.matching_fields), + &helper.state_reader, ) .unwrap(); diff --git a/bd-workflows/src/metrics.rs b/bd-workflows/src/metrics.rs index 5bf50842d..e6251bd69 100644 --- a/bd-workflows/src/metrics.rs +++ b/bd-workflows/src/metrics.rs @@ -12,6 +12,8 @@ mod metrics_test; use crate::config::{ActionEmitMetric, MetricMultiTag, TagValue}; use crate::engine::EmitMetricActionCount; use crate::workflow::{TriggeredActionEmitSankey, WorkflowEvent}; +use bd_log_matcher::matcher::field_value_with_state; +use bd_log_primitives::FieldsRef; use bd_state::state_value_as_cow; use bd_stats_common::{Counter, Histogram, MetricType}; use bd_workflow_stats::StatsCollector; @@ -53,9 +55,11 @@ impl MetricsCollector { #[allow(clippy::cast_precision_loss)] let maybe_value: anyhow::Result = match &action.increment { crate::config::ValueIncrement::Fixed(value) => Ok(*value as f64), - crate::config::ValueIncrement::Extract(extract) => Self::resolve_field_name(extract, event) - .ok_or_else(|| anyhow::anyhow!("field {extract:?} not found")) - .and_then(|value| value.parse::().map_err(Into::into)), + crate::config::ValueIncrement::Extract(extract) => { + Self::resolve_field_name(extract, event, state_reader) + .ok_or_else(|| anyhow::anyhow!("field {extract:?} not found")) + .and_then(|value| value.parse::().map_err(Into::into)) + }, }; let value = match maybe_value { @@ -102,6 +106,8 @@ impl MetricsCollector { value, |timestamped| ×tamped.value, ), + // Virtual log fields deliberately do not participate in state-driven workflows. + bd_state::Scope::CustomFields | bd_state::Scope::OotbFields => false, }; if matched_any { @@ -135,14 +141,24 @@ impl MetricsCollector { } } - fn resolve_field_name<'a>(key: &str, event: WorkflowEvent<'a>) -> Option> { + fn resolve_field_name<'a>( + key: &str, + event: WorkflowEvent<'a>, + state_reader: &'a dyn bd_state::StateReader, + ) -> Option> { match event { WorkflowEvent::Log(log) | WorkflowEvent::SessionStart(log) => match key { "log_level" => Some(log.log_level.to_string().into()), "log_type" => Some((log.log_type as u32).to_string().into()), - key => log.field_value(key), + key => field_value_with_state( + FieldsRef::new(&log.fields, &log.matching_fields), + state_reader, + key, + ), + }, + WorkflowEvent::StateChange(_state_change, fields) => { + field_value_with_state(fields, state_reader, key) }, - WorkflowEvent::StateChange(_state_change, fields) => fields.field_value(key), } } @@ -151,9 +167,13 @@ impl MetricsCollector { key: &str, state_reader: &'a dyn bd_state::StateReader, ) -> Option> { - state_reader - .get(scope, key) - .map(|value| state_value_as_cow(value).unwrap_or(Cow::Borrowed(""))) + state_reader.get(scope, key).and_then(|value| { + if value.value_type.is_none() { + Some(Cow::Borrowed("")) + } else { + state_value_as_cow(value) + } + }) } fn extract_tags( @@ -165,7 +185,9 @@ impl MetricsCollector { for (key, value) in tags { if let Some(extracted_value) = match value { - crate::config::TagValue::FieldExtract(extract) => Self::resolve_field_name(extract, event), + crate::config::TagValue::FieldExtract(extract) => { + Self::resolve_field_name(extract, event, state_reader) + }, crate::config::TagValue::StateExtract(scope, extract) => { Self::resolve_state_value(*scope, extract, state_reader) }, @@ -234,7 +256,9 @@ impl MetricsCollector { continue; } - let state_value = state_value_as_cow(state_value(entry)).unwrap_or(Cow::Borrowed("")); + let Some(state_value) = state_value_as_cow(state_value(entry)) else { + continue; + }; if !multi_tag.matches_value(state_value.as_ref()) { continue; } diff --git a/bd-workflows/src/metrics_test.rs b/bd-workflows/src/metrics_test.rs index ab246834f..df4fcc4ca 100644 --- a/bd-workflows/src/metrics_test.rs +++ b/bd-workflows/src/metrics_test.rs @@ -14,7 +14,7 @@ use crate::workflow::WorkflowEvent; use bd_client_stats::Stats; use bd_client_stats_store::test::StatsHelper; use bd_client_stats_store::{Collector, Counter, Histogram}; -use bd_log_primitives::{Log, LogFields, log_level}; +use bd_log_primitives::{DataValue, FieldsRef, Log, LogFields, LogMessage, log_level}; use bd_proto::protos::logging::payload::LogType; use bd_proto::protos::workflow::workflow::MultiTag as MultiTagProto; use bd_state::Scope; @@ -28,6 +28,53 @@ fn make_metrics_collector() -> (MetricsCollector, Collector) (MetricsCollector::new(stats), collector) } +#[test] +fn field_tag_value_reads_virtual_state_fields_with_log_precedence() { + let mut state_reader = bd_state::InMemoryStateReader::default(); + for (scope, key, value) in [ + (Scope::CustomFields, "custom_only", "custom"), + (Scope::CustomFields, "log_overrides_custom", "custom"), + (Scope::OotbFields, "ootb_overrides_log", "ootb"), + ] { + state_reader.insert( + scope, + key, + bd_state::Value { + value_type: bd_state::Value_type::Data(DataValue::String(value.to_string()).into_proto()) + .into(), + ..Default::default() + }, + ); + } + + let fields = [ + ( + "log_overrides_custom".into(), + DataValue::String("log".to_string()), + ), + ( + "ootb_overrides_log".into(), + DataValue::String("log".to_string()), + ), + ] + .into(); + let message = LogMessage::String("message".to_string()); + + for (field_key, expected) in [ + ("custom_only", "custom"), + ("log_overrides_custom", "log"), + ("ootb_overrides_log", "ootb"), + ] { + let tag = TagValue::FieldExtract(field_key.to_string()); + assert_eq!( + Some(expected), + tag + .extract_value(FieldsRef::new(&fields, &fields), &message, &state_reader) + .as_deref() + ); + } +} + #[test] fn metric_increment_value_extraction() { let fields = [("f1".into(), "1.1".into()), ("f2".into(), "10".into())].into(); @@ -90,6 +137,13 @@ fn metric_increment_value_extraction() { increment: crate::config::ValueIncrement::Extract("m1".to_string()), metric_type: MetricType::Histogram, }, + ActionEmitMetric { + id: "action_id_7".to_string(), + tags: BTreeMap::new(), + multi_tag: None, + increment: crate::config::ValueIncrement::Extract("state_increment".to_string()), + metric_type: MetricType::Counter, + }, ]; let action_counts: BTreeMap<&ActionEmitMetric, EmitMetricActionCount> = actions .iter() @@ -105,12 +159,19 @@ fn metric_increment_value_extraction() { }) .collect(); - metrics_collector.emit_metrics( - &action_counts, - WorkflowEvent::Log(&log), - &bd_state::InMemoryStateReader::default(), + let mut state_reader = bd_state::InMemoryStateReader::default(); + state_reader.insert( + Scope::CustomFields, + "state_increment", + bd_state::Value { + value_type: bd_state::Value_type::Data(DataValue::String("8".to_string()).into_proto()) + .into(), + ..Default::default() + }, ); + metrics_collector.emit_metrics(&action_counts, WorkflowEvent::Log(&log), &state_reader); + collector.assert_workflow_counter_eq(1, "action_id_1", labels! {}); collector.assert_workflow_counter_eq(10, "action_id_2", labels! {}); // The 1.0 is parsed as a float, then converted to an integer. @@ -127,6 +188,8 @@ fn metric_increment_value_extraction() { collector.assert_workflow_counter_eq(5, "action_id_5", labels! {}); // Values can be extracted from the matching_only_fields. collector.assert_workflow_histogram_observed(5.0, "action_id_6", labels! {}); + // Persistent custom fields can be referenced as ordinary fields. + collector.assert_workflow_counter_eq(8, "action_id_7", labels! {}); } #[test] @@ -301,6 +364,90 @@ fn counter_label_extraction() { ); } +#[test] +fn scalar_state_tags_skip_unsupported_values_but_keep_untyped_values_empty() { + let (metrics_collector, collector) = make_metrics_collector(); + let log = Log { + message: "message".into(), + session_id: "session_id".into(), + occurred_at: OffsetDateTime::now_utc(), + log_level: log_level::DEBUG, + log_type: LogType::NORMAL, + fields: LogFields::default(), + matching_fields: LogFields::default(), + capture_session: None, + }; + let mut state_reader = bd_state::InMemoryStateReader::default(); + state_reader.insert( + Scope::FeatureFlagExposure, + "binary", + bd_state::Value { + value_type: bd_state::Value_type::Data(DataValue::Bytes(vec![1, 2, 3].into()).into_proto()) + .into(), + ..Default::default() + }, + ); + state_reader.insert( + Scope::FeatureFlagExposure, + "untyped", + bd_state::Value::default(), + ); + + let unsupported_action = ActionEmitMetric { + id: "unsupported".to_string(), + tags: [( + "state".to_string(), + TagValue::StateExtract(Scope::FeatureFlagExposure, "binary".to_string()), + )] + .into(), + multi_tag: None, + increment: crate::config::ValueIncrement::Fixed(1), + metric_type: MetricType::Counter, + }; + let untyped_action = ActionEmitMetric { + id: "untyped".to_string(), + tags: [( + "state".to_string(), + TagValue::StateExtract(Scope::FeatureFlagExposure, "untyped".to_string()), + )] + .into(), + multi_tag: None, + increment: crate::config::ValueIncrement::Fixed(1), + metric_type: MetricType::Counter, + }; + let action_counts = BTreeMap::from([ + ( + &unsupported_action, + EmitMetricActionCount { + emission_count: 1, + is_parallel: false, + parallel_source_workflow_index: None, + }, + ), + ( + &untyped_action, + EmitMetricActionCount { + emission_count: 1, + is_parallel: false, + parallel_source_workflow_index: None, + }, + ), + ]); + + metrics_collector.emit_metrics(&action_counts, WorkflowEvent::Log(&log), &state_reader); + + collector.assert_workflow_counter_eq(1, "unsupported", BTreeMap::new()); + assert!( + collector + .find_counter( + &NameType::ActionId("unsupported".to_string()), + &labels! { "state" => "" }, + ) + .is_none() + ); + collector.assert_workflow_counter_eq(1, "untyped", labels! { "state" => "" }); +} + #[test] fn metric_multi_tag_fans_out_over_matching_state_entries() { let (metrics_collector, collector) = make_metrics_collector(); @@ -332,6 +479,15 @@ fn metric_multi_tag_fans_out_over_matching_state_entries() { "not_an_experiment", bd_state::string_value("ignored"), ); + state_reader.insert( + bd_state::Scope::FeatureFlagExposure, + "experiment_binary", + bd_state::Value { + value_type: bd_state::Value_type::Data(DataValue::Bytes(vec![1, 2, 3].into()).into_proto()) + .into(), + ..Default::default() + }, + ); let action = ActionEmitMetric { id: "action_id_multi".to_string(), @@ -379,6 +535,18 @@ fn metric_multi_tag_fans_out_over_matching_state_entries() { "variant" => "control", }, ); + assert!( + collector + .find_counter( + &NameType::ActionId("action_id_multi".to_string()), + &labels! { + "static" => "tag", + "experiment" => "experiment_binary", + "variant" => "", + }, + ) + .is_none() + ); } #[test] diff --git a/bd-workflows/src/workflow.rs b/bd-workflows/src/workflow.rs index e9116a72a..29d1df900 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 bd_log_matcher::matcher::MatchContext; +use bd_log_matcher::matcher::{MatchContext, field_value_with_state}; use bd_log_primitives::tiny_set::{TinyMap, TinySet}; use bd_log_primitives::{FieldsRef, Log, log_level}; use bd_proto::protos::logging::payload::LogType; @@ -1251,6 +1251,7 @@ fn process_transition<'a>( mut extractions: TraversalExtractions, actions: &'a [Action], fields: FieldsRef<'_>, + state_reader: &dyn bd_state::StateReader, current_state_index: usize, next_state_index: usize, transition_type: WorkflowDebugTransitionType, @@ -1261,7 +1262,7 @@ fn process_transition<'a>( // Collect triggered actions and injected logs. let (triggered_actions, logs_to_inject) = - Traversal::triggered_actions(actions, &mut extractions, fields); + Traversal::triggered_actions(actions, &mut extractions, fields, state_reader); result.triggered_actions.extend(triggered_actions); result.log_to_inject.extend(logs_to_inject); @@ -1467,6 +1468,7 @@ impl Traversal { TraversalExtractions::default(), actions, fields, + state_reader, self.state_index, next_state_index, WorkflowDebugTransitionType::Timeout, @@ -1532,6 +1534,7 @@ impl Traversal { self.do_log_extractions(config, index, log, state_reader), actions, FieldsRef::new(&log.fields, &log.matching_fields), + state_reader, self.state_index, next_state_index, WorkflowDebugTransitionType::Normal(index as u64), @@ -1634,6 +1637,7 @@ impl Traversal { self.do_state_change_extractions(config, index, state_change, fields, state_reader), actions, fields, + state_reader, self.state_index, next_state_index, WorkflowDebugTransitionType::Normal(index as u64), @@ -1666,6 +1670,7 @@ impl Traversal { self.do_log_extractions(config, index, log, state_reader), actions, FieldsRef::new(&log.fields, &log.matching_fields), + state_reader, self.state_index, next_state_index, WorkflowDebugTransitionType::Normal(index as u64), @@ -1728,9 +1733,12 @@ impl Traversal { for extraction in &extractions.field_extractions { let extracted_value = match &extraction.source { - FieldExtractionSource::FieldName(field_name) => log - .field_value(field_name) - .and_then(|value| extraction.extract_value(value.as_ref())), + FieldExtractionSource::FieldName(field_name) => field_value_with_state( + FieldsRef::new(&log.fields, &log.matching_fields), + state_reader, + field_name, + ) + .and_then(|value| extraction.extract_value(value.as_ref())), FieldExtractionSource::Message => log .message .as_str() @@ -1805,9 +1813,10 @@ impl Traversal { // app_version, etc.) for extraction in &extractions.field_extractions { let extracted_value = match &extraction.source { - FieldExtractionSource::FieldName(field_name) => fields - .field_value(field_name) - .and_then(|value| extraction.extract_value(value.as_ref())), + FieldExtractionSource::FieldName(field_name) => { + field_value_with_state(fields, state_reader, field_name) + .and_then(|value| extraction.extract_value(value.as_ref())) + }, // State changes don't include log messages. FieldExtractionSource::Message => None, }; @@ -1829,6 +1838,7 @@ impl Traversal { actions: &'a [Action], extractions: &mut TraversalExtractions, current_log_fields: FieldsRef<'_>, + state_reader: &dyn bd_state::StateReader, ) -> (Vec>, TinyMap<&'a str, Log>) { let mut triggered_actions = vec![]; let mut logs_to_inject = TinyMap::default(); @@ -1862,7 +1872,9 @@ impl Traversal { triggered_actions.push(TriggeredAction::StartTracing); }, Action::GenerateLog(action) => { - if let Some(log) = generate_log_action(extractions, action, current_log_fields) { + if let Some(log) = + generate_log_action(extractions, action, current_log_fields, state_reader) + { logs_to_inject.insert(action.id.as_str(), log); } }, diff --git a/fuzz/src/versioned_kv_journal.rs b/fuzz/src/versioned_kv_journal.rs index 85a7c0f65..0319a1a1c 100644 --- a/fuzz/src/versioned_kv_journal.rs +++ b/fuzz/src/versioned_kv_journal.rs @@ -7,8 +7,8 @@ use ahash::AHashMap; use arbitrary::{Arbitrary, Unstructured}; -use bd_proto::protos::state::payload::StateValue; -use bd_proto::protos::state::payload::state_value::Value_type; +use bd_proto::protos::state::state_payload::StateValue; +use bd_proto::protos::state::state_payload::state_value::Value_type; use bd_time::{TestTimeProvider, TimeProvider as _}; use bd_versioned_kv::{ DataLoss,