v0.3.12-pre.005

2026-09-09 12:45:29 +02:00
parent df1dbbb8fb
commit 3a8fa87acb
9 changed files with 846 additions and 11 deletions
--- a/crates/ksp-worker-raw-transaction-ingest-lib/src/lib.rs
+++ b/crates/ksp-worker-raw-transaction-ingest-lib/src/lib.rs
@@ -1,5 +1,5 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/src/lib.rs
-// version: 11
+// version: 12

 #![warn(missing_docs)]
 #![deny(unreachable_pub)]
@@ -12,8 +12,9 @@
 //! owns bounded source-neutral admission, common RAW canonicalization/assembly and backend-neutral
 //! Store persistence in normal mode plus concrete latest-value snapshots projected onto Worker API;
 //! the first validated Yellowstone/HTTP runtime-resource contract also closes the deterministic hydration
-//! qualification path in test configuration. Transaction/TransactionStatus signal projection and HTTP hydration
-//! remain inactive in production until the productive source task is wired; no live stream is opened yet.
+//! qualification path in test configuration. Transaction/TransactionStatus/Block signal projection, BlockMeta/Slot
+//! continuity projection and HTTP hydration remain inactive in production until the productive source task is wired;
+//! no live stream is opened yet.

 mod admission;
 mod error;
--- a/crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs
+++ b/crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs
@@ -1,5 +1,5 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs
-// version: 5
+// version: 6

 #[cfg(test)]
 use sha2::Digest; // rust-rules: trait-import
@@ -14,10 +14,64 @@ const RAW_TRANSACTION_INGEST_YELLOWSTONE_HTTP_SOURCE_KEY_DOMAIN: &[u8] = b"ksp.r
 #[cfg(test)]
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 enum RawTransactionIngestSourceFamily {
+    Block,
    Transaction,
    TransactionStatus,
 }

+#[cfg(test)]
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+enum RawTransactionIngestContinuityFamily {
+    BlockMeta,
+    Slot,
+}
+
+#[cfg(test)]
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+enum RawTransactionIngestContinuityStatus {
+    Completed,
+    Confirmed,
+    CreatedBank,
+    Dead,
+    Finalized,
+    FirstShredReceived,
+    Processed,
+}
+
+#[cfg(test)]
+#[derive(Clone, Eq, PartialEq)]
+struct RawTransactionIngestContinuitySignal {
+    created_at: std::option::Option<RawTransactionIngestSourceTimestamp>,
+    family: RawTransactionIngestContinuityFamily,
+    matched_filter_count: usize,
+    matched_filter_fingerprint: [u8; 32],
+    matched_filter_id: std::option::Option<ksp_store_lib::RawProvenanceCode>,
+    network: ksp_store_lib::RawNetworkId,
+    parent_slot: std::option::Option<u64>,
+    route: RawTransactionIngestSourceRoute,
+    slot: u64,
+    status: std::option::Option<RawTransactionIngestContinuityStatus>,
+}
+
+#[cfg(test)]
+impl std::fmt::Debug for RawTransactionIngestContinuitySignal {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        return formatter
+            .debug_struct("RawTransactionIngestContinuitySignal")
+            .field("created_at", &self.created_at)
+            .field("family", &self.family)
+            .field("matched_filter_count", &self.matched_filter_count)
+            .field("matched_filter_fingerprint_bytes", &self.matched_filter_fingerprint.len())
+            .field("has_direct_filter_id", &self.matched_filter_id.is_some())
+            .field("network", &self.network)
+            .field("parent_slot", &self.parent_slot)
+            .field("route", &self.route)
+            .field("slot", &self.slot)
+            .field("status", &self.status)
+            .finish_non_exhaustive();
+    }
+}
+
 #[derive(Clone, Debug, Eq, PartialEq)]
 struct RawTransactionIngestSourceRoute {
    endpoint_id: ksp_store_lib::RawProvenanceCode,
@@ -141,6 +195,115 @@ impl RawTransactionIngestYellowstoneSignalView for ksp_onchain_transport_lib::Ye
    }
 }

+#[cfg(test)]
+trait RawTransactionIngestYellowstoneBlockView {
+    fn created_at(&self) -> std::option::Option<ksp_onchain_transport_lib::YellowstoneUpdateTimestamp>;
+
+    fn filters(&self) -> &[ksp_onchain_transport_lib::YellowstoneSubscribeFilterName];
+
+    fn slot(&self) -> u64;
+
+    fn transaction_count(&self) -> usize;
+
+    fn transaction_identity(&self, position: usize) -> ksp_core_lib::Result<(ksp_onchain_transport_lib::YellowstoneTransactionSignature, u64)>;
+}
+
+#[cfg(test)]
+impl RawTransactionIngestYellowstoneBlockView for ksp_onchain_transport_lib::YellowstoneBlockUpdate {
+    fn created_at(&self) -> std::option::Option<ksp_onchain_transport_lib::YellowstoneUpdateTimestamp> {
+        return ksp_onchain_transport_lib::YellowstoneBlockUpdate::created_at(self);
+    }
+
+    fn filters(&self) -> &[ksp_onchain_transport_lib::YellowstoneSubscribeFilterName] {
+        return ksp_onchain_transport_lib::YellowstoneBlockUpdate::filters(self);
+    }
+
+    fn slot(&self) -> u64 {
+        return ksp_onchain_transport_lib::YellowstoneBlockUpdate::slot(self);
+    }
+
+    fn transaction_count(&self) -> usize {
+        return ksp_onchain_transport_lib::YellowstoneBlockUpdate::transactions(self).len();
+    }
+
+    fn transaction_identity(&self, position: usize) -> ksp_core_lib::Result<(ksp_onchain_transport_lib::YellowstoneTransactionSignature, u64)> {
+        let transaction = match ksp_onchain_transport_lib::YellowstoneBlockUpdate::transactions(self).get(position) {
+            std::option::Option::Some(value) => value,
+            std::option::Option::None => return std::result::Result::Err(crate::runtime_error("source.block_transaction_missing")),
+        };
+        return std::result::Result::Ok((transaction.signature(), transaction.index()));
+    }
+}
+
+#[cfg(test)]
+trait RawTransactionIngestYellowstoneContinuityView {
+    fn created_at(&self) -> std::option::Option<ksp_onchain_transport_lib::YellowstoneUpdateTimestamp>;
+
+    fn family(&self) -> RawTransactionIngestContinuityFamily;
+
+    fn filters(&self) -> &[ksp_onchain_transport_lib::YellowstoneSubscribeFilterName];
+
+    fn parent_slot(&self) -> std::option::Option<u64>;
+
+    fn slot(&self) -> u64;
+
+    fn status(&self) -> std::option::Option<RawTransactionIngestContinuityStatus>;
+}
+
+#[cfg(test)]
+impl RawTransactionIngestYellowstoneContinuityView for ksp_onchain_transport_lib::YellowstoneBlockMetaUpdate {
+    fn created_at(&self) -> std::option::Option<ksp_onchain_transport_lib::YellowstoneUpdateTimestamp> {
+        return ksp_onchain_transport_lib::YellowstoneBlockMetaUpdate::created_at(self);
+    }
+
+    fn family(&self) -> RawTransactionIngestContinuityFamily {
+        return RawTransactionIngestContinuityFamily::BlockMeta;
+    }
+
+    fn filters(&self) -> &[ksp_onchain_transport_lib::YellowstoneSubscribeFilterName] {
+        return ksp_onchain_transport_lib::YellowstoneBlockMetaUpdate::filters(self);
+    }
+
+    fn parent_slot(&self) -> std::option::Option<u64> {
+        return std::option::Option::Some(ksp_onchain_transport_lib::YellowstoneBlockMetaUpdate::parent_slot(self));
+    }
+
+    fn slot(&self) -> u64 {
+        return ksp_onchain_transport_lib::YellowstoneBlockMetaUpdate::slot(self);
+    }
+
+    fn status(&self) -> std::option::Option<RawTransactionIngestContinuityStatus> {
+        return std::option::Option::None;
+    }
+}
+
+#[cfg(test)]
+impl RawTransactionIngestYellowstoneContinuityView for ksp_onchain_transport_lib::YellowstoneSlotUpdate {
+    fn created_at(&self) -> std::option::Option<ksp_onchain_transport_lib::YellowstoneUpdateTimestamp> {
+        return ksp_onchain_transport_lib::YellowstoneSlotUpdate::created_at(self);
+    }
+
+    fn family(&self) -> RawTransactionIngestContinuityFamily {
+        return RawTransactionIngestContinuityFamily::Slot;
+    }
+
+    fn filters(&self) -> &[ksp_onchain_transport_lib::YellowstoneSubscribeFilterName] {
+        return ksp_onchain_transport_lib::YellowstoneSlotUpdate::filters(self);
+    }
+
+    fn parent_slot(&self) -> std::option::Option<u64> {
+        return ksp_onchain_transport_lib::YellowstoneSlotUpdate::parent(self);
+    }
+
+    fn slot(&self) -> u64 {
+        return ksp_onchain_transport_lib::YellowstoneSlotUpdate::slot(self);
+    }
+
+    fn status(&self) -> std::option::Option<RawTransactionIngestContinuityStatus> {
+        return std::option::Option::Some(map_yellowstone_slot_status(ksp_onchain_transport_lib::YellowstoneSlotUpdate::status(self)));
+    }
+}
+
 #[cfg(test)]
 impl std::convert::From<(&crate::RawTransactionIngestYellowstoneSource, &ksp_onchain_transport_lib::YellowstoneTransactionUpdate)>
    for RawTransactionIngestSourceSignal
@@ -234,6 +397,8 @@ impl std::fmt::Debug for crate::RawTransactionIngestYellowstoneSource {
            .field("transaction_filter_count", &self.subscribe_request.transaction_filter_count())
            .field("transaction_status_filter_count", &self.subscribe_request.transaction_status_filter_count())
            .field("block_filter_count", &self.subscribe_request.block_filter_count())
+            .field("blocks_meta_filter_count", &self.subscribe_request.blocks_meta_filter_count())
+            .field("slot_filter_count", &self.subscribe_request.slot_filter_count())
            .field("commitment", &self.subscribe_request.commitment())
            .field("has_from_slot", &self.subscribe_request.from_slot().is_some())
            .field("hydration_role", &self.hydration_role.as_str())
@@ -361,6 +526,80 @@ fn matched_filter_fingerprint(filters: &[ksp_onchain_transport_lib::YellowstoneS
    return hasher.finalize().into();
 }

+#[cfg(test)]
+fn project_yellowstone_block_signals<T: RawTransactionIngestYellowstoneBlockView>(
+    source: &crate::RawTransactionIngestYellowstoneSource,
+    update: &T,
+) -> ksp_core_lib::Result<std::vec::Vec<RawTransactionIngestSourceSignal>> {
+    let filters = update.filters();
+    let matched_filter_count = filters.len();
+    let matched_filter_fingerprint = matched_filter_fingerprint(filters);
+    let matched_filter_id = matched_filter_direct_id(filters);
+    let created_at = map_yellowstone_source_timestamp(update.created_at());
+    let mut signals = std::vec::Vec::with_capacity(update.transaction_count());
+    for position in 0..update.transaction_count() {
+        let (signature, transaction_index) = match update.transaction_identity(position) {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        signals.push(RawTransactionIngestSourceSignal {
+            created_at,
+            family: RawTransactionIngestSourceFamily::Block,
+            matched_filter_count,
+            matched_filter_fingerprint,
+            matched_filter_id: matched_filter_id.clone(),
+            network: source.network.clone(),
+            route: source.route.clone(),
+            signature: ksp_store_lib::RawTransactionSignature::new(signature.into_bytes()),
+            slot: update.slot(),
+            transaction_index: std::option::Option::Some(transaction_index),
+        });
+    }
+    return std::result::Result::Ok(signals);
+}
+
+#[cfg(test)]
+fn project_yellowstone_continuity_signal<T: RawTransactionIngestYellowstoneContinuityView>(
+    source: &crate::RawTransactionIngestYellowstoneSource,
+    update: &T,
+) -> RawTransactionIngestContinuitySignal {
+    let filters = update.filters();
+    return RawTransactionIngestContinuitySignal {
+        created_at: map_yellowstone_source_timestamp(update.created_at()),
+        family: update.family(),
+        matched_filter_count: filters.len(),
+        matched_filter_fingerprint: matched_filter_fingerprint(filters),
+        matched_filter_id: matched_filter_direct_id(filters),
+        network: source.network.clone(),
+        parent_slot: update.parent_slot(),
+        route: source.route.clone(),
+        slot: update.slot(),
+        status: update.status(),
+    };
+}
+
+#[cfg(test)]
+fn map_yellowstone_slot_status(status: ksp_onchain_transport_lib::YellowstoneSlotStatus) -> RawTransactionIngestContinuityStatus {
+    return match status {
+        ksp_onchain_transport_lib::YellowstoneSlotStatus::Processed => RawTransactionIngestContinuityStatus::Processed,
+        ksp_onchain_transport_lib::YellowstoneSlotStatus::Confirmed => RawTransactionIngestContinuityStatus::Confirmed,
+        ksp_onchain_transport_lib::YellowstoneSlotStatus::Finalized => RawTransactionIngestContinuityStatus::Finalized,
+        ksp_onchain_transport_lib::YellowstoneSlotStatus::FirstShredReceived => RawTransactionIngestContinuityStatus::FirstShredReceived,
+        ksp_onchain_transport_lib::YellowstoneSlotStatus::Completed => RawTransactionIngestContinuityStatus::Completed,
+        ksp_onchain_transport_lib::YellowstoneSlotStatus::CreatedBank => RawTransactionIngestContinuityStatus::CreatedBank,
+        ksp_onchain_transport_lib::YellowstoneSlotStatus::Dead => RawTransactionIngestContinuityStatus::Dead,
+    };
+}
+
+#[cfg(test)]
+fn map_yellowstone_source_timestamp(
+    created_at: std::option::Option<ksp_onchain_transport_lib::YellowstoneUpdateTimestamp>,
+) -> std::option::Option<RawTransactionIngestSourceTimestamp> {
+    return created_at.map(|value| {
+        return RawTransactionIngestSourceTimestamp { nanos: value.nanos(), seconds: value.seconds() };
+    });
+}
+
 #[cfg(test)]
 fn project_yellowstone_signal<T: RawTransactionIngestYellowstoneSignalView>(
    source: &crate::RawTransactionIngestYellowstoneSource,
@@ -543,6 +782,7 @@ fn build_hydration_provenance(
 #[cfg(test)]
 fn hydration_method_code(family: RawTransactionIngestSourceFamily) -> &'static str {
    return match family {
+        RawTransactionIngestSourceFamily::Block => "block_get_transaction",
        RawTransactionIngestSourceFamily::Transaction => "transaction_get_transaction",
        RawTransactionIngestSourceFamily::TransactionStatus => "status_get_transaction",
    };