From 7e97530f50a822513f6543aaf3fb5b6d32596881 Mon Sep 17 00:00:00 2001 From: SinuS Von SifriduS Date: Sat, 25 Jul 2026 10:35:16 +0200 Subject: [PATCH] v0.1.0-pre.030 --- CHANGELOG.md | 7 + README.md | 4 + ROADMAP.md | 2 + kb-lib/src/decoder.rs | 4 +- kb-lib/src/decoder/api.rs | 4 +- kb-lib/src/decoder/api/contracts.rs | 12 +- kb-lib/src/lib.rs | 4 +- kb-pipeline/Cargo.toml | 3 +- kb-pipeline/src/decode_replay.rs | 2245 +++++++++++++++++++++++++++ kb-pipeline/src/lib.rs | 27 +- 10 files changed, 2298 insertions(+), 14 deletions(-) create mode 100644 kb-pipeline/src/decode_replay.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 1075af6..a661b21 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,10 @@ + +## 0.1.0-pre.030 + +- Migration du replay contextuel de décodage dans `kb-pipeline`. +- Conservation de la sélection bornée, du dispatch déterministe, du ledger, du force replay et de la matérialisation optionnelle. +- Adaptation aux façades consolidées `kb-lib` et `kb-store`. +- Conservation des 19 tests de l’ancienne implémentation bot2. diff --git a/README.md b/README.md index 64a3e3c..88f7f53 100644 --- a/README.md +++ b/README.md @@ -94,3 +94,7 @@ Les modules de matérialisation sont privés. Tous les contrats, types concrets ### Migration pipeline La tranche `0.1.0-pre.029` migre la planification et l’extraction core de `kb-pipeline`. + +### Pipeline migré + +`kb-pipeline` contient désormais l’extraction core et le replay contextuel de décodage avec matérialisation optionnelle. diff --git a/ROADMAP.md b/ROADMAP.md index 80682de..01bd857 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -284,3 +284,5 @@ - [x] Extraction canonique vers les tables core. - [ ] Backfill et replay de décodage. - [ ] Orchestrations Solana stateful et exécution. + +- [x] Migrer `decode_replay` dans `kb-pipeline`. diff --git a/kb-lib/src/decoder.rs b/kb-lib/src/decoder.rs index 8d262aa..4c6a6ae 100644 --- a/kb-lib/src/decoder.rs +++ b/kb-lib/src/decoder.rs @@ -1,5 +1,5 @@ // file: kb-lib/src/decoder.rs -// version: 9 +// version: 10 //! Consolidated decoder modules. @@ -126,7 +126,7 @@ pub use self::api::DcApiInstructionDecoder; /// Exposes the common protocol decoder trait. pub use self::api::DcApiProtocolDecoder; /// Computes the deterministic contextual input hash. -pub use self::api::decoder_api_decoder_api_contextual_input_hash; +pub use self::api::decoder_api_contextual_input_hash; /// Returns true when a decoder declares the exact program id. pub use self::api::decoder_api_decoder_handles_program_id; /// Computes a deterministic lowercase SHA-256 hash for one JSON value. diff --git a/kb-lib/src/decoder/api.rs b/kb-lib/src/decoder/api.rs index 5553a24..ba5d582 100644 --- a/kb-lib/src/decoder/api.rs +++ b/kb-lib/src/decoder/api.rs @@ -1,5 +1,5 @@ // file: kb-lib/src/decoder/api.rs -// version: 4 +// version: 5 //! Decoder contracts consolidated from `kb_decoder_api`. @@ -33,7 +33,7 @@ pub use self::contracts::DcApiDecoderSurface; /// Contextual instruction decoder contract. pub use self::contracts::DcApiInstructionDecoder; /// Computes the deterministic contextual input hash. -pub use self::contracts::decoder_api_decoder_api_contextual_input_hash; +pub use self::contracts::decoder_api_contextual_input_hash; /// Returns true when a decoder declares the exact program id. pub use self::contracts::decoder_api_decoder_handles_program_id; /// Computes a deterministic lowercase SHA-256 hash for one JSON value. diff --git a/kb-lib/src/decoder/api/contracts.rs b/kb-lib/src/decoder/api/contracts.rs index 7e31de3..ce5aa20 100644 --- a/kb-lib/src/decoder/api/contracts.rs +++ b/kb-lib/src/decoder/api/contracts.rs @@ -1,5 +1,5 @@ // file: kb-lib/src/decoder/api/contracts.rs -// version: 7 +// version: 8 //! Backend-neutral contextual instruction decoder contracts. @@ -383,7 +383,7 @@ pub fn decoder_api_decoder_handles_program_id( } /// Computes a deterministic lowercase SHA-256 hash for one contextual input. -pub fn decoder_api_decoder_api_contextual_input_hash( +pub fn decoder_api_contextual_input_hash( input: &crate::MdCoreInstructionReplayInput, ) -> kb_core::Result { let serialization_result = serde_json::to_value(input); @@ -501,8 +501,8 @@ mod tests { #[test] fn contextual_input_hash_is_stable() { - let first = crate::decoder_api_decoder_api_contextual_input_hash(&replay_input()); - let second = crate::decoder_api_decoder_api_contextual_input_hash(&replay_input()); + let first = crate::decoder_api_contextual_input_hash(&replay_input()); + let second = crate::decoder_api_contextual_input_hash(&replay_input()); assert!(first.is_ok()); assert_eq!(first, second); } @@ -518,8 +518,8 @@ mod tests { "payloadJson": {"dataBase64": "CAcGBQQDAgE="}, "payloadHash": "different-payload-hash" }]); - let first_hash = crate::decoder_api_decoder_api_contextual_input_hash(&first); - let second_hash = crate::decoder_api_decoder_api_contextual_input_hash(&second); + let first_hash = crate::decoder_api_contextual_input_hash(&first); + let second_hash = crate::decoder_api_contextual_input_hash(&second); assert!(first_hash.is_ok()); assert!(second_hash.is_ok()); assert_ne!(first_hash, second_hash); diff --git a/kb-lib/src/lib.rs b/kb-lib/src/lib.rs index 73ab7af..bf95d01 100644 --- a/kb-lib/src/lib.rs +++ b/kb-lib/src/lib.rs @@ -1,5 +1,5 @@ // file: kb-lib/src/lib.rs -// version: 23 +// version: 24 //! Consolidated decoder, executor, materializer and shared model library. #![warn(missing_docs)] @@ -312,7 +312,7 @@ pub use self::decoder::DcWalletJupiterApeproSmartWalletDecoder; /// Reserved `kb_decoder_weighted_swap_stabble` decoder. pub use self::decoder::DcWeightedSwapStabbleDecoder; /// Computes the deterministic contextual input hash. -pub use self::decoder::decoder_api_decoder_api_contextual_input_hash; +pub use self::decoder::decoder_api_contextual_input_hash; /// Returns true when a decoder declares the exact program id. pub use self::decoder::decoder_api_decoder_handles_program_id; /// Computes a deterministic lowercase SHA-256 hash for one JSON value. diff --git a/kb-pipeline/Cargo.toml b/kb-pipeline/Cargo.toml index f314820..93c6728 100644 --- a/kb-pipeline/Cargo.toml +++ b/kb-pipeline/Cargo.toml @@ -1,5 +1,5 @@ # file: kb-pipeline/Cargo.toml -# version: 4 +# version: 5 [package] name = "kb-pipeline" @@ -15,6 +15,7 @@ futures-util.workspace = true kb-core = { path = "../kb-core" } kb-lib = { path = "../kb-lib" } kb-onchain-transport = { path = "../kb-onchain-transport" } +kb-program-ids = { path = "../kb-program-ids" } kb-store = { path = "../kb-store" } kb-wallet = { path = "../kb-wallet" } serde_json.workspace = true diff --git a/kb-pipeline/src/decode_replay.rs b/kb-pipeline/src/decode_replay.rs new file mode 100644 index 0000000..1a11aac --- /dev/null +++ b/kb-pipeline/src/decode_replay.rs @@ -0,0 +1,2245 @@ +// file: kb-pipeline/src/decode_replay.rs +// version: 1 + +//! Common contextual instruction decode and optional materialization pipeline. + +use futures_util::StreamExt; // rust-rules: trait-import +use tracing::Instrument; // rust-rules: trait-import + +static DECODE_CAMPAIGN_SEQUENCE: std::sync::atomic::AtomicU64 = + std::sync::atomic::AtomicU64::new(1); + +/// Stable processing ledger stage used by contextual instruction decoders. +pub const INSTRUCTION_DECODE_STAGE: &str = "instruction_decode"; +/// Stable processing ledger stage used by decoded event materializers. +pub const EVENT_MATERIALIZATION_STAGE: &str = "event_materialization"; +/// Current common decode pipeline orchestration version. +pub const DECODE_PIPELINE_VERSION: &str = "1"; + +/// Deterministic policy used when several decoders recognize one input. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum DecodeDispatchPolicy { + /// Run only the highest-ranked compatible decoder. + HighestPriority, + /// Run every compatible decoder in deterministic rank order. + AllCompatible, +} + +/// Complete bounded contextual decode replay request. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct DecodeReplayRequest { + /// Stable caller-provided campaign identifier used by structured tracing. + pub campaign_id: std::string::String, + /// Bounded backend-neutral input selection. + pub selection: kb_store::DecodeSelectionFilter, + /// Explicit decoder names to enable, or every supplied decoder when empty. + pub decoder_names: std::vec::Vec, + /// Deterministic multi-decoder dispatch policy. + pub dispatch_policy: crate::DecodeDispatchPolicy, + /// Maximum concurrent contextual input executions. + pub max_concurrent_inputs: u32, + /// Replaces processor-owned outputs for the selected version and input. + pub force_replay: bool, + /// Explicitly authorizes a bounded force replay without exact signatures. + pub force_replay_all_matching: bool, + /// Runs compatible materializers after decoded observations are committed. + pub materialize_after_decode: bool, +} + +impl DecodeReplayRequest { + /// Validates campaign bounds and explicit decoder names. + pub fn validate(&self) -> kb_core::Result<()> { + if self.campaign_id.trim().is_empty() { + return std::result::Result::Err(kb_core::Error::config( + "decode replay campaign id must not be empty", + )); + } + if self.max_concurrent_inputs == 0 { + return std::result::Result::Err(kb_core::Error::config( + "decode replay concurrency must be greater than zero", + )); + } + if self.force_replay + && self.selection.signatures.is_empty() + && !self.force_replay_all_matching + { + return std::result::Result::Err(kb_core::Error::config( + "force replay requires exact signatures or explicit all-matching authorization", + )); + } + if self.force_replay_all_matching && !self.force_replay { + return std::result::Result::Err(kb_core::Error::config( + "all-matching replay authorization requires force replay", + )); + } + if self.force_replay_all_matching && !self.selection.signatures.is_empty() { + return std::result::Result::Err(kb_core::Error::config( + "all-matching replay authorization cannot be combined with exact signatures", + )); + } + if self.decoder_names.iter().any(|name| return name.trim().is_empty()) { + return std::result::Result::Err(kb_core::Error::config( + "decode replay decoder names must not be empty", + )); + } + return std::result::Result::Ok(()); + } +} + +/// Decode replay progress severity. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum DecodeReplayProgressLevel { + /// Diagnostic detail. + Debug, + /// Normal campaign information. + Info, + /// Recoverable issue or cancellation. + Warning, + /// Decode or persistence failure. + Error, +} + +impl DecodeReplayProgressLevel { + /// Returns the stable lowercase level code. + pub fn code(&self) -> &'static str { + return match self { + crate::DecodeReplayProgressLevel::Debug => "debug", + crate::DecodeReplayProgressLevel::Info => "info", + crate::DecodeReplayProgressLevel::Warning => "warn", + crate::DecodeReplayProgressLevel::Error => "error", + }; + } +} + +/// One operator-visible decode replay progress event. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct DecodeReplayProgressEvent { + /// UTC timestamp rendered in RFC 3339. + pub timestamp: std::string::String, + /// Severity. + pub level: crate::DecodeReplayProgressLevel, + /// Human-readable message. + pub message: std::string::String, + /// Number of terminal contextual inputs. + pub completed: u64, + /// Total selected contextual inputs. + pub total: u64, +} + +impl DecodeReplayProgressEvent { + fn new( + level: crate::DecodeReplayProgressLevel, + message: impl std::convert::Into, + completed: u64, + total: u64, + ) -> Self { + return Self { + timestamp: chrono::Utc::now().to_rfc3339(), + level, + message: message.into(), + completed, + total, + }; + } +} + +/// Progress and cooperative cancellation contract implemented by applications. +pub trait DecodeReplayObserver: Sync { + /// Receives one progress event. + fn on_progress(&self, event: &crate::DecodeReplayProgressEvent); + /// Returns true when cooperative cancellation was requested. + fn is_cancelled(&self) -> bool; +} + +/// Aggregated terminal counters for one decoder version. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct DecodeProcessorSummary { + /// Stable decoder name. + pub processor_name: std::string::String, + /// Stable decoder version. + pub processor_version: std::string::String, + /// Number of compatible dispatches. + pub dispatched: u64, + /// Number skipped because version and contextual input hash already succeeded. + pub skipped: u64, + /// Number producing decoded observations. + pub decoded: u64, + /// Number intentionally ignored. + pub ignored: u64, + /// Number recognized but unsupported. + pub unsupported: u64, + /// Number failing decode, validation or persistence. + pub failed: u64, + /// Number of materialized outputs committed. + pub materialized_outputs: u64, + /// Number of materializer policy refusals committed to the ledger. + pub materialization_refused: u64, +} + +/// Final counters for one bounded contextual decode replay campaign. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct DecodeReplaySummary { + /// Stable process-local campaign identifier used by structured tracing. + pub campaign_id: std::string::String, + /// Common orchestration implementation version. + pub pipeline_version: std::string::String, + /// Number of selected contextual inputs. + pub selected: u64, + /// Number admitted to execution. + pub started: u64, + /// Number reaching a terminal input result. + pub completed: u64, + /// Number with no compatible enabled decoder. + pub unmatched: u64, + /// Number never decoded because cancellation was already requested. + pub not_started: u64, + /// Number of input-level failures. + pub failed_inputs: u64, + /// Whether cooperative cancellation was observed. + pub cancelled: bool, + /// Per-decoder counters ordered by name and version. + pub processors: std::vec::Vec, + /// Campaign start timestamp. + pub started_at: std::string::String, + /// Campaign finish timestamp. + pub finished_at: std::string::String, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +struct DecodeItemOutcome { + signature: std::string::String, + slot: u64, + instruction_path: std::string::String, + program_id: std::string::String, + started: bool, + unmatched: bool, + failed: bool, + cancelled: bool, + processors: std::vec::Vec, +} + +struct RankedDecoder<'decoder> { + decoder: &'decoder dyn kb_lib::DcApiInstructionDecoder, + identity: kb_lib::DcApiDecoderIdentity, + recognition: kb_lib::DcApiDecoderRecognition, +} + +/// Executes one bounded contextual instruction decode and optional materialization campaign. +pub async fn execute_decode_replay( + store: &S, + request: &crate::DecodeReplayRequest, + decoders: &[std::sync::Arc], + materializers: &[std::sync::Arc], + observer: &O, +) -> kb_core::Result +where + S: kb_store::DecodePipelineStore + Sync, + O: crate::DecodeReplayObserver, +{ + let campaign_id = request.campaign_id.clone(); + let campaign_span = tracing::debug_span!( + target: crate::TRACING_TARGET, + "decode_replay_campaign", + campaign_id = %campaign_id, + pipeline_version = crate::DECODE_PIPELINE_VERSION + ); + return execute_decode_replay_campaign( + store, + request, + decoders, + materializers, + observer, + campaign_id, + ) + .instrument(campaign_span) + .await; +} + +async fn execute_decode_replay_campaign( + store: &S, + request: &crate::DecodeReplayRequest, + decoders: &[std::sync::Arc], + materializers: &[std::sync::Arc], + observer: &O, + campaign_id: std::string::String, +) -> kb_core::Result +where + S: kb_store::DecodePipelineStore + Sync, + O: crate::DecodeReplayObserver, +{ + let signature_sample = text_sample(request.selection.signatures.as_slice(), 5); + tracing::debug!( + target: crate::TRACING_TARGET, + action = "execute_campaign", + campaign_id = %campaign_id, + signature_count = request.selection.signatures.len(), + signature_sample = ?signature_sample, + processing_states = ?request.selection.processing_states, + min_slot = ?request.selection.min_slot, + max_slot = ?request.selection.max_slot, + program_ids = ?request.selection.program_ids, + instruction_paths = ?request.selection.instruction_paths, + limit = request.selection.limit, + decoder_names = ?request.decoder_names, + dispatch_policy = ?request.dispatch_policy, + max_concurrent_inputs = request.max_concurrent_inputs, + force_replay = request.force_replay, + force_replay_all_matching = request.force_replay_all_matching, + materialize_after_decode = request.materialize_after_decode, + supplied_decoder_count = decoders.len(), + supplied_materializer_count = materializers.len(), + "received contextual decode replay campaign" + ); + let validation_result = request.validate(); + if let std::result::Result::Err(error) = validation_result { + return std::result::Result::Err(error); + } + let enabled = enabled_decoders(request, decoders); + if enabled.is_empty() { + tracing::error!(target: crate::TRACING_TARGET, action = "resolve_decoders", campaign_id = %campaign_id, requested_decoder_names = ?request.decoder_names, supplied_decoder_count = decoders.len(), "no contextual decoder is enabled"); + return std::result::Result::Err(kb_core::Error::config( + "decode replay requires at least one enabled decoder", + )); + } + let enabled_decoder_identities = enabled + .iter() + .map(|decoder| { + let identity = decoder.identity(); + return format!("{}@{}", identity.name, identity.version); + }) + .collect::>(); + tracing::debug!(target: crate::TRACING_TARGET, action = "resolve_decoders", campaign_id = %campaign_id, enabled_decoders = ?enabled_decoder_identities, "resolved contextual decode replay decoders"); + let decoder_registry_result = validate_decoder_registry(&enabled); + if let std::result::Result::Err(error) = decoder_registry_result { + return std::result::Result::Err(error); + } + let effective_selection_result = effective_selection(request, enabled.as_slice()); + let effective_selection = match effective_selection_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + tracing::debug!( + target: crate::TRACING_TARGET, + action = "resolve_selection", + campaign_id = %campaign_id, + requested_program_ids = ?request.selection.program_ids, + effective_program_ids = ?effective_selection.program_ids, + requested_processing_states = ?request.selection.processing_states, + effective_processing_states = ?effective_selection.processing_states, + force_replay = request.force_replay, + explicit_signature_selection = !effective_selection.signatures.is_empty(), + processing_state_filter_bypassed = request.force_replay + && (!effective_selection.signatures.is_empty() || request.force_replay_all_matching), + force_replay_all_matching = request.force_replay_all_matching, + "resolved contextual decode replay selection" + ); + let ordered_materializers_result = ordered_materializers(materializers); + let ordered_materializers = match ordered_materializers_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let ordered_materializer_identities = ordered_materializers + .iter() + .map(|materializer| { + let identity = materializer.identity(); + return format!("{}@{}", identity.name, identity.version); + }) + .collect::>(); + if request.materialize_after_decode && ordered_materializers.is_empty() { + tracing::error!(target: crate::TRACING_TARGET, action = "resolve_materializers", campaign_id = %campaign_id, requested = true, available_count = 0_usize, "materialization requested without available materializer"); + return std::result::Result::Err(kb_core::Error::config( + "decode replay materialization requires at least one available materializer", + )); + } + tracing::debug!(target: crate::TRACING_TARGET, action = "resolve_materializers", campaign_id = %campaign_id, materializers = ?ordered_materializer_identities, "resolved contextual decode replay materializers"); + for decoder in &enabled { + let identity = decoder.identity(); + let declarations = coverage_declaration_inserts(decoder.as_ref()); + tracing::debug!(target: crate::TRACING_TARGET, action = "persist_coverage_declarations", campaign_id = %campaign_id, processor_name = %identity.name, processor_version = %identity.version, declaration_count = declarations.len(), "persist decoder coverage declarations"); + let persist_result = store.persist_decode_coverage_declarations(&declarations).await; + match persist_result { + std::result::Result::Ok(outcome) => { + tracing::debug!(target: crate::TRACING_TARGET, action = "persist_coverage_declarations", campaign_id = %campaign_id, processor_name = %identity.name, processor_version = %identity.version, outcome = ?outcome, "decoder coverage declarations persisted"); + }, + std::result::Result::Err(error) => { + tracing::error!(target: crate::TRACING_TARGET, action = "persist_coverage_declarations", campaign_id = %campaign_id, processor_name = %identity.name, processor_version = %identity.version, error = %error, "decoder coverage declaration persistence failed"); + return std::result::Result::Err(error); + }, + } + } + tracing::debug!(target: crate::TRACING_TARGET, action = "select_inputs", campaign_id = %campaign_id, signature_count = effective_selection.signatures.len(), processing_states = ?effective_selection.processing_states, min_slot = ?effective_selection.min_slot, max_slot = ?effective_selection.max_slot, program_ids = ?effective_selection.program_ids, instruction_paths = ?effective_selection.instruction_paths, limit = effective_selection.limit, "select contextual decode replay inputs"); + let inputs_result = store.list_decode_inputs(&effective_selection).await; + let inputs = match inputs_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + tracing::error!(target: crate::TRACING_TARGET, action = "select_inputs", campaign_id = %campaign_id, error = %error, "contextual decode replay input selection failed"); + return std::result::Result::Err(error); + }, + }; + tracing::debug!(target: crate::TRACING_TARGET, action = "select_inputs", campaign_id = %campaign_id, selected_count = inputs.len(), "selected contextual decode replay inputs"); + for input in &inputs { + tracing::debug!( + target: crate::TRACING_TARGET, + action = "selected_input", + campaign_id = %campaign_id, + signature = %input.signature, + slot = input.slot, + instruction_path = %input.instruction_path, + program_id = %input.program_id, + input_key = %input.replay_input_key, + transaction_failed = input.transaction_failed, + surface_code_hint = ?input.surface_code_hint, + payload_hash = ?input.instruction_payload_hash, + "selected contextual decode replay input" + ); + } + let started_at = chrono::Utc::now().to_rfc3339(); + let selected_result = u64::try_from(inputs.len()); + let selected = match selected_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + return std::result::Result::Err(kb_core::Error::invalid_state(format!( + "decode selected count conversion failed: {error}" + ))); + }, + }; + observer.on_progress(&crate::DecodeReplayProgressEvent::new( + crate::DecodeReplayProgressLevel::Info, + format!( + "decode replay campaign {campaign_id} selected {selected} contextual inputs for {} decoders", + enabled.len() + ), + 0, + selected, + )); + let concurrency_result = usize::try_from(request.max_concurrent_inputs); + let concurrency = match concurrency_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + return std::result::Result::Err(kb_core::Error::config(format!( + "decode concurrency conversion failed: {error}" + ))); + }, + }; + let enabled_ref = enabled.as_slice(); + let materializers_ref = ordered_materializers.as_slice(); + let campaign_id_ref = campaign_id.as_str(); + let stream = futures_util::stream::iter(inputs.into_iter().map(|input| { + return async move { + let input_span = tracing::debug_span!( + target: crate::TRACING_TARGET, + "decode_replay_input", + campaign_id = %campaign_id_ref, + signature = %input.signature, + slot = input.slot, + instruction_path = %input.instruction_path, + program_id = %input.program_id, + input_key = %input.replay_input_key + ); + return execute_decode_input( + store, + request, + enabled_ref, + materializers_ref, + observer, + campaign_id_ref, + &input, + ) + .instrument(input_span) + .await; + }; + })) + .buffer_unordered(concurrency); + futures_util::pin_mut!(stream); + let mut outcomes = std::vec::Vec::new(); + let mut settled = 0_u64; + while let std::option::Option::Some(outcome) = stream.next().await { + settled += 1; + let level = if outcome.failed { + crate::DecodeReplayProgressLevel::Error + } else if outcome.cancelled { + crate::DecodeReplayProgressLevel::Warning + } else { + crate::DecodeReplayProgressLevel::Debug + }; + observer.on_progress(&crate::DecodeReplayProgressEvent::new( + level, + format!( + "decode replay input terminal signature={} slot={} instruction_path={} program_id={} unmatched={} failed={} cancelled={} processors={:?}", + outcome.signature, + outcome.slot, + outcome.instruction_path, + outcome.program_id, + outcome.unmatched, + outcome.failed, + outcome.cancelled, + outcome.processors + ), + settled, + selected, + )); + outcomes.push(outcome); + } + let mut summary = crate::DecodeReplaySummary { + campaign_id: campaign_id.clone(), + pipeline_version: crate::DECODE_PIPELINE_VERSION.to_string(), + selected, + started: 0, + completed: 0, + unmatched: 0, + not_started: 0, + failed_inputs: 0, + cancelled: observer.is_cancelled(), + processors: std::vec::Vec::new(), + started_at, + finished_at: chrono::Utc::now().to_rfc3339(), + }; + for outcome in outcomes { + if outcome.started { + summary.started += 1; + summary.completed += 1; + } + if outcome.unmatched { + summary.unmatched += 1; + } + if outcome.failed { + summary.failed_inputs += 1; + } + if outcome.cancelled && !outcome.started { + summary.not_started += 1; + } + for processor in outcome.processors { + merge_processor_summary(&mut summary.processors, processor); + } + } + summary.processors.sort_by(|left, right| { + return left + .processor_name + .cmp(&right.processor_name) + .then(left.processor_version.cmp(&right.processor_version)); + }); + if summary.unmatched > 0 || summary.failed_inputs > 0 { + tracing::error!(target: crate::TRACING_TARGET, action = "execute_campaign", campaign_id = %summary.campaign_id, selected = summary.selected, started = summary.started, completed = summary.completed, unmatched = summary.unmatched, not_started = summary.not_started, failed_inputs = summary.failed_inputs, cancelled = summary.cancelled, processors = ?summary.processors, started_at = %summary.started_at, finished_at = %summary.finished_at, "contextual decode replay completed with failures"); + } else if summary.cancelled { + tracing::warn!(target: crate::TRACING_TARGET, action = "execute_campaign", campaign_id = %summary.campaign_id, selected = summary.selected, started = summary.started, completed = summary.completed, unmatched = summary.unmatched, not_started = summary.not_started, failed_inputs = summary.failed_inputs, cancelled = summary.cancelled, processors = ?summary.processors, started_at = %summary.started_at, finished_at = %summary.finished_at, "contextual decode replay cancelled"); + } else { + tracing::info!(target: crate::TRACING_TARGET, action = "execute_campaign", campaign_id = %summary.campaign_id, selected = summary.selected, started = summary.started, completed = summary.completed, unmatched = summary.unmatched, not_started = summary.not_started, failed_inputs = summary.failed_inputs, cancelled = summary.cancelled, processors = ?summary.processors, started_at = %summary.started_at, finished_at = %summary.finished_at, "contextual decode replay completed"); + } + return std::result::Result::Ok(summary); +} + +async fn execute_decode_input( + store: &S, + request: &crate::DecodeReplayRequest, + decoders: &[std::sync::Arc], + materializers: &[std::sync::Arc], + observer: &O, + campaign_id: &str, + input: &kb_lib::MdCoreInstructionReplayInput, +) -> DecodeItemOutcome +where + S: kb_store::DecodePipelineStore + Sync, + O: crate::DecodeReplayObserver, +{ + tracing::debug!(target: crate::TRACING_TARGET, action = "admit_input", campaign_id = %campaign_id, signature = %input.signature, slot = input.slot, instruction_path = %input.instruction_path, program_id = %input.program_id, input_key = %input.replay_input_key, transaction_failed = input.transaction_failed, "consider contextual decode replay input"); + if observer.is_cancelled() { + tracing::debug!(target: crate::TRACING_TARGET, action = "admit_input", campaign_id = %campaign_id, signature = %input.signature, instruction_path = %input.instruction_path, admitted = false, reason = "cancelled_before_admission", "contextual decode replay input not admitted"); + return DecodeItemOutcome { + signature: input.signature.clone(), + slot: input.slot, + instruction_path: input.instruction_path.clone(), + program_id: input.program_id.clone(), + started: false, + unmatched: false, + failed: false, + cancelled: true, + processors: std::vec::Vec::new(), + }; + } + let ranked = rank_decoders(campaign_id, input, decoders); + if ranked.is_empty() { + tracing::error!(target: crate::TRACING_TARGET, action = "dispatch_input", campaign_id = %campaign_id, signature = %input.signature, slot = input.slot, instruction_path = %input.instruction_path, program_id = %input.program_id, matched_decoder_count = 0_usize, error_code = "decode_input_unmatched", "no contextual decoder matched selected input"); + return DecodeItemOutcome { + signature: input.signature.clone(), + slot: input.slot, + instruction_path: input.instruction_path.clone(), + program_id: input.program_id.clone(), + started: true, + unmatched: true, + failed: false, + cancelled: false, + processors: std::vec::Vec::new(), + }; + } + let selected = match request.dispatch_policy { + crate::DecodeDispatchPolicy::HighestPriority => ranked.into_iter().take(1).collect(), + crate::DecodeDispatchPolicy::AllCompatible => ranked, + }; + let selected_decoder_identities = selected + .iter() + .map(|ranked_decoder: &RankedDecoder<'_>| { + return format!("{}@{}", ranked_decoder.identity.name, ranked_decoder.identity.version); + }) + .collect::>(); + tracing::debug!(target: crate::TRACING_TARGET, action = "dispatch_input", campaign_id = %campaign_id, signature = %input.signature, slot = input.slot, instruction_path = %input.instruction_path, program_id = %input.program_id, dispatch_policy = ?request.dispatch_policy, selected_decoders = ?selected_decoder_identities, "dispatch contextual decode replay input"); + let mut processors = std::vec::Vec::new(); + let mut failed = false; + for ranked_decoder in selected { + if observer.is_cancelled() { + break; + } + tracing::debug!(target: crate::TRACING_TARGET, action = "invoke_decoder", campaign_id = %campaign_id, signature = %input.signature, instruction_path = %input.instruction_path, program_id = %input.program_id, processor_name = %ranked_decoder.identity.name, processor_version = %ranked_decoder.identity.version, recognition = ?ranked_decoder.recognition, "invoke contextual instruction decoder"); + let processor_span = tracing::debug_span!( + target: crate::TRACING_TARGET, + "decode_replay_processor", + campaign_id = %campaign_id, + signature = %input.signature, + instruction_path = %input.instruction_path, + program_id = %input.program_id, + processor_name = %ranked_decoder.identity.name, + processor_version = %ranked_decoder.identity.version + ); + let processor_result = + execute_decoder(store, request, campaign_id, &ranked_decoder, materializers, input) + .instrument(processor_span) + .await; + match processor_result { + std::result::Result::Ok(value) => { + tracing::debug!(target: crate::TRACING_TARGET, action = "invoke_decoder", campaign_id = %campaign_id, signature = %input.signature, instruction_path = %input.instruction_path, program_id = %input.program_id, processor_name = %value.processor_name, processor_version = %value.processor_version, summary = ?value, "contextual instruction decoder completed"); + if value.failed > 0 { + failed = true; + } + processors.push(value); + }, + std::result::Result::Err(error) => { + failed = true; + processors.push(crate::DecodeProcessorSummary { + processor_name: ranked_decoder.identity.name.clone(), + processor_version: ranked_decoder.identity.version.clone(), + dispatched: 1, + skipped: 0, + decoded: 0, + ignored: 0, + unsupported: 0, + failed: 1, + materialized_outputs: 0, + materialization_refused: 0, + }); + tracing::error!( + target: crate::TRACING_TARGET, + action = "invoke_decoder", campaign_id = %campaign_id, + signature = %input.signature, + slot = input.slot, + instruction_path = %input.instruction_path, + program_id = %input.program_id, + processor_name = %ranked_decoder.identity.name, + processor_version = %ranked_decoder.identity.version, + input_key = %input.replay_input_key, + error = %error, + "contextual instruction decode failed" + ); + }, + } + } + let cancelled = observer.is_cancelled(); + tracing::debug!(target: crate::TRACING_TARGET, action = "complete_input", campaign_id = %campaign_id, signature = %input.signature, slot = input.slot, instruction_path = %input.instruction_path, program_id = %input.program_id, failed, cancelled, processors = ?processors, "contextual decode replay input completed"); + return DecodeItemOutcome { + signature: input.signature.clone(), + slot: input.slot, + instruction_path: input.instruction_path.clone(), + program_id: input.program_id.clone(), + started: true, + unmatched: false, + failed, + cancelled, + processors, + }; +} + +async fn execute_decoder( + store: &S, + request: &crate::DecodeReplayRequest, + campaign_id: &str, + ranked: &RankedDecoder<'_>, + materializers: &[std::sync::Arc], + input: &kb_lib::MdCoreInstructionReplayInput, +) -> kb_core::Result +where + S: kb_store::DecodePipelineStore + Sync, +{ + let input_hash_result = kb_lib::decoder_api_contextual_input_hash(input); + let input_hash = match input_hash_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let ledger_result = kb_store::ProcessingLedgerIdentity::new( + crate::INSTRUCTION_DECODE_STAGE, + ranked.identity.name.clone(), + ranked.identity.version.clone(), + input.replay_input_key.clone(), + input_hash.clone(), + ); + let ledger_identity = match ledger_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + tracing::debug!( + target: crate::TRACING_TARGET, + action = "execute_decoder", + campaign_id = %campaign_id, + signature = %input.signature, + slot = input.slot, + instruction_path = %input.instruction_path, + program_id = %input.program_id, + input_key = %input.replay_input_key, + input_hash = %input_hash, + processor_name = %ranked.identity.name, + processor_version = %ranked.identity.version, + recognition = ?ranked.recognition, + force_replay = request.force_replay, + materialize_after_decode = request.materialize_after_decode, + "execute contextual instruction decoder" + ); + let mut summary = crate::DecodeProcessorSummary { + processor_name: ranked.identity.name.clone(), + processor_version: ranked.identity.version.clone(), + dispatched: 1, + skipped: 0, + decoded: 0, + ignored: 0, + unsupported: 0, + failed: 0, + materialized_outputs: 0, + materialization_refused: 0, + }; + if !request.force_replay && !request.selection.incomplete_signatures { + tracing::debug!(target: crate::TRACING_TARGET, action = "check_decode_ledger", campaign_id = %campaign_id, signature = %input.signature, instruction_path = %input.instruction_path, processor_name = %ranked.identity.name, processor_version = %ranked.identity.version, input_key = %ledger_identity.input_key, input_hash = %ledger_identity.input_hash, "check current contextual decode ledger identity"); + let current_result = store.is_decode_current(&ledger_identity).await; + match current_result { + std::result::Result::Ok(true) => { + summary.skipped = 1; + tracing::debug!(target: crate::TRACING_TARGET, action = "skip_decoder", campaign_id = %campaign_id, signature = %input.signature, instruction_path = %input.instruction_path, program_id = %input.program_id, processor_name = %ranked.identity.name, processor_version = %ranked.identity.version, input_key = %ledger_identity.input_key, input_hash = %ledger_identity.input_hash, reason = "same_processor_version_and_input_hash", "skip current contextual instruction decode"); + return std::result::Result::Ok(summary); + }, + std::result::Result::Ok(false) => { + tracing::debug!(target: crate::TRACING_TARGET, action = "check_decode_ledger", campaign_id = %campaign_id, signature = %input.signature, instruction_path = %input.instruction_path, processor_name = %ranked.identity.name, processor_version = %ranked.identity.version, current = false, "contextual decode ledger identity is not current"); + }, + std::result::Result::Err(error) => return std::result::Result::Err(error), + } + } else { + tracing::debug!(target: crate::TRACING_TARGET, action = "check_decode_ledger", campaign_id = %campaign_id, signature = %input.signature, instruction_path = %input.instruction_path, processor_name = %ranked.identity.name, processor_version = %ranked.identity.version, force_replay = request.force_replay, incomplete_signatures = request.selection.incomplete_signatures, ledger_check_bypassed = true, "selected replay scope bypasses contextual decode skip check"); + } + tracing::debug!(target: crate::TRACING_TARGET, action = "call_decoder", campaign_id = %campaign_id, signature = %input.signature, slot = input.slot, instruction_path = %input.instruction_path, program_id = %input.program_id, processor_name = %ranked.identity.name, processor_version = %ranked.identity.version, transaction_failed = input.transaction_failed, "call contextual instruction decoder implementation"); + let decode_result = ranked.decoder.decode(input); + tracing::debug!(target: crate::TRACING_TARGET, action = "call_decoder", campaign_id = %campaign_id, signature = %input.signature, instruction_path = %input.instruction_path, processor_name = %ranked.identity.name, processor_version = %ranked.identity.version, status = ?decode_result.status, recognized_entry_code = ?decode_result.recognized_entry_code, observation_count = decode_result.observations.len(), diagnostics = ?decode_result.diagnostics, "contextual instruction decoder returned"); + if matches!( + decode_result.status, + kb_lib::DcApiDecoderOutcomeStatus::Failed | kb_lib::DcApiDecoderOutcomeStatus::Unsupported + ) { + tracing::error!( + target: crate::TRACING_TARGET, + action = "decoder_outcome_failure", + campaign_id = %campaign_id, + signature = %input.signature, + slot = input.slot, + instruction_path = %input.instruction_path, + program_id = %input.program_id, + processor_name = %ranked.identity.name, + processor_version = %ranked.identity.version, + input_key = %input.replay_input_key, + status = ?decode_result.status, + recognized_entry_code = ?decode_result.recognized_entry_code, + diagnostics = ?decode_result.diagnostics, + "contextual instruction was not decoded successfully" + ); + } + let validation_result = decode_result.validate(); + if let std::result::Result::Err(error) = validation_result { + tracing::error!(target: crate::TRACING_TARGET, action = "validate_decoder_result", campaign_id = %campaign_id, signature = %input.signature, instruction_path = %input.instruction_path, processor_name = %ranked.identity.name, processor_version = %ranked.identity.version, error = %error, "contextual decoder returned an invalid result"); + let failure = kb_store::DecodeFailure { + ledger_identity, + signature: input.signature.clone(), + instruction_path: input.instruction_path.clone(), + error_code: "invalid_decoder_result".to_string(), + error_message: error.to_string(), + }; + let persist_failure_result = store.mark_decode_failed(&failure).await; + if let std::result::Result::Err(persist_error) = persist_failure_result { + return std::result::Result::Err(persist_error); + } + return std::result::Result::Err(error); + } + let observation_inserts_result = observation_inserts( + &ranked.identity, + input, + input_hash.as_str(), + &decode_result.observations, + ); + let observation_inserts = match observation_inserts_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let status = decoder_status_code(decode_result.status).to_string(); + let decoded_count_result = u32::try_from(decode_result.observations.len()); + let decoded_count = match decoded_count_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + return std::result::Result::Err(kb_core::Error::invalid_state(format!( + "decoded observation count conversion failed: {error}" + ))); + }, + }; + let error_count_result = u32::try_from(decode_result.diagnostics.len()); + let error_count = match error_count_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + return std::result::Result::Err(kb_core::Error::invalid_state(format!( + "decoder diagnostic count conversion failed: {error}" + ))); + }, + }; + let coverage = kb_store::DecodeCoverageObservationInsert { + processor_name: ranked.identity.name.clone(), + processor_version: ranked.identity.version.clone(), + input_key: input.replay_input_key.clone(), + input_hash: input_hash.clone(), + signature: input.signature.clone(), + slot: input.slot, + instruction_path: input.instruction_path.clone(), + program_id: input.program_id.clone(), + surface_code: ranked.recognition.surface_code.clone(), + entry_code: decode_result + .recognized_entry_code + .clone() + .or_else(|| return ranked.recognition.entry_code.clone()), + discriminator_hex: ranked.recognition.discriminator_hex.clone(), + status: status.clone(), + recognized: ranked.recognition.compatible, + decoded_count, + materialized_count: 0, + error_count, + transaction_failed: input.transaction_failed, + }; + let decode_error = if decode_result.status == kb_lib::DcApiDecoderOutcomeStatus::Failed { + decode_result.diagnostics.first() + } else { + std::option::Option::None + }; + let bundle = kb_store::DecodePersistenceBundle { + ledger_identity, + signature: input.signature.clone(), + instruction_path: input.instruction_path.clone(), + status: status.clone(), + error_code: decode_error.map(|diagnostic| return diagnostic.code.clone()), + error_message: decode_error.map(|diagnostic| return diagnostic.message.clone()), + observations: observation_inserts, + coverage, + }; + tracing::debug!(target: crate::TRACING_TARGET, action = "persist_decode_result", campaign_id = %campaign_id, signature = %input.signature, slot = input.slot, instruction_path = %input.instruction_path, program_id = %input.program_id, processor_name = %ranked.identity.name, processor_version = %ranked.identity.version, input_key = %bundle.ledger_identity.input_key, input_hash = %bundle.ledger_identity.input_hash, status = %bundle.status, observation_count = bundle.observations.len(), force_replay = request.force_replay, "persist contextual decode result"); + let persist_result = store.persist_decode_result(&bundle, request.force_replay).await; + let persist_outcome = match persist_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + tracing::error!(target: crate::TRACING_TARGET, action = "persist_decode_result", campaign_id = %campaign_id, signature = %input.signature, instruction_path = %input.instruction_path, processor_name = %ranked.identity.name, processor_version = %ranked.identity.version, error = %error, "contextual decode result persistence failed"); + return std::result::Result::Err(error); + }, + }; + tracing::debug!(target: crate::TRACING_TARGET, action = "persist_decode_result", campaign_id = %campaign_id, signature = %input.signature, instruction_path = %input.instruction_path, processor_name = %ranked.identity.name, processor_version = %ranked.identity.version, outcome = ?persist_outcome, "contextual decode result persisted"); + match decode_result.status { + kb_lib::DcApiDecoderOutcomeStatus::Decoded => summary.decoded = 1, + kb_lib::DcApiDecoderOutcomeStatus::Ignored => summary.ignored = 1, + kb_lib::DcApiDecoderOutcomeStatus::Unsupported => summary.unsupported = 1, + kb_lib::DcApiDecoderOutcomeStatus::Failed => summary.failed = 1, + } + if request.materialize_after_decode + && decode_result.status == kb_lib::DcApiDecoderOutcomeStatus::Decoded + { + tracing::debug!(target: crate::TRACING_TARGET, action = "materialize_after_decode", campaign_id = %campaign_id, signature = %input.signature, instruction_path = %input.instruction_path, processor_name = %ranked.identity.name, processor_version = %ranked.identity.version, observation_count = decode_result.observations.len(), materializer_count = materializers.len(), "run materializers after contextual decode"); + for observation in &decode_result.observations { + let materialization_result = execute_materializers( + store, + request.force_replay, + campaign_id, + &ranked.identity, + input, + observation, + materializers, + ) + .await; + let materialization = match materialization_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + summary.materialized_outputs += materialization.0; + summary.materialization_refused += materialization.1; + } + } else { + tracing::debug!(target: crate::TRACING_TARGET, action = "materialize_after_decode", campaign_id = %campaign_id, signature = %input.signature, instruction_path = %input.instruction_path, processor_name = %ranked.identity.name, processor_version = %ranked.identity.version, requested = request.materialize_after_decode, decode_status = ?decode_result.status, executed = false, "materialization not executed for contextual decode result"); + } + tracing::debug!(target: crate::TRACING_TARGET, action = "execute_decoder", campaign_id = %campaign_id, signature = %input.signature, instruction_path = %input.instruction_path, processor_name = %ranked.identity.name, processor_version = %ranked.identity.version, summary = ?summary, "contextual instruction decoder execution completed"); + return std::result::Result::Ok(summary); +} + +async fn execute_materializers( + store: &S, + force_replay: bool, + campaign_id: &str, + decoder_identity: &kb_lib::DcApiDecoderIdentity, + input: &kb_lib::MdCoreInstructionReplayInput, + observation: &kb_lib::DcApiDecodedObservation, + materializers: &[std::sync::Arc], +) -> kb_core::Result<(u64, u64)> +where + S: kb_store::DecodePipelineStore + Sync, +{ + tracing::debug!(target: crate::TRACING_TARGET, action = "execute_materializers", campaign_id = %campaign_id, signature = %input.signature, slot = input.slot, instruction_path = %input.instruction_path, program_id = %input.program_id, source_decoder_name = %decoder_identity.name, source_decoder_version = %decoder_identity.version, event_key = %observation.event_key, event_family = ?observation.event.event_family, materializer_count = materializers.len(), force_replay, "evaluate materializers for decoded observation"); + let mut output_count = 0_u64; + let mut refusal_count = 0_u64; + for materializer in materializers { + let identity = materializer.identity(); + if !kb_lib::materializer_api_accepts_observation(materializer.as_ref(), observation) { + tracing::debug!(target: crate::TRACING_TARGET, action = "select_materializer", campaign_id = %campaign_id, signature = %input.signature, instruction_path = %input.instruction_path, event_key = %observation.event_key, materializer_name = %identity.name, materializer_version = %identity.version, accepted = false, event_family = ?observation.event.event_family, surface_code = %observation.event.surface_code.0.as_str(), entry_code = %observation.event.event_name.0.as_str(), "materializer does not accept decoded observation"); + continue; + } + tracing::debug!(target: crate::TRACING_TARGET, action = "select_materializer", campaign_id = %campaign_id, signature = %input.signature, instruction_path = %input.instruction_path, event_key = %observation.event_key, materializer_name = %identity.name, materializer_version = %identity.version, accepted = true, event_family = ?observation.event.event_family, surface_code = %observation.event.surface_code.0.as_str(), entry_code = %observation.event.event_name.0.as_str(), "materializer accepts decoded observation"); + let materializer_input_key = + materializer_input_key(input, decoder_identity, observation.event_key.as_str()); + let serialize_result = serde_json::to_value(observation); + let mut observation_json = match serialize_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + return std::result::Result::Err(kb_core::Error::json(format!( + "cannot serialize decoded observation for materialization hash: {error}" + ))); + }, + }; + let hash_result = kb_lib::decoder_api_deterministic_json_hash(&mut observation_json); + let input_hash = match hash_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let ledger_result = kb_store::ProcessingLedgerIdentity::new( + crate::EVENT_MATERIALIZATION_STAGE, + identity.name.clone(), + identity.version.clone(), + materializer_input_key.clone(), + input_hash.clone(), + ); + let ledger_identity = match ledger_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + if !force_replay { + tracing::debug!(target: crate::TRACING_TARGET, action = "check_materialization_ledger", campaign_id = %campaign_id, signature = %input.signature, instruction_path = %input.instruction_path, event_key = %observation.event_key, materializer_name = %identity.name, materializer_version = %identity.version, input_key = %ledger_identity.input_key, input_hash = %ledger_identity.input_hash, "check current materialization ledger identity"); + let current_result = store.is_decode_current(&ledger_identity).await; + match current_result { + std::result::Result::Ok(true) => { + tracing::debug!(target: crate::TRACING_TARGET, action = "skip_materializer", campaign_id = %campaign_id, signature = %input.signature, instruction_path = %input.instruction_path, event_key = %observation.event_key, materializer_name = %identity.name, materializer_version = %identity.version, reason = "same_processor_version_and_input_hash", "skip current materialization"); + continue; + }, + std::result::Result::Ok(false) => {}, + std::result::Result::Err(error) => return std::result::Result::Err(error), + } + } else { + tracing::debug!(target: crate::TRACING_TARGET, action = "check_materialization_ledger", campaign_id = %campaign_id, signature = %input.signature, instruction_path = %input.instruction_path, event_key = %observation.event_key, materializer_name = %identity.name, materializer_version = %identity.version, force_replay = true, ledger_check_bypassed = true, "force replay bypasses materialization skip check"); + } + let policy_result = kb_lib::materializer_api_validate_materialization_policy( + materializer.as_ref(), + observation, + ); + let mut execution = match policy_result { + std::result::Result::Ok(()) => materializer.materialize(observation), + std::result::Result::Err(refusal) => refusal, + }; + let output_policy_result = + kb_lib::materializer_api_validate_materialized_output_policy(observation, &execution); + if let std::result::Result::Err(refusal) = output_policy_result { + execution = refusal; + } + if execution.status == kb_lib::MtApiMaterializerOutcomeStatus::Refused { + refusal_count += 1; + } + tracing::debug!(target: crate::TRACING_TARGET, action = "call_materializer", campaign_id = %campaign_id, signature = %input.signature, instruction_path = %input.instruction_path, event_key = %observation.event_key, materializer_name = %identity.name, materializer_version = %identity.version, status = ?execution.status, output_count = execution.outputs.len(), diagnostics = ?execution.diagnostics, "materializer returned"); + if execution.status == kb_lib::MtApiMaterializerOutcomeStatus::Failed { + tracing::error!( + target: crate::TRACING_TARGET, + action = "materializer_outcome_failure", + campaign_id = %campaign_id, + signature = %input.signature, + slot = input.slot, + instruction_path = %input.instruction_path, + program_id = %input.program_id, + event_key = %observation.event_key, + materializer_name = %identity.name, + materializer_version = %identity.version, + surface_code = %observation.event.surface_code.0.as_str(), + entry_code = %observation.event.event_name.0.as_str(), + diagnostics = ?execution.diagnostics, + "decoded observation materialization failed" + ); + } + let validation_result = execution.validate(); + if let std::result::Result::Err(error) = validation_result { + return std::result::Result::Err(error); + } + let mut outputs = std::vec::Vec::with_capacity(execution.outputs.len()); + for output in &execution.outputs { + outputs.push(kb_store::MaterializedOutputInsert { + processor_name: identity.name.clone(), + processor_version: identity.version.clone(), + input_key: materializer_input_key.clone(), + input_hash: input_hash.clone(), + output_key: output.output_key.clone(), + source_event_key: observation.event_key.clone(), + signature: input.signature.clone(), + slot: input.slot, + materialized_family: materialized_family_code(output.family).to_string(), + payload_json: output.payload_json.clone(), + }); + } + let materialization_error = + if execution.status == kb_lib::MtApiMaterializerOutcomeStatus::Failed { + execution.diagnostics.first() + } else { + std::option::Option::None + }; + let bundle = kb_store::MaterializationPersistenceBundle { + ledger_identity, + source_decoder_name: decoder_identity.name.clone(), + source_decoder_version: decoder_identity.version.clone(), + source_decode_input_key: input.replay_input_key.clone(), + signature: input.signature.clone(), + instruction_path: input.instruction_path.clone(), + status: materializer_status_code(execution.status).to_string(), + error_code: materialization_error.map(|diagnostic| return diagnostic.code.clone()), + error_message: materialization_error + .map(|diagnostic| return diagnostic.message.clone()), + outputs, + }; + tracing::debug!(target: crate::TRACING_TARGET, action = "persist_materialization_result", campaign_id = %campaign_id, signature = %input.signature, instruction_path = %input.instruction_path, event_key = %observation.event_key, materializer_name = %identity.name, materializer_version = %identity.version, input_key = %bundle.ledger_identity.input_key, input_hash = %bundle.ledger_identity.input_hash, status = %bundle.status, output_count = bundle.outputs.len(), force_replay, "persist materialization result"); + let persist_result = store.persist_materialization_result(&bundle, force_replay).await; + let persist_outcome = match persist_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + tracing::error!(target: crate::TRACING_TARGET, action = "persist_materialization_result", campaign_id = %campaign_id, signature = %input.signature, instruction_path = %input.instruction_path, event_key = %observation.event_key, materializer_name = %identity.name, materializer_version = %identity.version, error = %error, "materialization result persistence failed"); + return std::result::Result::Err(error); + }, + }; + tracing::debug!(target: crate::TRACING_TARGET, action = "persist_materialization_result", campaign_id = %campaign_id, signature = %input.signature, instruction_path = %input.instruction_path, event_key = %observation.event_key, materializer_name = %identity.name, materializer_version = %identity.version, outcome = ?persist_outcome, "materialization result persisted"); + let count_result = u64::try_from(execution.outputs.len()); + let count = match count_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + return std::result::Result::Err(kb_core::Error::invalid_state(format!( + "materialized output count conversion failed: {error}" + ))); + }, + }; + output_count += count; + } + tracing::debug!(target: crate::TRACING_TARGET, action = "execute_materializers", campaign_id = %campaign_id, signature = %input.signature, instruction_path = %input.instruction_path, event_key = %observation.event_key, output_count, refusal_count, "materializer evaluation completed"); + return std::result::Result::Ok((output_count, refusal_count)); +} + +fn materializer_input_key( + input: &kb_lib::MdCoreInstructionReplayInput, + decoder_identity: &kb_lib::DcApiDecoderIdentity, + event_key: &str, +) -> std::string::String { + return format!( + "{}:{}:{}:{}", + input.replay_input_key, decoder_identity.name, decoder_identity.version, event_key + ); +} + +fn validate_decoder_registry( + decoders: &[std::sync::Arc], +) -> kb_core::Result<()> { + let mut identities = std::collections::BTreeSet::new(); + for decoder in decoders { + let identity = decoder.identity(); + if identity.name.trim().is_empty() || identity.version.trim().is_empty() { + return std::result::Result::Err(kb_core::Error::config( + "decoder registry contains an empty processor identity", + )); + } + if !identities.insert((identity.name, identity.version)) { + return std::result::Result::Err(kb_core::Error::config( + "decoder registry contains a duplicate processor identity", + )); + } + } + return std::result::Result::Ok(()); +} + +fn ordered_materializers( + materializers: &[std::sync::Arc], +) -> kb_core::Result>> { + let mut output = materializers.to_vec(); + output.sort_by(|left, right| { + let left_identity = left.identity(); + let right_identity = right.identity(); + return left_identity + .name + .cmp(&right_identity.name) + .then(left_identity.version.cmp(&right_identity.version)); + }); + let mut previous = std::option::Option::None; + for materializer in &output { + let identity = materializer.identity(); + if identity.name.trim().is_empty() || identity.version.trim().is_empty() { + return std::result::Result::Err(kb_core::Error::config( + "materializer registry contains an empty processor identity", + )); + } + let key = (identity.name, identity.version); + if previous.as_ref() == std::option::Option::Some(&key) { + return std::result::Result::Err(kb_core::Error::config( + "materializer registry contains a duplicate processor identity", + )); + } + previous = std::option::Option::Some(key); + } + return std::result::Result::Ok(output); +} + +/// Creates one stable process-local contextual decode campaign identifier. +pub fn new_decode_campaign_id() -> std::string::String { + let sequence = DECODE_CAMPAIGN_SEQUENCE.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + return format!("decode-{}-{sequence}", chrono::Utc::now().timestamp_micros()); +} + +fn text_sample(values: &[std::string::String], limit: usize) -> std::vec::Vec<&str> { + return values.iter().take(limit).map(std::string::String::as_str).collect(); +} + +fn effective_selection( + request: &crate::DecodeReplayRequest, + decoders: &[std::sync::Arc], +) -> kb_core::Result { + let mut supported_program_ids = std::collections::BTreeSet::::new(); + for decoder in decoders { + for surface in decoder.surfaces() { + supported_program_ids.insert(surface.program_id.to_string()); + } + } + if supported_program_ids.is_empty() { + return std::result::Result::Err(kb_core::Error::config( + "enabled decoders declare no exact program ids", + )); + } + let effective_program_ids = if request.selection.program_ids.is_empty() { + supported_program_ids.iter().cloned().collect::>() + } else { + let mut requested = std::collections::BTreeSet::::new(); + let mut unsupported = std::vec::Vec::::new(); + for program_id in &request.selection.program_ids { + if supported_program_ids.contains(program_id) { + requested.insert(program_id.clone()); + } else { + unsupported.push(program_id.clone()); + } + } + if !unsupported.is_empty() { + unsupported.sort(); + return std::result::Result::Err(kb_core::Error::config(format!( + "selected program ids are not supported by enabled decoders: {}", + unsupported.join(", ") + ))); + } + requested.into_iter().collect::>() + }; + let bypass_processing_states = request.force_replay + && (!request.selection.signatures.is_empty() || request.force_replay_all_matching); + let processing_states = if bypass_processing_states { + std::vec::Vec::new() + } else { + request.selection.processing_states.clone() + }; + return kb_store::DecodeSelectionFilter::new( + request.selection.signatures.clone(), + processing_states, + request.selection.min_slot, + request.selection.max_slot, + effective_program_ids, + request.selection.instruction_paths.clone(), + request.selection.incomplete_signatures, + request.selection.limit, + ); +} + +fn enabled_decoders( + request: &crate::DecodeReplayRequest, + decoders: &[std::sync::Arc], +) -> std::vec::Vec> { + let mut output = decoders + .iter() + .filter(|decoder| { + if request.decoder_names.is_empty() { + return true; + } + let identity = decoder.identity(); + return request.decoder_names.iter().any(|name| return *name == identity.name); + }) + .cloned() + .collect::>(); + output.sort_by(|left, right| { + let left_identity = left.identity(); + let right_identity = right.identity(); + return left_identity + .name + .cmp(&right_identity.name) + .then(left_identity.version.cmp(&right_identity.version)); + }); + return output; +} + +fn rank_decoders<'decoder>( + campaign_id: &str, + input: &kb_lib::MdCoreInstructionReplayInput, + decoders: &'decoder [std::sync::Arc], +) -> std::vec::Vec> { + let mut output = std::vec::Vec::new(); + for decoder in decoders { + let identity = decoder.identity(); + if !kb_lib::decoder_api_decoder_handles_program_id( + decoder.as_ref(), + input.program_id.as_str(), + ) { + tracing::debug!(target: crate::TRACING_TARGET, action = "recognize_input", campaign_id = %campaign_id, signature = %input.signature, instruction_path = %input.instruction_path, program_id = %input.program_id, processor_name = %identity.name, processor_version = %identity.version, handles_program_id = false, recognize_called = false, "decoder rejected by exact program id before recognition"); + continue; + } + tracing::debug!(target: crate::TRACING_TARGET, action = "recognize_input", campaign_id = %campaign_id, signature = %input.signature, instruction_path = %input.instruction_path, program_id = %input.program_id, processor_name = %identity.name, processor_version = %identity.version, handles_program_id = true, recognize_called = true, "call decoder recognition"); + let recognition = decoder.recognize(input); + if !recognition.compatible { + tracing::debug!(target: crate::TRACING_TARGET, action = "recognize_input", campaign_id = %campaign_id, signature = %input.signature, instruction_path = %input.instruction_path, program_id = %input.program_id, processor_name = %identity.name, processor_version = %identity.version, recognition = ?recognition, compatible = false, "decoder recognition rejected contextual input"); + continue; + } + tracing::debug!(target: crate::TRACING_TARGET, action = "recognize_input", campaign_id = %campaign_id, signature = %input.signature, instruction_path = %input.instruction_path, program_id = %input.program_id, processor_name = %identity.name, processor_version = %identity.version, recognition = ?recognition, compatible = true, "decoder recognition accepted contextual input"); + output.push(RankedDecoder { + decoder: decoder.as_ref(), + identity, + recognition, + }); + } + output.sort_by(|left, right| { + return right + .recognition + .exact + .cmp(&left.recognition.exact) + .then(right.recognition.priority.cmp(&left.recognition.priority)) + .then(left.identity.name.cmp(&right.identity.name)) + .then(left.identity.version.cmp(&right.identity.version)); + }); + return output; +} + +fn coverage_declaration_inserts( + decoder: &dyn kb_lib::DcApiInstructionDecoder, +) -> std::vec::Vec { + let identity = decoder.identity(); + return decoder + .coverage() + .into_iter() + .map(|entry| { + return kb_store::DecodeCoverageDeclarationInsert { + processor_name: identity.name.clone(), + processor_version: identity.version.clone(), + program_id: entry.program_id, + surface_code: entry.surface_code, + entry_kind: coverage_entry_kind_code(entry.entry_kind).to_string(), + entry_code: entry.entry_code, + discriminator_hex: entry.discriminator_hex, + historical: entry.historical, + }; + }) + .collect(); +} + +fn observation_inserts( + identity: &kb_lib::DcApiDecoderIdentity, + input: &kb_lib::MdCoreInstructionReplayInput, + input_hash: &str, + observations: &[kb_lib::DcApiDecodedObservation], +) -> kb_core::Result> { + let mut output = std::vec::Vec::with_capacity(observations.len()); + for observation in observations { + if observation.event.signature.0 != input.signature + || observation.event.slot.0 != input.slot + || observation.event.instruction_path.0 != input.instruction_path + || observation.event.program_id.0 != input.program_id + || observation.transaction_failed != input.transaction_failed + || observation.transaction_error != input.transaction_err_json + { + return std::result::Result::Err(kb_core::Error::invalid_state( + "decoded observation context does not match contextual input", + )); + } + let proof_result = serde_json::to_value(&observation.proof); + let proof_json = match proof_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + return std::result::Result::Err(kb_core::Error::json(format!( + "cannot serialize decoder proof: {error}" + ))); + }, + }; + output.push(kb_store::DecodeObservationInsert { + processor_name: identity.name.clone(), + processor_version: identity.version.clone(), + input_key: input.replay_input_key.clone(), + input_hash: input_hash.to_string(), + event_key: observation.event_key.clone(), + signature: observation.event.signature.0.clone(), + slot: observation.event.slot.0, + instruction_path: observation.event.instruction_path.0.clone(), + program_id: observation.event.program_id.0.clone(), + protocol_code: observation.event.protocol_code.0.clone(), + surface_code: observation.event.surface_code.0.clone(), + event_code: observation.event.event_code.0.clone(), + event_name: observation.event.event_name.0.clone(), + event_family: event_family_code(observation.event.event_family).to_string(), + source_kind: event_source_kind_code(observation.event.source_kind).to_string(), + confidence: decoder_confidence_code(observation.event.confidence).to_string(), + proof_kind: proof_kind_code(observation.proof.kind).to_string(), + proof_json, + payload_json: observation.payload_json.clone(), + transaction_failed: observation.transaction_failed, + transaction_error: observation.transaction_error.clone(), + observation_committed: observation.observation_committed, + }); + } + return std::result::Result::Ok(output); +} + +fn merge_processor_summary( + summaries: &mut std::vec::Vec, + value: crate::DecodeProcessorSummary, +) { + for summary in summaries.iter_mut() { + if summary.processor_name == value.processor_name + && summary.processor_version == value.processor_version + { + summary.dispatched += value.dispatched; + summary.skipped += value.skipped; + summary.decoded += value.decoded; + summary.ignored += value.ignored; + summary.unsupported += value.unsupported; + summary.failed += value.failed; + summary.materialized_outputs += value.materialized_outputs; + summary.materialization_refused += value.materialization_refused; + return; + } + } + summaries.push(value); +} + +fn decoder_status_code(status: kb_lib::DcApiDecoderOutcomeStatus) -> &'static str { + return match status { + kb_lib::DcApiDecoderOutcomeStatus::Decoded => "decoded", + kb_lib::DcApiDecoderOutcomeStatus::Ignored => "ignored", + kb_lib::DcApiDecoderOutcomeStatus::Unsupported => "unsupported", + kb_lib::DcApiDecoderOutcomeStatus::Failed => "failed", + }; +} + +fn materializer_status_code(status: kb_lib::MtApiMaterializerOutcomeStatus) -> &'static str { + return match status { + kb_lib::MtApiMaterializerOutcomeStatus::Inserted => "inserted", + kb_lib::MtApiMaterializerOutcomeStatus::Replaced => "replaced", + kb_lib::MtApiMaterializerOutcomeStatus::Ignored => "ignored", + kb_lib::MtApiMaterializerOutcomeStatus::Refused => "refused", + kb_lib::MtApiMaterializerOutcomeStatus::Failed => "failed", + }; +} + +fn coverage_entry_kind_code(kind: kb_lib::DcApiDecoderCoverageEntryKind) -> &'static str { + return match kind { + kb_lib::DcApiDecoderCoverageEntryKind::Instruction => "instruction", + kb_lib::DcApiDecoderCoverageEntryKind::Event => "event", + kb_lib::DcApiDecoderCoverageEntryKind::Discriminator => "discriminator", + }; +} + +fn proof_kind_code(kind: kb_lib::DcApiDecoderProofKind) -> &'static str { + return match kind { + kb_lib::DcApiDecoderProofKind::ExactDiscriminator => "exact_discriminator", + kb_lib::DcApiDecoderProofKind::ExactLayout => "exact_layout", + kb_lib::DcApiDecoderProofKind::Idl => "idl", + kb_lib::DcApiDecoderProofKind::Manual => "manual", + kb_lib::DcApiDecoderProofKind::LogCorrelation => "log_correlation", + kb_lib::DcApiDecoderProofKind::BalanceDelta => "balance_delta", + kb_lib::DcApiDecoderProofKind::Heuristic => "heuristic", + kb_lib::DcApiDecoderProofKind::Audit => "audit", + kb_lib::DcApiDecoderProofKind::Unknown => "unknown", + }; +} + +fn decoder_confidence_code(confidence: kb_lib::MdDecoderConfidence) -> &'static str { + return match confidence { + kb_lib::MdDecoderConfidence::Exact => "exact", + kb_lib::MdDecoderConfidence::IdlExact => "idl_exact", + kb_lib::MdDecoderConfidence::ManualExact => "manual_exact", + kb_lib::MdDecoderConfidence::Inferred => "inferred", + kb_lib::MdDecoderConfidence::Unsafe => "unsafe", + kb_lib::MdDecoderConfidence::AuditOnly => "audit_only", + kb_lib::MdDecoderConfidence::Unknown => "unknown", + }; +} + +fn event_source_kind_code(kind: kb_lib::MdEventSourceKind) -> &'static str { + return match kind { + kb_lib::MdEventSourceKind::Instruction => "instruction", + kb_lib::MdEventSourceKind::InnerInstruction => "inner_instruction", + kb_lib::MdEventSourceKind::Log => "log", + kb_lib::MdEventSourceKind::AnchorEvent => "anchor_event", + kb_lib::MdEventSourceKind::AnchorSelfCpiEvent => "anchor_self_cpi_event", + kb_lib::MdEventSourceKind::BalanceDelta => "balance_delta", + kb_lib::MdEventSourceKind::Synthetic => "synthetic", + kb_lib::MdEventSourceKind::Inferred => "inferred", + kb_lib::MdEventSourceKind::Audit => "audit", + }; +} + +fn event_family_code(family: kb_lib::MdEventFamily) -> &'static str { + return match family { + kb_lib::MdEventFamily::Trade => "trade", + kb_lib::MdEventFamily::Liquidity => "liquidity", + kb_lib::MdEventFamily::Lifecycle => "lifecycle", + kb_lib::MdEventFamily::Fee => "fee", + kb_lib::MdEventFamily::Admin => "admin", + kb_lib::MdEventFamily::Reward => "reward", + kb_lib::MdEventFamily::Orderbook => "orderbook", + kb_lib::MdEventFamily::TokenAccount => "token_account", + kb_lib::MdEventFamily::TokenMint => "token_mint", + kb_lib::MdEventFamily::TokenBurn => "token_burn", + kb_lib::MdEventFamily::Nft => "nft", + kb_lib::MdEventFamily::Metadata => "metadata", + kb_lib::MdEventFamily::Oracle => "oracle", + kb_lib::MdEventFamily::Lending => "lending", + kb_lib::MdEventFamily::Staking => "staking", + kb_lib::MdEventFamily::Governance => "governance", + kb_lib::MdEventFamily::Bridge => "bridge", + kb_lib::MdEventFamily::Perpetuals => "perpetuals", + kb_lib::MdEventFamily::Vault => "vault", + kb_lib::MdEventFamily::Routing => "routing", + kb_lib::MdEventFamily::ComplianceAudit => "compliance_audit", + kb_lib::MdEventFamily::TokenMetadataRisk => "token_metadata_risk", + kb_lib::MdEventFamily::Risk => "risk", + kb_lib::MdEventFamily::Audit => "audit", + kb_lib::MdEventFamily::Unknown => "unknown", + }; +} + +fn materialized_family_code(family: kb_lib::MdMaterializedEventFamily) -> &'static str { + return match family { + kb_lib::MdMaterializedEventFamily::Trade => "trade", + kb_lib::MdMaterializedEventFamily::Liquidity => "liquidity", + kb_lib::MdMaterializedEventFamily::Lifecycle => "lifecycle", + kb_lib::MdMaterializedEventFamily::Fee => "fee", + kb_lib::MdMaterializedEventFamily::Admin => "admin", + kb_lib::MdMaterializedEventFamily::TokenAccount => "token_account", + kb_lib::MdMaterializedEventFamily::PoolState => "pool_state", + kb_lib::MdMaterializedEventFamily::Orderbook => "orderbook", + kb_lib::MdMaterializedEventFamily::Reward => "reward", + kb_lib::MdMaterializedEventFamily::Nft => "nft", + kb_lib::MdMaterializedEventFamily::Metadata => "metadata", + kb_lib::MdMaterializedEventFamily::Oracle => "oracle", + kb_lib::MdMaterializedEventFamily::Lending => "lending", + kb_lib::MdMaterializedEventFamily::Staking => "staking", + kb_lib::MdMaterializedEventFamily::Governance => "governance", + kb_lib::MdMaterializedEventFamily::Bridge => "bridge", + kb_lib::MdMaterializedEventFamily::Perpetuals => "perpetuals", + kb_lib::MdMaterializedEventFamily::Vault => "vault", + kb_lib::MdMaterializedEventFamily::Routing => "routing", + kb_lib::MdMaterializedEventFamily::ComplianceAudit => "compliance_audit", + kb_lib::MdMaterializedEventFamily::TokenMetadataRisk => "token_metadata_risk", + kb_lib::MdMaterializedEventFamily::Risk => "risk", + kb_lib::MdMaterializedEventFamily::TransactionAnnotation => "transaction_annotation", + kb_lib::MdMaterializedEventFamily::Unknown => "unknown", + }; +} + +#[cfg(test)] +mod tests { + fn lock_or_panic(mutex: &std::sync::Mutex) -> std::sync::MutexGuard<'_, T> { + return match mutex.lock() { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("test mutex is poisoned: {error}"), + }; + } + + fn result_or_panic(result: std::result::Result) -> T + where + E: std::fmt::Display, + { + return match result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("test operation failed: {error}"), + }; + } + + static SURFACES: &[kb_lib::DcApiDecoderSurface] = &[kb_lib::DcApiDecoderSurface { + program_id: "program_a", + surface_code: "surface_a", + priority: 10, + }]; + + struct TestDecoder { + version: &'static str, + exact: bool, + priority: u16, + } + + impl kb_lib::DcApiInstructionDecoder for TestDecoder { + fn identity(&self) -> kb_lib::DcApiDecoderIdentity { + return kb_lib::DcApiDecoderIdentity { + name: "test_decoder".to_string(), + version: self.version.to_string(), + }; + } + + fn surfaces(&self) -> &'static [kb_lib::DcApiDecoderSurface] { + return SURFACES; + } + + fn coverage(&self) -> std::vec::Vec { + return std::vec![kb_lib::DcApiDecoderCoverageDeclaration { + program_id: "program_a".to_string(), + surface_code: std::option::Option::Some("surface_a".to_string()), + entry_kind: kb_lib::DcApiDecoderCoverageEntryKind::Instruction, + entry_code: "attempt".to_string(), + discriminator_hex: std::option::Option::None, + historical: false, + }]; + } + + fn recognize( + &self, + input: &kb_lib::MdCoreInstructionReplayInput, + ) -> kb_lib::DcApiDecoderRecognition { + if input.program_id != "program_a" { + return kb_lib::DcApiDecoderRecognition::incompatible(); + } + return kb_lib::DcApiDecoderRecognition::compatible( + self.exact, + self.priority, + std::option::Option::Some("surface_a".to_string()), + std::option::Option::Some("attempt".to_string()), + kb_lib::decoder_api_discriminator_8_hex(input), + ); + } + + fn decode( + &self, + input: &kb_lib::MdCoreInstructionReplayInput, + ) -> kb_lib::DcApiDecoderExecutionResult { + let observation = kb_lib::DcApiDecodedObservation { + event_key: "attempt".to_string(), + event: kb_lib::MdDecodedProtocolEvent { + signature: kb_lib::MdSignature(input.signature.clone()), + slot: kb_lib::MdSlot(input.slot), + instruction_path: kb_lib::MdInstructionPath(input.instruction_path.clone()), + program_id: kb_lib::MdProgramId(input.program_id.clone()), + protocol_code: kb_lib::MdProtocolCode("test".to_string()), + surface_code: kb_lib::MdSurfaceCode("surface_a".to_string()), + event_code: kb_lib::MdEventCode("attempt".to_string()), + event_name: kb_lib::MdEventName("attempt".to_string()), + event_family: kb_lib::MdEventFamily::Audit, + source_kind: kb_lib::MdEventSourceKind::Instruction, + confidence: kb_lib::MdDecoderConfidence::Exact, + }, + payload_json: serde_json::json!({"attempted": true}), + transaction_failed: input.transaction_failed, + transaction_error: input.transaction_err_json.clone(), + observation_committed: !input.transaction_failed, + proof: kb_lib::DcApiDecoderProof { + kind: kb_lib::DcApiDecoderProofKind::ExactLayout, + confidence: kb_lib::MdDecoderConfidence::Exact, + evidence: std::vec!["test".to_string()], + }, + }; + return kb_lib::DcApiDecoderExecutionResult { + status: kb_lib::DcApiDecoderOutcomeStatus::Decoded, + recognized_entry_code: std::option::Option::Some("attempt".to_string()), + observations: std::vec![observation], + diagnostics: std::vec::Vec::new(), + }; + } + } + + struct SelectiveAuditMaterializer; + + impl kb_lib::MtApiEventMaterializer for SelectiveAuditMaterializer { + fn identity(&self) -> kb_lib::MtApiMaterializerIdentity { + return kb_lib::MtApiMaterializerIdentity { + name: "selective_audit_materializer".to_string(), + version: "1".to_string(), + }; + } + + fn accepted_families(&self) -> &'static [kb_lib::MdEventFamily] { + return &[kb_lib::MdEventFamily::Audit]; + } + + fn accepts_observation(&self, observation: &kb_lib::DcApiDecodedObservation) -> bool { + return observation.event.surface_code.0.as_str() == "accepted_surface"; + } + + fn transaction_policy( + &self, + _family: kb_lib::MdEventFamily, + ) -> kb_lib::MtApiMaterializationTransactionPolicy { + return kb_lib::MtApiMaterializationTransactionPolicy::SuccessfulCommittedOnly; + } + + fn materialize( + &self, + _observation: &kb_lib::DcApiDecodedObservation, + ) -> kb_lib::MtApiMaterializerExecutionResult { + return kb_lib::MtApiMaterializerExecutionResult { + status: kb_lib::MtApiMaterializerOutcomeStatus::Failed, + outputs: std::vec::Vec::new(), + diagnostics: std::vec![kb_lib::MtApiMaterializerDiagnostic { + code: "unexpected_materializer_call".to_string(), + message: + "selective materializer must not be called for an unaccepted observation" + .to_string(), + retriable: false, + }], + }; + } + } + + struct TestObserver { + cancelled: std::sync::atomic::AtomicBool, + } + + impl crate::DecodeReplayObserver for TestObserver { + fn on_progress(&self, _event: &crate::DecodeReplayProgressEvent) {} + + fn is_cancelled(&self) -> bool { + return self.cancelled.load(std::sync::atomic::Ordering::SeqCst); + } + } + + #[derive(Default)] + struct TestStore { + inputs: std::sync::Mutex>, + filters: std::sync::Mutex>, + current: std::sync::Mutex>, + decode_bundles: std::sync::Mutex>, + materialization_bundles: + std::sync::Mutex>, + declarations: std::sync::Mutex>, + } + + #[async_trait::async_trait] + impl kb_store::DecodePipelineStore for TestStore { + #[expect( + clippy::implicit_return, + reason = "async_trait expansion triggers implicit_return on generated async trait methods." + )] + async fn list_decode_inputs( + &self, + filter: &kb_store::DecodeSelectionFilter, + ) -> kb_core::Result> { + super::tests::lock_or_panic(&self.filters).push(filter.clone()); + let limit_result = usize::try_from(filter.limit); + let limit = match limit_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_error) => usize::MAX, + }; + let inputs = super::tests::lock_or_panic(&self.inputs) + .iter() + .filter(|input| { + return (filter.signatures.is_empty() + || filter.signatures.contains(&input.signature)) + && (filter.program_ids.is_empty() + || filter.program_ids.contains(&input.program_id)) + && (filter.instruction_paths.is_empty() + || filter.instruction_paths.contains(&input.instruction_path)); + }) + .take(limit) + .cloned() + .collect::>(); + return std::result::Result::Ok(inputs); + } + + #[expect( + clippy::implicit_return, + reason = "async_trait expansion triggers implicit_return on generated async trait methods." + )] + async fn is_decode_current( + &self, + identity: &kb_store::ProcessingLedgerIdentity, + ) -> kb_core::Result { + let key = format!( + "{}:{}:{}:{}:{}", + identity.stage, + identity.processor_name, + identity.processor_version, + identity.input_key, + identity.input_hash + ); + return std::result::Result::Ok( + super::tests::lock_or_panic(&self.current).contains(&key), + ); + } + + #[expect( + clippy::implicit_return, + reason = "async_trait expansion triggers implicit_return on generated async trait methods." + )] + async fn persist_decode_coverage_declarations( + &self, + declarations: &[kb_store::DecodeCoverageDeclarationInsert], + ) -> kb_core::Result { + super::tests::lock_or_panic(&self.declarations).extend_from_slice(declarations); + return std::result::Result::Ok(kb_store::InsertOutcome::new( + declarations.len() as u64, + 0, + 0, + )); + } + + #[expect( + clippy::implicit_return, + reason = "async_trait expansion triggers implicit_return on generated async trait methods." + )] + async fn persist_decode_result( + &self, + bundle: &kb_store::DecodePersistenceBundle, + force_replay: bool, + ) -> kb_core::Result { + let key = format!( + "{}:{}:{}:{}:{}", + bundle.ledger_identity.stage, + bundle.ledger_identity.processor_name, + bundle.ledger_identity.processor_version, + bundle.ledger_identity.input_key, + bundle.ledger_identity.input_hash + ); + super::tests::lock_or_panic(&self.current).insert(key); + let mut bundles = super::tests::lock_or_panic(&self.decode_bundles); + if force_replay { + bundles.retain(|existing| { + return existing.ledger_identity.processor_name + != bundle.ledger_identity.processor_name + || existing.ledger_identity.processor_version + != bundle.ledger_identity.processor_version + || existing.ledger_identity.input_key != bundle.ledger_identity.input_key; + }); + } + bundles.push(bundle.clone()); + return std::result::Result::Ok(kb_store::InsertOutcome::new(1, 1, 0)); + } + + #[expect( + clippy::implicit_return, + reason = "async_trait expansion triggers implicit_return on generated async trait methods." + )] + async fn mark_decode_failed( + &self, + _failure: &kb_store::DecodeFailure, + ) -> kb_core::Result { + return std::result::Result::Ok(kb_store::InsertOutcome::new(0, 1, 0)); + } + + #[expect( + clippy::implicit_return, + reason = "async_trait expansion triggers implicit_return on generated async trait methods." + )] + async fn persist_materialization_result( + &self, + bundle: &kb_store::MaterializationPersistenceBundle, + force_replay: bool, + ) -> kb_core::Result { + let key = format!( + "{}:{}:{}:{}:{}", + bundle.ledger_identity.stage, + bundle.ledger_identity.processor_name, + bundle.ledger_identity.processor_version, + bundle.ledger_identity.input_key, + bundle.ledger_identity.input_hash + ); + super::tests::lock_or_panic(&self.current).insert(key); + let mut bundles = super::tests::lock_or_panic(&self.materialization_bundles); + if force_replay { + bundles.retain(|existing| { + return existing.ledger_identity.processor_name + != bundle.ledger_identity.processor_name + || existing.ledger_identity.processor_version + != bundle.ledger_identity.processor_version + || existing.ledger_identity.input_key != bundle.ledger_identity.input_key; + }); + } + bundles.push(bundle.clone()); + let count_result = u64::try_from(bundle.outputs.len()); + let count = match count_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + return std::result::Result::Err(kb_core::Error::invalid_state(format!( + "test materialization output count conversion failed: {error}" + ))); + }, + }; + return std::result::Result::Ok(kb_store::InsertOutcome::new(count, 1, 0)); + } + + #[expect( + clippy::implicit_return, + reason = "async_trait expansion triggers implicit_return on generated async trait methods." + )] + async fn list_decode_coverage_summary( + &self, + _processor_name: std::option::Option<&str>, + _processor_version: std::option::Option<&str>, + _limit: u32, + ) -> kb_core::Result> { + return std::result::Result::Ok(std::vec::Vec::new()); + } + + #[expect( + clippy::implicit_return, + reason = "async_trait expansion triggers implicit_return on generated async trait methods." + )] + async fn list_materialized_events( + &self, + _filter: &kb_store::MaterializedEventFilter, + ) -> kb_core::Result> { + return std::result::Result::Ok(std::vec::Vec::new()); + } + } + + fn replay_input(program_id: &str, failed: bool) -> kb_lib::MdCoreInstructionReplayInput { + let result = kb_lib::MdCoreInstructionReplayInput::new( + "signature:0", + "signature", + 42, + "0", + program_id, + failed, + if failed { + std::option::Option::Some(serde_json::json!({"InstructionError": [0, "Custom"]})) + } else { + std::option::Option::None + }, + serde_json::json!([]), + serde_json::json!([]), + std::option::Option::Some(serde_json::json!({"dataBase64": "AQIDBAUGBwg="})), + std::option::Option::Some("payload-hash".to_string()), + serde_json::json!([]), + serde_json::json!([]), + serde_json::json!([]), + serde_json::json!([]), + ); + return super::tests::result_or_panic(result); + } + + fn request(force_replay: bool) -> crate::DecodeReplayRequest { + return crate::DecodeReplayRequest { + campaign_id: "decode-test-campaign".to_string(), + selection: super::tests::result_or_panic(kb_store::DecodeSelectionFilter::actionable( + 10, + )), + decoder_names: std::vec::Vec::new(), + dispatch_policy: crate::DecodeDispatchPolicy::HighestPriority, + max_concurrent_inputs: 2, + force_replay, + force_replay_all_matching: force_replay, + materialize_after_decode: false, + }; + } + + fn memo_replay_input(failed: bool) -> kb_lib::MdCoreInstructionReplayInput { + let result = kb_lib::MdCoreInstructionReplayInput::new( + "memo-signature:0", + "memo-signature", + 84, + "0", + kb_program_ids::SPL_MEMO_V4_PROGRAM_ID, + failed, + if failed { + std::option::Option::Some(serde_json::json!({ + "InstructionError": [0, "InvalidInstructionData"] + })) + } else { + std::option::Option::None + }, + serde_json::json!([]), + serde_json::json!([]), + std::option::Option::Some(serde_json::json!({ + "dataBase64": "ZGVtbyBhbm5vdGF0aW9u" + })), + std::option::Option::Some("memo-payload-hash".to_string()), + serde_json::json!([]), + serde_json::json!([]), + serde_json::json!([]), + serde_json::json!([]), + ); + return super::tests::result_or_panic(result); + } + + #[test] + fn dispatch_is_exact_by_program_and_rank() { + let decoders: std::vec::Vec> = std::vec![ + std::sync::Arc::new(TestDecoder { + version: "1", + exact: false, + priority: 100 + }), + std::sync::Arc::new(TestDecoder { version: "2", exact: true, priority: 1 }), + ]; + let ranked = + super::rank_decoders("test-campaign", &replay_input("program_a", false), &decoders); + assert_eq!(ranked.len(), 2); + assert_eq!(ranked[0].identity.version, "2"); + assert!( + super::rank_decoders("test-campaign", &replay_input("unknown", false), &decoders) + .is_empty() + ); + } + + #[test] + fn decoded_observation_context_must_match_core_input() { + let input = replay_input("program_a", true); + let decoder = TestDecoder { version: "1", exact: true, priority: 1 }; + let mut execution = kb_lib::DcApiInstructionDecoder::decode(&decoder, &input); + execution.observations[0].event.slot = kb_lib::MdSlot(input.slot + 1); + let identity = kb_lib::DcApiInstructionDecoder::identity(&decoder); + let result = + super::observation_inserts(&identity, &input, "input-hash", &execution.observations); + assert!(result.is_err()); + } + + #[test] + fn duplicate_decoder_identity_is_rejected() { + let decoders: std::vec::Vec> = std::vec![ + std::sync::Arc::new(TestDecoder { version: "1", exact: true, priority: 1 }), + std::sync::Arc::new(TestDecoder { version: "1", exact: false, priority: 2 }), + ]; + let result = super::validate_decoder_registry(&decoders); + assert!(result.is_err()); + } + + #[test] + fn materializer_input_key_changes_with_decoder_version() { + let input = replay_input("program_a", false); + let version_one = kb_lib::DcApiDecoderIdentity { + name: "test_decoder".to_string(), + version: "1".to_string(), + }; + let version_two = kb_lib::DcApiDecoderIdentity { + name: "test_decoder".to_string(), + version: "2".to_string(), + }; + let first = super::materializer_input_key(&input, &version_one, "attempt"); + let second = super::materializer_input_key(&input, &version_two, "attempt"); + assert_ne!(first, second); + } + + #[test] + fn effective_selection_uses_enabled_decoder_programs() { + let decoders: std::vec::Vec> = + std::vec![std::sync::Arc::new(TestDecoder { version: "1", exact: true, priority: 1 })]; + let selection = super::tests::result_or_panic(super::effective_selection( + &request(false), + decoders.as_slice(), + )); + assert_eq!(selection.program_ids, std::vec!["program_a".to_string()]); + assert_eq!(selection.processing_states.len(), 3); + } + + #[test] + fn force_replay_with_explicit_signatures_bypasses_state_filter() { + let decoders: std::vec::Vec> = + std::vec![std::sync::Arc::new(TestDecoder { version: "1", exact: true, priority: 1 })]; + let mut replay_request = request(true); + replay_request.force_replay_all_matching = false; + replay_request.selection.signatures = std::vec!["signature".to_string()]; + let selection = super::tests::result_or_panic(super::effective_selection( + &replay_request, + decoders.as_slice(), + )); + assert!(selection.processing_states.is_empty()); + } + + #[test] + fn force_replay_without_explicit_scope_is_rejected() { + let mut replay_request = request(false); + replay_request.force_replay = true; + let result = replay_request.validate(); + assert!(result.is_err()); + } + + #[test] + fn all_matching_authorization_rejects_explicit_signatures() { + let mut replay_request = super::tests::request(true); + replay_request.selection.signatures = std::vec!["signature".to_string()]; + let result = replay_request.validate(); + assert!(result.is_err()); + } + + #[test] + fn force_replay_all_matching_bypasses_state_filter() { + let decoders: std::vec::Vec> = + std::vec![std::sync::Arc::new(TestDecoder { version: "1", exact: true, priority: 1 })]; + let replay_request = request(true); + let selection = super::tests::result_or_panic(super::effective_selection( + &replay_request, + decoders.as_slice(), + )); + assert!(selection.processing_states.is_empty()); + } + + #[tokio::test] + async fn observation_level_materializer_filter_avoids_false_refusals() { + let store = TestStore::default(); + super::tests::lock_or_panic(&store.inputs).push(replay_input("program_a", false)); + let decoders: std::vec::Vec> = + std::vec![std::sync::Arc::new(TestDecoder { version: "1", exact: true, priority: 1 })]; + let materializers: std::vec::Vec> = + std::vec![std::sync::Arc::new(SelectiveAuditMaterializer)]; + let observer = TestObserver { + cancelled: std::sync::atomic::AtomicBool::new(false), + }; + let mut replay_request = request(false); + replay_request.materialize_after_decode = true; + let summary = super::tests::result_or_panic( + crate::execute_decode_replay( + &store, + &replay_request, + &decoders, + &materializers, + &observer, + ) + .await, + ); + assert_eq!(summary.processors[0].decoded, 1); + assert_eq!(summary.processors[0].materialized_outputs, 0); + assert_eq!(summary.processors[0].materialization_refused, 0); + } + + #[tokio::test] + async fn materialization_without_materializer_is_rejected() { + let store = TestStore::default(); + super::tests::lock_or_panic(&store.inputs).push(replay_input("program_a", false)); + let decoders: std::vec::Vec> = + std::vec![std::sync::Arc::new(TestDecoder { version: "1", exact: true, priority: 1 })]; + let observer = TestObserver { + cancelled: std::sync::atomic::AtomicBool::new(false), + }; + let mut replay_request = request(false); + replay_request.materialize_after_decode = true; + let result = + crate::execute_decode_replay(&store, &replay_request, &decoders, &[], &observer).await; + assert!(result.is_err()); + } + + #[test] + fn explicit_unsupported_program_is_rejected_before_selection() { + let decoders: std::vec::Vec> = + std::vec![std::sync::Arc::new(TestDecoder { version: "1", exact: true, priority: 1 })]; + let mut replay_request = request(false); + replay_request.selection.program_ids = std::vec!["unrelated".to_string()]; + let result = super::effective_selection(&replay_request, decoders.as_slice()); + assert!(result.is_err()); + } + + #[tokio::test] + async fn automatic_program_scope_prevents_unrelated_unmatched_inputs() { + let store = TestStore::default(); + { + let mut inputs = super::tests::lock_or_panic(&store.inputs); + inputs.push(replay_input("program_a", false)); + inputs.push(replay_input("unrelated", false)); + } + let decoders: std::vec::Vec> = + std::vec![std::sync::Arc::new(TestDecoder { version: "1", exact: true, priority: 1 })]; + let observer = TestObserver { + cancelled: std::sync::atomic::AtomicBool::new(false), + }; + let summary = super::tests::result_or_panic( + crate::execute_decode_replay(&store, &request(false), &decoders, &[], &observer).await, + ); + assert_eq!(summary.selected, 1); + assert_eq!(summary.unmatched, 0); + assert_eq!(summary.processors[0].decoded, 1); + } + + #[tokio::test] + async fn same_version_and_hash_is_skipped() { + let store = TestStore::default(); + super::tests::lock_or_panic(&store.inputs).push(replay_input("program_a", false)); + let decoders: std::vec::Vec> = + std::vec![std::sync::Arc::new(TestDecoder { version: "1", exact: true, priority: 1 })]; + let observer = TestObserver { + cancelled: std::sync::atomic::AtomicBool::new(false), + }; + let first = super::tests::result_or_panic( + crate::execute_decode_replay(&store, &request(false), &decoders, &[], &observer).await, + ); + let second = super::tests::result_or_panic( + crate::execute_decode_replay(&store, &request(false), &decoders, &[], &observer).await, + ); + assert_eq!(first.processors[0].decoded, 1); + assert_eq!(second.processors[0].skipped, 1); + } + + #[tokio::test] + async fn version_change_replays_and_force_replaces_target_version_only() { + let store = TestStore::default(); + super::tests::lock_or_panic(&store.inputs).push(replay_input("program_a", false)); + let observer = TestObserver { + cancelled: std::sync::atomic::AtomicBool::new(false), + }; + let version_one: std::vec::Vec> = + std::vec![std::sync::Arc::new(TestDecoder { version: "1", exact: true, priority: 1 })]; + let version_two: std::vec::Vec> = + std::vec![std::sync::Arc::new(TestDecoder { version: "2", exact: true, priority: 1 })]; + super::tests::result_or_panic( + crate::execute_decode_replay(&store, &request(false), &version_one, &[], &observer) + .await, + ); + super::tests::result_or_panic( + crate::execute_decode_replay(&store, &request(false), &version_two, &[], &observer) + .await, + ); + super::tests::result_or_panic( + crate::execute_decode_replay(&store, &request(true), &version_two, &[], &observer) + .await, + ); + let bundles = super::tests::lock_or_panic(&store.decode_bundles); + assert_eq!(bundles.len(), 2); + assert!( + bundles + .iter() + .any(|bundle| return bundle.ledger_identity.processor_version == "1") + ); + assert!( + bundles + .iter() + .any(|bundle| return bundle.ledger_identity.processor_version == "2") + ); + } + + #[tokio::test] + async fn failed_transaction_observation_is_persisted_uncommitted() { + let store = TestStore::default(); + super::tests::lock_or_panic(&store.inputs).push(replay_input("program_a", true)); + let decoders: std::vec::Vec> = + std::vec![std::sync::Arc::new(TestDecoder { version: "1", exact: true, priority: 1 })]; + let observer = TestObserver { + cancelled: std::sync::atomic::AtomicBool::new(false), + }; + super::tests::result_or_panic( + crate::execute_decode_replay(&store, &request(false), &decoders, &[], &observer).await, + ); + let bundles = super::tests::lock_or_panic(&store.decode_bundles); + assert!(bundles[0].observations[0].transaction_failed); + assert!(!bundles[0].observations[0].observation_committed); + } + + #[tokio::test] + async fn concrete_memo_replay_materializes_only_committed_annotations_idempotently() { + let store = TestStore::default(); + super::tests::lock_or_panic(&store.inputs).push(memo_replay_input(false)); + let decoders: std::vec::Vec> = + std::vec![std::sync::Arc::new(kb_lib::DcSplMemoDecoder)]; + let materializers: std::vec::Vec> = + std::vec![std::sync::Arc::new(kb_lib::MtTransactionAnnotationMaterializer,)]; + let observer = TestObserver { + cancelled: std::sync::atomic::AtomicBool::new(false), + }; + let mut replay_request = request(false); + replay_request.materialize_after_decode = true; + let first = super::tests::result_or_panic( + crate::execute_decode_replay( + &store, + &replay_request, + &decoders, + &materializers, + &observer, + ) + .await, + ); + let second = super::tests::result_or_panic( + crate::execute_decode_replay( + &store, + &replay_request, + &decoders, + &materializers, + &observer, + ) + .await, + ); + assert_eq!(first.processors[0].decoded, 1); + assert_eq!(first.processors[0].materialized_outputs, 1); + assert_eq!(second.processors[0].skipped, 1); + { + let bundles = super::tests::lock_or_panic(&store.materialization_bundles); + assert_eq!(bundles.len(), 1); + assert_eq!(bundles[0].outputs.len(), 1); + assert_eq!(bundles[0].outputs[0].materialized_family, "transaction_annotation"); + assert_eq!(bundles[0].outputs[0].payload_json["text"], "demo annotation"); + } + + let failed_store = TestStore::default(); + super::tests::lock_or_panic(&failed_store.inputs).push(memo_replay_input(true)); + let failed = super::tests::result_or_panic( + crate::execute_decode_replay( + &failed_store, + &replay_request, + &decoders, + &materializers, + &observer, + ) + .await, + ); + assert_eq!(failed.processors[0].decoded, 1); + assert_eq!(failed.processors[0].materialized_outputs, 0); + assert_eq!(failed.processors[0].materialization_refused, 1); + } + + #[tokio::test] + async fn declared_and_observed_coverage_are_recorded() { + let store = TestStore::default(); + super::tests::lock_or_panic(&store.inputs).push(replay_input("program_a", false)); + let decoders: std::vec::Vec> = + std::vec![std::sync::Arc::new(TestDecoder { version: "1", exact: true, priority: 1 })]; + let observer = TestObserver { + cancelled: std::sync::atomic::AtomicBool::new(false), + }; + super::tests::result_or_panic( + crate::execute_decode_replay(&store, &request(false), &decoders, &[], &observer).await, + ); + let declarations = super::tests::lock_or_panic(&store.declarations); + assert_eq!(declarations.len(), 1); + assert_eq!(declarations[0].entry_code, "attempt"); + let bundles = super::tests::lock_or_panic(&store.decode_bundles); + assert_eq!(bundles.len(), 1); + assert_eq!(bundles[0].coverage.entry_code.as_deref(), std::option::Option::Some("attempt")); + assert!(bundles[0].coverage.recognized); + assert_eq!(bundles[0].coverage.decoded_count, 1); + assert_eq!(bundles[0].coverage.status, "decoded"); + } + + #[tokio::test] + async fn cancellation_leaves_candidates_not_started() { + let store = TestStore::default(); + super::tests::lock_or_panic(&store.inputs).push(replay_input("program_a", false)); + let decoders: std::vec::Vec> = + std::vec![std::sync::Arc::new(TestDecoder { version: "1", exact: true, priority: 1 })]; + let observer = TestObserver { + cancelled: std::sync::atomic::AtomicBool::new(true), + }; + let summary = super::tests::result_or_panic( + crate::execute_decode_replay(&store, &request(false), &decoders, &[], &observer).await, + ); + assert_eq!(summary.selected, 1); + assert_eq!(summary.not_started, 1); + assert_eq!(summary.started, 0); + assert_eq!(summary.completed, 0); + } +} diff --git a/kb-pipeline/src/lib.rs b/kb-pipeline/src/lib.rs index 9a367fb..f0d7520 100644 --- a/kb-pipeline/src/lib.rs +++ b/kb-pipeline/src/lib.rs @@ -1,5 +1,5 @@ // file: kb-pipeline/src/lib.rs -// version: 3 +// version: 4 #![forbid(unsafe_code)] #![deny(unreachable_pub)] @@ -9,6 +9,7 @@ mod constants; mod core_extraction; +mod decode_replay; mod plan; /// Canonical tracing target for pipeline orchestration. @@ -35,6 +36,30 @@ pub use self::core_extraction::CoreExtractionSummary; pub use self::core_extraction::execute_core_extraction; /// Extracts one canonical raw transaction into normalized core rows. pub use self::core_extraction::extract_raw_transaction_to_core; +/// Current common decode pipeline orchestration version. +pub use self::decode_replay::DECODE_PIPELINE_VERSION; +/// Deterministic decoder dispatch policy. +pub use self::decode_replay::DecodeDispatchPolicy; +/// Per-decoder terminal counters. +pub use self::decode_replay::DecodeProcessorSummary; +/// Decode replay observer contract. +pub use self::decode_replay::DecodeReplayObserver; +/// Decode replay progress event. +pub use self::decode_replay::DecodeReplayProgressEvent; +/// Decode replay progress severity. +pub use self::decode_replay::DecodeReplayProgressLevel; +/// Bounded contextual decode replay request. +pub use self::decode_replay::DecodeReplayRequest; +/// Final contextual decode replay summary. +pub use self::decode_replay::DecodeReplaySummary; +/// Stable materialization processing stage. +pub use self::decode_replay::EVENT_MATERIALIZATION_STAGE; +/// Stable instruction decode processing stage. +pub use self::decode_replay::INSTRUCTION_DECODE_STAGE; +/// Executes one contextual decode replay campaign. +pub use self::decode_replay::execute_decode_replay; +/// Creates one stable process-local decode campaign identifier. +pub use self::decode_replay::new_decode_campaign_id; /// Pipeline stage identifier. pub use self::plan::PipelineStage; /// Replay selection scope.