v0.5.3-pre.003

ks-pipeline/CHANGELOG.md

@@ -1,8 +1,16 @@
 <!-- file: ks-pipeline/CHANGELOG.md -->
-<!-- version: 35 -->
+<!-- version: 37 -->
 
 # CHANGELOG — ks-pipeline
 
+## `0.5.3-pre.003`
+
+- corrige le premier delta `pre.003` en utilisant le nom canonique exporté `MdCanonicalInnerInstructionGroup` pour la reconstruction des chemins CPI.
+- aligne Core extraction sur le nouveau baseline `ks-store` : conservation de `block_time`, `stack_height`, `returnData` et persistance du contexte N2 nécessaire au replay ;
+- reconstruit les chemins CPI imbriqués à partir de `stackHeight` lorsque l'information est disponible et conserve un fallback déterministe vers la top-level lorsque Solana ne fournit pas cette hauteur ;
+- enrichit les inputs N2 -> N3 avec le contexte Core v3 sans modifier le dispatch fonctionnel des décodeurs actuels ;
+- adapte les lectures de matérialisation au contrat `MaterializedOutput*` renommé avec `k_sol_mat_outputs`.
+
 ## `0.5.1-pre.006`
 
 - déplace ses fixtures de configuration vers le contrat runtime résolu sous `test-fixtures/config/` ;

ks-pipeline/src/backfill.rs

@@ -1,5 +1,5 @@
 // file: ks-pipeline/src/backfill.rs
-// version: 11
+// version: 12
 
 //! Reusable HTTP transaction backfill orchestration.
 
@@ -1409,7 +1409,7 @@ where
         observation_with_timings.with_payload_metadata(payload_size_bytes, source_payload_hash);
     let observation = match payload_result {
         std::result::Result::Ok(value) => {
-            value.with_status(ks_store::TransactionObservationStatus::Persisted)
+            value.with_status(ks_store::AcquisitionObservationStatus::Persisted)
         },
         std::result::Result::Err(error) => return std::result::Result::Err(error),
     };
@@ -1456,7 +1456,7 @@ where
         observation_with_timings.with_payload_metadata(payload_size_bytes, source_payload_hash);
     let observation = match payload_result {
         std::result::Result::Ok(value) => {
-            value.with_status(ks_store::TransactionObservationStatus::Missing)
+            value.with_status(ks_store::AcquisitionObservationStatus::Missing)
         },
         std::result::Result::Err(error) => return std::result::Result::Err(error),
     };
@@ -1538,7 +1538,7 @@ fn base_observation(
         client.provider().to_string(),
         "solana_http_json_rpc".to_string(),
         "getTransaction".to_string(),
-        ks_store::TransactionObservationOrigin::Backfill,
+        ks_store::AcquisitionObservationOrigin::Backfill,
         received_at,
     );
     let observation = match observation_result {

ks-pipeline/src/core_extraction.rs

@@ -1,5 +1,5 @@
 // file: ks-pipeline/src/core_extraction.rs
-// version: 11
+// version: 13
 
 //! Canonical Solana transaction to normalized core extraction pipeline.
 
@@ -568,6 +568,11 @@ pub fn extract_raw_transaction_to_core(
             "canonical transaction slot does not match raw row slot",
         ));
     }
+    if transaction.block_time != row.block_time {
+        return std::result::Result::Err(ks_core::Error::invalid_state(
+            "canonical transaction block time does not match raw row block time",
+        ));
+    }
     let computed_hash_result = transaction.canonical_json_hash();
     let computed_hash = match computed_hash_result {
         std::result::Result::Ok(value) => value,
@@ -594,6 +599,7 @@ pub fn extract_raw_transaction_to_core(
     let core_transaction_result = ks_store::CoreTransactionInsert::new(
         transaction.primary_signature.clone(),
         transaction.slot,
+        transaction.block_time,
         false,
         std::option::Option::None,
     );
@@ -620,6 +626,7 @@ pub fn extract_raw_transaction_to_core(
         inner_instructions,
         logs: std::vec::Vec::new(),
         balance_changes: std::vec::Vec::new(),
+        return_data: std::option::Option::None,
     });
 }
 
@@ -633,6 +640,7 @@ fn build_bundle_with_metadata(
     let core_transaction_result = ks_store::CoreTransactionInsert::new(
         transaction.primary_signature.clone(),
         transaction.slot,
+        transaction.block_time,
         failed,
         metadata.error.clone(),
     );
@@ -660,6 +668,11 @@ fn build_bundle_with_metadata(
         std::result::Result::Ok(value) => value,
         std::result::Result::Err(error) => return std::result::Result::Err(error),
     };
+    let return_data_result = crate::core_extraction::extract_return_data(transaction, metadata);
+    let return_data = match return_data_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
     return std::result::Result::Ok(ks_store::CoreExtractionBundle {
         raw_transaction_id: row.id,
         ledger_identity,
@@ -669,9 +682,32 @@ fn build_bundle_with_metadata(
         inner_instructions,
         logs,
         balance_changes,
+        return_data,
     });
 }
 
+fn extract_return_data(
+    transaction: &ks_lib::MdCanonicalTransaction,
+    metadata: &ks_lib::MdCanonicalTransactionMetadata,
+) -> ks_core::Result<std::option::Option<ks_store::CoreReturnDataInsert>> {
+    let value = match &metadata.return_data {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return std::result::Result::Ok(std::option::Option::None),
+    };
+    let insert_result = ks_store::CoreReturnDataInsert::new(
+        transaction.primary_signature.clone(),
+        transaction.slot,
+        value.program_id.clone(),
+        value.data_base64.clone(),
+    );
+    return match insert_result {
+        std::result::Result::Ok(insert) => {
+            std::result::Result::Ok(std::option::Option::Some(insert))
+        },
+        std::result::Result::Err(error) => std::result::Result::Err(error),
+    };
+}
+
 fn ledger_identity_from_raw(
     row: &ks_store::RawTransactionRow,
 ) -> ks_core::Result<ks_store::ProcessingLedgerIdentity> {
@@ -819,6 +855,7 @@ fn extract_instructions(
             transaction.slot,
             path,
             program_id,
+            instruction.stack_height,
             accounts_json,
             payload_json,
         );
@@ -836,9 +873,12 @@ fn extract_instructions(
     let mut inner_instructions = std::vec::Vec::new();
     if let std::option::Option::Some(metadata) = &transaction.metadata {
         for group in &metadata.inner_instructions {
-            let parent_path = group.parent_instruction_index.to_string();
-            for (inner_index, instruction) in group.instructions.iter().enumerate() {
-                let path = format!("{parent_path}/{inner_index}");
+            let paths_result = crate::core_extraction::resolve_inner_instruction_paths(group);
+            let paths = match paths_result {
+                std::result::Result::Ok(value) => value,
+                std::result::Result::Err(error) => return std::result::Result::Err(error),
+            };
+            for (instruction, path) in group.instructions.iter().zip(paths.iter()) {
                 let resolved_result =
                     crate::core_extraction::resolve_instruction(&keys, instruction);
                 let (program_id, accounts_json, payload_json, payload_hash) = match resolved_result
@@ -849,9 +889,10 @@ fn extract_instructions(
                 let insert_result = ks_store::CoreInnerInstructionInsert::new(
                     transaction.primary_signature.clone(),
                     transaction.slot,
-                    parent_path.clone(),
-                    path,
+                    path.0.clone(),
+                    path.1.clone(),
                     program_id,
+                    instruction.stack_height,
                     accounts_json,
                     payload_json,
                 );
@@ -871,6 +912,57 @@ fn extract_instructions(
     return std::result::Result::Ok((instructions, inner_instructions));
 }
 
+fn resolve_inner_instruction_paths(
+    group: &ks_lib::MdCanonicalInnerInstructionGroup,
+) -> ks_core::Result<std::vec::Vec<(std::string::String, std::string::String)>> {
+    let root_path = group.parent_instruction_index.to_string();
+    let mut stack = std::vec::Vec::<(u32, std::string::String)>::new();
+    let mut child_counts = std::collections::BTreeMap::<std::string::String, u32>::new();
+    let mut output = std::vec::Vec::with_capacity(group.instructions.len());
+    for instruction in &group.instructions {
+        let parent_path = match instruction.stack_height {
+            std::option::Option::Some(height) if height <= 1 => {
+                return std::result::Result::Err(ks_core::Error::invalid_state(
+                    "inner instruction stack height must be greater than one",
+                ));
+            },
+            std::option::Option::Some(2) => {
+                stack.clear();
+                root_path.clone()
+            },
+            std::option::Option::Some(height) => {
+                while stack.last().is_some_and(|entry| return entry.0 >= height) {
+                    stack.pop();
+                }
+                match stack.last() {
+                    std::option::Option::Some((parent_height, path))
+                        if parent_height.saturating_add(1) == height =>
+                    {
+                        path.clone()
+                    },
+                    _ => {
+                        return std::result::Result::Err(ks_core::Error::invalid_state(
+                            "inner instruction stack height is inconsistent with invocation order",
+                        ));
+                    },
+                }
+            },
+            std::option::Option::None => {
+                stack.clear();
+                root_path.clone()
+            },
+        };
+        let child_index = child_counts.entry(parent_path.clone()).or_insert(0);
+        let instruction_path = format!("{parent_path}/{}", *child_index);
+        *child_index = child_index.saturating_add(1);
+        if let std::option::Option::Some(height) = instruction.stack_height {
+            stack.push((height, instruction_path.clone()));
+        }
+        output.push((parent_path, instruction_path));
+    }
+    return std::result::Result::Ok(output);
+}
+
 fn resolve_instruction(
     keys: &[std::string::String],
     instruction: &ks_lib::MdCanonicalCompiledInstruction,
@@ -918,7 +1010,11 @@ fn extract_logs(
     metadata: &ks_lib::MdCanonicalTransactionMetadata,
 ) -> ks_core::Result<std::vec::Vec<ks_store::CoreLogInsert>> {
     let mut output = std::vec::Vec::with_capacity(metadata.log_messages.len());
-    let links = crate::core_extraction::resolve_log_links(transaction, metadata);
+    let links_result = crate::core_extraction::resolve_log_links(transaction, metadata);
+    let links = match links_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
     for (index, text) in metadata.log_messages.iter().enumerate() {
         let log_index_result = u32::try_from(index);
         let log_index = match log_index_result {
@@ -959,10 +1055,12 @@ fn extract_logs(
 fn resolve_log_links(
     transaction: &ks_lib::MdCanonicalTransaction,
     metadata: &ks_lib::MdCanonicalTransactionMetadata,
-) -> std::vec::Vec<(
-    std::option::Option<std::string::String>,
-    std::option::Option<std::string::String>,
-)> {
+) -> ks_core::Result<
+    std::vec::Vec<(
+        std::option::Option<std::string::String>,
+        std::option::Option<std::string::String>,
+    )>,
+> {
     let keys = crate::core_extraction::resolved_account_keys(transaction);
     let mut top_level = std::vec::Vec::<(std::string::String, std::string::String)>::new();
     for (index, instruction) in transaction.message.instructions.iter().enumerate() {
@@ -972,18 +1070,23 @@ fn resolve_log_links(
             top_level.push((program_id.clone(), index.to_string()));
         }
     }
-    let mut inner_by_parent = std::collections::BTreeMap::<
+    let mut inner_by_root = std::collections::BTreeMap::<
         std::string::String,
-        std::vec::Vec<(std::string::String, std::string::String)>,
+        std::vec::Vec<(std::string::String, std::string::String, std::option::Option<u32>)>,
     >::new();
     for group in &metadata.inner_instructions {
-        let parent = group.parent_instruction_index.to_string();
-        let entries = inner_by_parent.entry(parent.clone()).or_default();
-        for (index, instruction) in group.instructions.iter().enumerate() {
+        let root = group.parent_instruction_index.to_string();
+        let paths_result = crate::core_extraction::resolve_inner_instruction_paths(group);
+        let paths = match paths_result {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        let entries = inner_by_root.entry(root).or_default();
+        for (instruction, path) in group.instructions.iter().zip(paths.iter()) {
             if let std::option::Option::Some(program_id) =
                 keys.get(usize::from(instruction.program_id_index))
             {
-                entries.push((program_id.clone(), format!("{parent}/{index}")));
+                entries.push((program_id.clone(), path.1.clone(), instruction.stack_height));
             }
         }
     }
@@ -1011,18 +1114,20 @@ fn resolve_log_links(
                     }
                     resolved
                 } else {
-                    let parent_path = stack.first().and_then(|entry| return entry.1.clone());
-                    match parent_path {
-                        std::option::Option::Some(parent) => {
-                            let cursor = inner_cursors.entry(parent.clone()).or_insert(0);
+                    let root_path = stack.first().and_then(|entry| return entry.1.clone());
+                    match root_path {
+                        std::option::Option::Some(root) => {
+                            let cursor = inner_cursors.entry(root.clone()).or_insert(0);
                             let mut resolved = std::option::Option::None;
                             if let std::option::Option::Some(entries) =
-                                inner_by_parent.get(parent.as_str())
+                                inner_by_root.get(root.as_str())
                             {
                                 while *cursor < entries.len() {
                                     let candidate = &entries[*cursor];
                                     *cursor += 1;
-                                    if candidate.0 == program_id {
+                                    let depth_matches =
+                                        candidate.2.is_none_or(|height| return height == depth);
+                                    if candidate.0 == program_id && depth_matches {
                                         resolved = std::option::Option::Some(candidate.1.clone());
                                         break;
                                     }
@@ -1067,7 +1172,7 @@ fn resolve_log_links(
             },
         }
     }
-    return output;
+    return std::result::Result::Ok(output);
 }
 
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
@@ -1570,6 +1675,7 @@ mod tests {
             id: 1,
             signature: transaction.primary_signature.clone(),
             slot: 42,
+            block_time: transaction.block_time,
             canonical_json: std::option::Option::Some(json),
             canonical_json_hash: std::option::Option::Some(hash),
             canonical_format_version: 1,
@@ -1598,6 +1704,105 @@ mod tests {
         assert_eq!(bundle.logs[1].instruction_path.as_deref(), std::option::Option::Some("0/0"));
     }
 
+    #[test]
+    fn nested_cpi_paths_preserve_immediate_parent_from_stack_height() {
+        let mut transaction = crate::core_extraction::tests::sample_transaction();
+        let metadata = match transaction.metadata.as_mut() {
+            std::option::Option::Some(value) => value,
+            std::option::Option::None => panic!("sample metadata is required"),
+        };
+        let template = metadata.inner_instructions[0].instructions[0].clone();
+        let mut direct = template.clone();
+        direct.stack_height = std::option::Option::Some(2);
+        let mut nested = template.clone();
+        nested.stack_height = std::option::Option::Some(3);
+        let mut sibling = template;
+        sibling.stack_height = std::option::Option::Some(2);
+        metadata.inner_instructions[0].instructions = std::vec![direct, nested, sibling];
+        let extraction_result = crate::core_extraction::extract_instructions(&transaction);
+        let (_top_level, inner) = match extraction_result {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => panic!("nested CPI extraction failed: {error}"),
+        };
+        assert_eq!(inner[0].parent_instruction_path, "0");
+        assert_eq!(inner[0].instruction_path, "0/0");
+        assert_eq!(inner[1].parent_instruction_path, "0/0");
+        assert_eq!(inner[1].instruction_path, "0/0/0");
+        assert_eq!(inner[2].parent_instruction_path, "0");
+        assert_eq!(inner[2].instruction_path, "0/1");
+        assert_eq!(inner[1].stack_height, std::option::Option::Some(3));
+    }
+
+    #[test]
+    fn inconsistent_cpi_stack_height_is_rejected() {
+        let mut transaction = crate::core_extraction::tests::sample_transaction();
+        let metadata = match transaction.metadata.as_mut() {
+            std::option::Option::Some(value) => value,
+            std::option::Option::None => panic!("sample metadata is required"),
+        };
+        let template = metadata.inner_instructions[0].instructions[0].clone();
+        let mut direct = template.clone();
+        direct.stack_height = std::option::Option::Some(2);
+        let mut skipped_parent = template;
+        skipped_parent.stack_height = std::option::Option::Some(4);
+        metadata.inner_instructions[0].instructions = std::vec![direct, skipped_parent];
+        let extraction_result = crate::core_extraction::extract_instructions(&transaction);
+        assert!(extraction_result.is_err());
+    }
+
+    #[test]
+    fn absent_cpi_stack_height_falls_back_to_top_level_parent() {
+        let mut transaction = crate::core_extraction::tests::sample_transaction();
+        let metadata = match transaction.metadata.as_mut() {
+            std::option::Option::Some(value) => value,
+            std::option::Option::None => panic!("sample metadata is required"),
+        };
+        metadata.inner_instructions[0].instructions[0].stack_height = std::option::Option::None;
+        let extraction_result = crate::core_extraction::extract_instructions(&transaction);
+        let (_top_level, inner) = match extraction_result {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => panic!("CPI extraction failed: {error}"),
+        };
+        assert_eq!(inner[0].parent_instruction_path, "0");
+        assert_eq!(inner[0].instruction_path, "0/0");
+        assert_eq!(inner[0].stack_height, std::option::Option::None);
+    }
+
+    #[test]
+    fn extraction_preserves_block_time_and_return_data() {
+        let mut transaction = crate::core_extraction::tests::sample_transaction();
+        let metadata = match transaction.metadata.as_mut() {
+            std::option::Option::Some(value) => value,
+            std::option::Option::None => panic!("sample metadata is required"),
+        };
+        metadata.return_data = std::option::Option::Some(ks_lib::MdCanonicalReturnData {
+            program_id: "11111111111111111111111111111111".to_string(),
+            data_base64: "AQID".to_string(),
+        });
+        let row = crate::core_extraction::tests::raw_row(&transaction);
+        let extraction_result = crate::extract_raw_transaction_to_core(&row);
+        let bundle = match extraction_result {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => panic!("Core extraction failed: {error}"),
+        };
+        assert_eq!(bundle.transaction.block_time, transaction.block_time);
+        let return_data = match bundle.return_data {
+            std::option::Option::Some(value) => value,
+            std::option::Option::None => panic!("return data must be preserved"),
+        };
+        assert_eq!(return_data.program_id, "11111111111111111111111111111111");
+        assert_eq!(return_data.data_base64, "AQID");
+    }
+
+    #[test]
+    fn extraction_rejects_block_time_mismatch() {
+        let transaction = crate::core_extraction::tests::sample_transaction();
+        let mut row = crate::core_extraction::tests::raw_row(&transaction);
+        row.block_time = std::option::Option::Some(1_700_000_001);
+        let result = crate::extract_raw_transaction_to_core(&row);
+        assert!(result.is_err());
+    }
+
     #[test]
     fn extraction_rejects_hash_mismatch() {
         let transaction = crate::core_extraction::tests::sample_transaction();

ks-pipeline/src/decode_replay.rs

@@ -1,5 +1,5 @@
 // file: ks-pipeline/src/decode_replay.rs
-// version: 3
+// version: 4
 
 //! Common contextual instruction decode and optional materialization pipeline.
 
@@ -1320,6 +1320,7 @@ fn observation_inserts(
             proof_kind: proof_kind_code(observation.proof.kind).to_string(),
             proof_json,
             payload_json: observation.payload_json.clone(),
+            schema_provenance: std::option::Option::None,
             transaction_failed: observation.transaction_failed,
             transaction_error: observation.transaction_error.clone(),
             observation_committed: observation.observation_committed,
@@ -1821,10 +1822,10 @@ mod tests {
             clippy::implicit_return,
             reason = "async_trait expansion triggers implicit_return on generated async trait methods."
         )]
-        async fn list_materialized_events(
+        async fn list_materialized_outputs(
             &self,
-            _filter: &ks_store::MaterializedEventFilter,
-        ) -> ks_core::Result<std::vec::Vec<ks_store::MaterializedEventQueryRow>> {
+            _filter: &ks_store::MaterializedOutputFilter,
+        ) -> ks_core::Result<std::vec::Vec<ks_store::MaterializedOutputQueryRow>> {
             return std::result::Result::Ok(std::vec::Vec::new());
         }
     }

ks-pipeline/src/plan.rs

@@ -1,5 +1,5 @@
 // file: ks-pipeline/src/plan.rs
-// version: 3
+// version: 4
 
 //! Pipeline planning primitives.
 
@@ -16,7 +16,7 @@ pub enum PipelineStage {
     Decode,
     /// Materialize business events.
     Materialize,
-    /// Aggregate materialized events.
+    /// Aggregate materialized outputs.
     Aggregate,
     /// Validate outputs.
     Validate,