v0.3.13-pre.002

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: 17
+// version: 18
 
 #![warn(missing_docs)]
 #![deny(unreachable_pub)]
@@ -10,8 +10,9 @@
 //! This tranche owns the concrete Worker family identity, validated technical settings
 //! and the caller-runtime-owned lifecycle with private child-task supervision. This tranche also
 //! 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 validated Yellowstone/HTTP runtime-resource contract now drives one productive supervised source task.
+//! Store persistence in normal mode plus concrete latest-value snapshots projected onto Worker API. `pre.002` adds a bounded 1..32 caller-composed live-source
+//! aggregate with deterministic logical source identities while deliberately keeping productive multi-source activation gated until the supervisor tranche; the
+//! validated Yellowstone/HTTP runtime-resource contract still drives the only productive supervised source task.
 //! Transaction/TransactionStatus/Block updates feed bounded HTTP `getTransaction` hydration and the existing central
 //! admission path; BlockMeta/Slot remain continuity-only signals. Hydration is coalesced by network/signature/commitment
 //! under bounded in-flight and pending budgets. A bounded run-local processing frontier projects hydration pending,
@@ -43,6 +44,8 @@ pub use self::error::ERROR_CODE_RAW_TRANSACTION_INGEST_SOURCE_FAILED;
 pub use self::error::ERROR_CODE_RAW_TRANSACTION_INGEST_STORE_FAILED;
 /// Stable Worker kind code used by the continuous RAW transaction ingest vertical.
 pub use self::identity::RAW_TRANSACTION_INGEST_WORKER_KIND_CODE;
+/// Maximum number of logical live sources accepted by one runtime-resource aggregate.
+pub use self::runtime_resources::MAX_RAW_TRANSACTION_INGEST_LIVE_SOURCES;
 /// Cloneable external control handle for one continuous RAW transaction ingest Worker.
 pub use self::runtime::RawTransactionIngestHandle;
 /// Runtime-neutral boxed future resolving after one RAW transaction ingest Worker has fully reached a terminal lifecycle state.

crates/ksp-worker-raw-transaction-ingest-lib/src/runtime.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/src/runtime.rs
-// version: 11
+// version: 12
 
 type PersistencePort = std::sync::Arc<dyn crate::RawTransactionIngestPersistencePort + 'static>;
 type PersistenceTasks = tokio::task::JoinSet<ksp_core_lib::Result<crate::RawTransactionIngestPersistenceOutcome>>;
@@ -97,7 +97,10 @@ impl crate::RawTransactionIngestWorker {
         return start_foundation(settings, runtime, std::option::Option::Some(store));
     }
 
-    /// Starts one Worker with caller-composed runtime resources and one supervised productive Yellowstone source task.
+    /// Starts one Worker with caller-composed runtime resources.
+    ///
+    /// `pre.002` accepts a validated multi-source aggregate but intentionally activates only the existing single-Yellowstone supervisor path; aggregates with
+    /// more than one source fail closed before any source task is spawned.
     pub fn start_with_runtime_resources(
         settings: crate::RawTransactionIngestSettings,
         store: std::sync::Arc<ksp_store_lib::Store>,
@@ -114,7 +117,10 @@ impl crate::RawTransactionIngestWorker {
         if let std::result::Result::Err(error) = runtime_resources.validate_network(settings.network()) {
             return std::result::Result::Err(error);
         }
-        let source = runtime_resources.into_yellowstone_source();
+        let source = match runtime_resources.into_yellowstone_source() {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
         let source_settings = settings.clone();
         let (processing_frontier_sender, processing_frontier_receiver) =
             tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty());

crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs

@@ -1,8 +1,12 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs
-// version: 12
+// version: 13
 
 use sha2::Digest; // rust-rules: trait-import
 
+/// Maximum number of logical live sources accepted in one RAW transaction ingest runtime-resource aggregate.
+pub const MAX_RAW_TRANSACTION_INGEST_LIVE_SOURCES: usize = 32;
+
+const RAW_TRANSACTION_INGEST_LIVE_SOURCE_KEY_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.live_source.source_key.v1\0";
 const RAW_TRANSACTION_INGEST_YELLOWSTONE_FILTER_FINGERPRINT_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.yellowstone.filters.v1\0";
 const RAW_TRANSACTION_INGEST_YELLOWSTONE_HTTP_PROTOCOL: &str = "yellowstone_http";
 const RAW_TRANSACTION_INGEST_YELLOWSTONE_HTTP_SOURCE_KEY_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.yellowstone_http.source_key.v1\0";
@@ -14,6 +18,24 @@ enum RawTransactionIngestSourceFamily {
     TransactionStatus,
 }
 
+enum RawTransactionIngestLiveSource {
+    Yellowstone(crate::RawTransactionIngestYellowstoneSource),
+}
+
+impl RawTransactionIngestLiveSource {
+    fn network(&self) -> &ksp_store_lib::RawNetworkId {
+        return match self {
+            Self::Yellowstone(source) => &source.network,
+        };
+    }
+
+    fn source_key(&self) -> [u8; 32] {
+        return match self {
+            Self::Yellowstone(source) => source.source_key,
+        };
+    }
+}
+
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 enum RawTransactionIngestContinuityFamily {
     BlockMeta,
@@ -310,6 +332,7 @@ pub struct RawTransactionIngestYellowstoneSource {
     hydration_role: ksp_onchain_transport_lib::HttpRoleName,
     network: ksp_store_lib::RawNetworkId,
     route: RawTransactionIngestSourceRoute,
+    source_key: [u8; 32],
 }
 
 impl crate::RawTransactionIngestYellowstoneSource {
@@ -358,7 +381,12 @@ impl crate::RawTransactionIngestYellowstoneSource {
         if compatible_http_routes == 0 {
             return std::result::Result::Err(crate::runtime_error("runtime_resources.hydration_role_unsupported"));
         }
-        return std::result::Result::Ok(Self { yellowstone_channel, subscribe_request, http_pool, hydration_role, network, route });
+        let request_identity = match subscribe_request.identity() {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("runtime_resources.yellowstone_identity_invalid")),
+        };
+        let source_key = yellowstone_live_source_key(&network, &route, subscribe_request.commitment(), &request_identity);
+        return std::result::Result::Ok(Self { yellowstone_channel, subscribe_request, http_pool, hydration_role, network, route, source_key });
     }
 
     /// Runs the productive Yellowstone source task until cooperative stop or one safe terminal source failure.
@@ -510,45 +538,138 @@ impl std::fmt::Debug for crate::RawTransactionIngestYellowstoneSource {
             .field("has_from_slot", &self.subscribe_request.from_slot().is_some())
             .field("hydration_role", &self.hydration_role.as_str())
             .field("http_endpoint_count", &http_snapshot.endpoint_count())
+            .field("source_key_bytes", &self.source_key.len())
             .finish();
     }
 }
 
 /// Caller-composed runtime resources accepted by the continuous RAW transaction ingest Worker.
 ///
-/// This first runtime-resource contract owns exactly one Yellowstone source. It deliberately does not expose a provider enum, source collection, callback,
-/// enqueue surface or lower-layer client escape hatch.
+/// The aggregate owns a bounded collection of capability-specific live-source contracts. `pre.002` validates composition and stable logical identities but
+/// deliberately keeps productive multi-source supervision gated until the dedicated supervisor tranche.
 pub struct RawTransactionIngestRuntimeResources {
-    yellowstone_source: crate::RawTransactionIngestYellowstoneSource,
+    sources: std::vec::Vec<RawTransactionIngestLiveSource>,
 }
 
 impl crate::RawTransactionIngestRuntimeResources {
-    /// Owns the validated productive Yellowstone source contract consumed by `start_with_runtime_resources`.
+    /// Creates one runtime-resource aggregate from the already productive Yellowstone source contract.
     #[must_use]
     pub fn new(yellowstone_source: crate::RawTransactionIngestYellowstoneSource) -> Self {
-        return Self { yellowstone_source };
+        return Self { sources: std::vec![RawTransactionIngestLiveSource::Yellowstone(yellowstone_source)] };
     }
 
-    /// Validates that caller-owned Worker settings target the same logical network as the composed Yellowstone/HTTP source.
+    /// Returns the number of validated logical live sources currently owned by this aggregate.
+    #[must_use]
+    pub fn source_count(&self) -> usize {
+        return self.sources.len();
+    }
+
+    /// Adds one validated Yellowstone source while preserving the global 1..32 bound, one-network invariant and unique logical source identity.
+    pub fn try_push_yellowstone_source(&mut self, yellowstone_source: crate::RawTransactionIngestYellowstoneSource) -> ksp_core_lib::Result<()> {
+        if self.sources.len() >= crate::MAX_RAW_TRANSACTION_INGEST_LIVE_SOURCES {
+            return std::result::Result::Err(crate::runtime_error("runtime_resources.source_count_exceeded"));
+        }
+        let candidate = RawTransactionIngestLiveSource::Yellowstone(yellowstone_source);
+        let expected_network = match self.sources.first() {
+            std::option::Option::Some(source) => source.network(),
+            std::option::Option::None => return std::result::Result::Err(crate::runtime_error("runtime_resources.source_collection_empty")),
+        };
+        if candidate.network() != expected_network {
+            return std::result::Result::Err(crate::runtime_error("runtime_resources.source_network_mismatch"));
+        }
+        let source_key = candidate.source_key();
+        if self.sources.iter().any(|source| return source.source_key() == source_key) {
+            return std::result::Result::Err(crate::runtime_error("runtime_resources.duplicate_source_identity"));
+        }
+        self.sources.push(candidate);
+        return std::result::Result::Ok(());
+    }
+
+    /// Validates that caller-owned Worker settings target the same logical network as every composed live source.
     pub(crate) fn validate_network(&self, network: &ksp_store_lib::RawNetworkId) -> ksp_core_lib::Result<()> {
-        if &self.yellowstone_source.network != network {
-            return std::result::Result::Err(crate::runtime_error("runtime_resources.worker_network_mismatch"));
+        if self.sources.is_empty() || self.sources.len() > crate::MAX_RAW_TRANSACTION_INGEST_LIVE_SOURCES {
+            return std::result::Result::Err(crate::runtime_error("runtime_resources.source_collection_invalid"));
+        }
+        let mut source_keys = std::collections::BTreeSet::new();
+        for source in &self.sources {
+            if source.network() != network {
+                return std::result::Result::Err(crate::runtime_error("runtime_resources.worker_network_mismatch"));
+            }
+            if !source_keys.insert(source.source_key()) {
+                return std::result::Result::Err(crate::runtime_error("runtime_resources.duplicate_source_identity"));
+            }
         }
         return std::result::Result::Ok(());
     }
 
-    /// Consumes the runtime-resource aggregate into its single validated Yellowstone source for private supervisor wiring.
-    pub(crate) fn into_yellowstone_source(self) -> crate::RawTransactionIngestYellowstoneSource {
-        return self.yellowstone_source;
+    /// Consumes the aggregate into the single Yellowstone source supported by the current productive supervisor wiring.
+    ///
+    /// Multi-source activation is intentionally deferred to the dedicated supervisor/source-inventory tranche; a collection larger than one therefore fails
+    /// closed rather than silently dropping sources or starting only a subset.
+    pub(crate) fn into_yellowstone_source(self) -> ksp_core_lib::Result<crate::RawTransactionIngestYellowstoneSource> {
+        if self.sources.is_empty() {
+            return std::result::Result::Err(crate::runtime_error("runtime_resources.source_collection_empty"));
+        }
+        if self.sources.len() != 1 {
+            return std::result::Result::Err(crate::runtime_error("runtime_resources.multi_source_activation_pending"));
+        }
+        let mut sources = self.sources.into_iter();
+        return match sources.next() {
+            std::option::Option::Some(RawTransactionIngestLiveSource::Yellowstone(source)) => std::result::Result::Ok(source),
+            std::option::Option::None => std::result::Result::Err(crate::runtime_error("runtime_resources.source_collection_empty")),
+        };
     }
 }
 
 impl std::fmt::Debug for crate::RawTransactionIngestRuntimeResources {
     fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        return formatter.debug_struct("RawTransactionIngestRuntimeResources").field("yellowstone_source", &self.yellowstone_source).finish();
+        return formatter.debug_struct("RawTransactionIngestRuntimeResources").field("source_count", &self.sources.len()).finish();
     }
 }
 
+struct RawTransactionIngestSourceKeyHashWriter<'a> {
+    hasher: &'a mut sha2::Sha256,
+}
+
+impl std::hash::Hasher for RawTransactionIngestSourceKeyHashWriter<'_> {
+    fn finish(&self) -> u64 {
+        return 0;
+    }
+
+    fn write(&mut self, bytes: &[u8]) {
+        self.hasher.update(bytes);
+        return;
+    }
+}
+
+fn yellowstone_live_source_key(
+    network: &ksp_store_lib::RawNetworkId,
+    route: &RawTransactionIngestSourceRoute,
+    commitment: std::option::Option<ksp_onchain_transport_lib::SolanaCommitment>,
+    request_identity: &ksp_onchain_transport_lib::YellowstoneSubscribeRequestIdentity,
+) -> [u8; 32] {
+    let mut hasher = sha2::Sha256::new();
+    hasher.update(RAW_TRANSACTION_INGEST_LIVE_SOURCE_KEY_DOMAIN);
+    hash_live_source_key_component(&mut hasher, b"yellowstone");
+    hash_live_source_key_component(&mut hasher, network.as_str().as_bytes());
+    hash_live_source_key_component(&mut hasher, route.provider.as_str().as_bytes());
+    hash_live_source_key_component(&mut hasher, route.endpoint_id.as_str().as_bytes());
+    let commitment = match commitment {
+        std::option::Option::Some(value) => value.as_str(),
+        std::option::Option::None => "none",
+    };
+    hash_live_source_key_component(&mut hasher, commitment.as_bytes());
+    let mut writer = RawTransactionIngestSourceKeyHashWriter { hasher: &mut hasher };
+    std::hash::Hash::hash(request_identity, &mut writer);
+    return hasher.finalize().into();
+}
+
+fn hash_live_source_key_component(hasher: &mut sha2::Sha256, value: &[u8]) {
+    hasher.update((value.len() as u64).to_be_bytes());
+    hasher.update(value);
+    return;
+}
+
 fn compatible_http_route_count(
     pool: &ksp_onchain_transport_lib::HttpTransportPool,
     hydration_role: &ksp_onchain_transport_lib::HttpRoleName,