v0.3.13-pre.006

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: 22
+// version: 23
 
 #![warn(missing_docs)]
 #![deny(unreachable_pub)]
@@ -10,8 +10,8 @@
 //! 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. `pre.005` keeps the bounded 1..32 caller-composed
-//! aggregate and adds a productive Helius `transactionSubscribe` + HTTP hydration source beside Yellowstone, Standard Logs and Standard Block while
+//! Store persistence in normal mode plus concrete latest-value snapshots projected onto Worker API. `pre.006` keeps the bounded 1..32 caller-composed
+//! aggregate and adds productive HTTP live block polling beside Yellowstone, Standard Logs, Standard Block and Helius Transaction while
 //! simultaneous multi-source activation remains gated until the dedicated supervisor tranche. Helius Full notifications, Standard Logs and Yellowstone
 //! reference paths converge into one source-neutral hydration coordinator contract; qualified Standard Block Legacy/V0/V1 transactions enter the existing
 //! central admission path directly. Yellowstone
@@ -51,10 +51,24 @@ pub use self::runtime::RawTransactionIngestHandle;
 pub use self::runtime::RawTransactionIngestTerminalFuture;
 /// Entry point owning synchronous validation and task launch for one RAW transaction ingest Worker run.
 pub use self::runtime::RawTransactionIngestWorker;
+/// Default interval between HTTP live block polling cycles.
+pub use self::runtime_resources::DEFAULT_RAW_TRANSACTION_INGEST_HTTP_POLL_INTERVAL;
+/// Default maximum number of confirmed blocks discovered during one HTTP live block polling cycle.
+pub use self::runtime_resources::DEFAULT_RAW_TRANSACTION_INGEST_HTTP_POLL_MAX_BLOCKS_PER_CYCLE;
+/// Maximum interval accepted between HTTP live block polling cycles.
+pub use self::runtime_resources::MAX_RAW_TRANSACTION_INGEST_HTTP_POLL_INTERVAL;
+/// Maximum number of confirmed blocks accepted during one HTTP live block polling cycle.
+pub use self::runtime_resources::MAX_RAW_TRANSACTION_INGEST_HTTP_POLL_MAX_BLOCKS_PER_CYCLE;
 /// Maximum number of logical live sources accepted by one runtime-resource aggregate.
 pub use self::runtime_resources::MAX_RAW_TRANSACTION_INGEST_LIVE_SOURCES;
+/// Minimum interval accepted between HTTP live block polling cycles.
+pub use self::runtime_resources::MIN_RAW_TRANSACTION_INGEST_HTTP_POLL_INTERVAL;
+/// Minimum number of confirmed blocks accepted during one HTTP live block polling cycle.
+pub use self::runtime_resources::MIN_RAW_TRANSACTION_INGEST_HTTP_POLL_MAX_BLOCKS_PER_CYCLE;
 /// Validated Helius `transactionSubscribe` + HTTP hydration source contract owned by the continuous RAW transaction ingest Worker.
 pub use self::runtime_resources::RawTransactionIngestHeliusTransactionSource;
+/// Validated standard Solana HTTP live block polling source contract owned by the continuous RAW transaction ingest Worker.
+pub use self::runtime_resources::RawTransactionIngestHttpBlockPollingSource;
 /// Caller-composed bounded runtime resources for supported continuous RAW transaction live-source families.
 pub use self::runtime_resources::RawTransactionIngestRuntimeResources;
 /// Validated standard Solana `blockSubscribe` direct RAW source contract owned by the continuous RAW transaction ingest Worker.

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

@@ -1,14 +1,29 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs
-// version: 18
+// version: 19
 
 use sha2::Digest; // rust-rules: trait-import
 
+/// Default interval between HTTP live block polling cycles.
+pub const DEFAULT_RAW_TRANSACTION_INGEST_HTTP_POLL_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1);
+/// Default maximum number of confirmed blocks discovered during one HTTP live polling cycle.
+pub const DEFAULT_RAW_TRANSACTION_INGEST_HTTP_POLL_MAX_BLOCKS_PER_CYCLE: u16 = 128;
+/// Maximum interval accepted between HTTP live block polling cycles.
+pub const MAX_RAW_TRANSACTION_INGEST_HTTP_POLL_INTERVAL: std::time::Duration = std::time::Duration::from_secs(30);
+/// Maximum number of confirmed blocks accepted during one HTTP live polling cycle.
+pub const MAX_RAW_TRANSACTION_INGEST_HTTP_POLL_MAX_BLOCKS_PER_CYCLE: u16 = 1024;
 /// 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;
+/// Minimum interval accepted between HTTP live block polling cycles.
+pub const MIN_RAW_TRANSACTION_INGEST_HTTP_POLL_INTERVAL: std::time::Duration = std::time::Duration::from_millis(100);
+/// Minimum number of confirmed blocks accepted during one HTTP live polling cycle.
+pub const MIN_RAW_TRANSACTION_INGEST_HTTP_POLL_MAX_BLOCKS_PER_CYCLE: u16 = 1;
 
 const RAW_TRANSACTION_INGEST_HELIUS_TRANSACTION_FILTER_FINGERPRINT_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.helius_transaction.filter.v1\0";
 const RAW_TRANSACTION_INGEST_HELIUS_TRANSACTION_HTTP_PROTOCOL: &str = "helius_ws_http";
 const RAW_TRANSACTION_INGEST_HELIUS_TRANSACTION_HTTP_SOURCE_KEY_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.helius_transaction_http.source_key.v1\0";
+const RAW_TRANSACTION_INGEST_HTTP_BLOCK_POLLING_ACQUISITION_METHOD: &str = "block_polling_get_block";
+const RAW_TRANSACTION_INGEST_HTTP_BLOCK_POLLING_PROFILE_FINGERPRINT_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.http_block_polling.profile.v1\0";
+const RAW_TRANSACTION_INGEST_HTTP_BLOCK_POLLING_PROTOCOL: &str = "solana_http";
 const RAW_TRANSACTION_INGEST_LIVE_SOURCE_KEY_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.live_source.source_key.v1\0";
 const RAW_TRANSACTION_INGEST_STANDARD_BLOCK_ACQUISITION_METHOD: &str = "block_subscribe";
 const RAW_TRANSACTION_INGEST_STANDARD_BLOCK_FILTER_FINGERPRINT_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.standard_block.filter.v1\0";
@@ -30,6 +45,7 @@ enum RawTransactionIngestSourceFamily {
 
 enum RawTransactionIngestLiveSource {
     HeliusTransaction(crate::RawTransactionIngestHeliusTransactionSource),
+    HttpBlockPolling(crate::RawTransactionIngestHttpBlockPollingSource),
     StandardBlock(crate::RawTransactionIngestStandardBlockSource),
     StandardLogs(crate::RawTransactionIngestStandardLogsSource),
     Yellowstone(crate::RawTransactionIngestYellowstoneSource),
@@ -39,6 +55,7 @@ impl RawTransactionIngestLiveSource {
     fn network(&self) -> &ksp_store_lib::RawNetworkId {
         return match self {
             Self::HeliusTransaction(source) => &source.network,
+            Self::HttpBlockPolling(source) => &source.network,
             Self::StandardBlock(source) => &source.network,
             Self::StandardLogs(source) => &source.network,
             Self::Yellowstone(source) => &source.network,
@@ -48,6 +65,7 @@ impl RawTransactionIngestLiveSource {
     fn source_key(&self) -> [u8; 32] {
         return match self {
             Self::HeliusTransaction(source) => source.source_key,
+            Self::HttpBlockPolling(source) => source.source_key,
             Self::StandardBlock(source) => source.source_key,
             Self::StandardLogs(source) => source.source_key,
             Self::Yellowstone(source) => source.source_key,
@@ -63,6 +81,7 @@ impl RawTransactionIngestLiveSource {
     ) -> ksp_core_lib::Result<()> {
         return match self {
             Self::HeliusTransaction(source) => source.run(settings, stop_receiver, admission_sender, processing_frontier_sender).await,
+            Self::HttpBlockPolling(source) => source.run(settings, stop_receiver, admission_sender, processing_frontier_sender).await,
             Self::StandardBlock(source) => source.run(settings, stop_receiver, admission_sender, processing_frontier_sender).await,
             Self::StandardLogs(source) => source.run(settings, stop_receiver, admission_sender, processing_frontier_sender).await,
             Self::Yellowstone(source) => source.run(settings, stop_receiver, admission_sender, processing_frontier_sender).await,
@@ -855,6 +874,265 @@ impl std::fmt::Debug for crate::RawTransactionIngestHeliusTransactionSource {
     }
 }
 
+/// Validated standard Solana HTTP live block polling source owned by the continuous RAW transaction ingest Worker.
+///
+/// The caller provides one Transport-owned HTTP pool/role, a Confirmed/Finalized commitment and bounded polling controls. Construction proves that the role
+/// has same-network routes for `getSlot`, `getBlocksWithLimit` and `getBlock` without network I/O. Runtime starts at the first observed committed slot and
+/// never requests an earlier slot, keeping this source live/run-local rather than turning it into historical Backfill.
+pub struct RawTransactionIngestHttpBlockPollingSource {
+    http_pool: ksp_onchain_transport_lib::HttpTransportPool,
+    polling_role: ksp_onchain_transport_lib::HttpRoleName,
+    commitment: ksp_onchain_transport_lib::SolanaCommitment,
+    network: ksp_store_lib::RawNetworkId,
+    poll_interval: std::time::Duration,
+    max_discovered_blocks_per_cycle: u16,
+    profile_fingerprint: [u8; 32],
+    source_key: [u8; 32],
+}
+
+impl crate::RawTransactionIngestHttpBlockPollingSource {
+    /// Creates one HTTP live block polling source using the default 1-second cadence and 128-block discovery bound.
+    pub fn new(
+        http_pool: ksp_onchain_transport_lib::HttpTransportPool,
+        polling_role: ksp_onchain_transport_lib::HttpRoleName,
+        commitment: ksp_onchain_transport_lib::SolanaCommitment,
+    ) -> ksp_core_lib::Result<Self> {
+        return Self::new_with_limits(
+            http_pool,
+            polling_role,
+            commitment,
+            crate::DEFAULT_RAW_TRANSACTION_INGEST_HTTP_POLL_INTERVAL,
+            crate::DEFAULT_RAW_TRANSACTION_INGEST_HTTP_POLL_MAX_BLOCKS_PER_CYCLE,
+        );
+    }
+
+    /// Creates one HTTP live block polling source with explicit bounded cadence and discovery controls.
+    pub fn new_with_limits(
+        http_pool: ksp_onchain_transport_lib::HttpTransportPool,
+        polling_role: ksp_onchain_transport_lib::HttpRoleName,
+        commitment: ksp_onchain_transport_lib::SolanaCommitment,
+        poll_interval: std::time::Duration,
+        max_discovered_blocks_per_cycle: u16,
+    ) -> ksp_core_lib::Result<Self> {
+        match commitment {
+            ksp_onchain_transport_lib::SolanaCommitment::Confirmed | ksp_onchain_transport_lib::SolanaCommitment::Finalized => {},
+            ksp_onchain_transport_lib::SolanaCommitment::Processed => {
+                return std::result::Result::Err(crate::runtime_error("runtime_resources.http_block_polling_commitment_invalid"));
+            },
+        }
+        if !(crate::MIN_RAW_TRANSACTION_INGEST_HTTP_POLL_INTERVAL..=crate::MAX_RAW_TRANSACTION_INGEST_HTTP_POLL_INTERVAL).contains(&poll_interval) {
+            return std::result::Result::Err(crate::runtime_error("runtime_resources.http_block_polling_interval_invalid"));
+        }
+        if !(crate::MIN_RAW_TRANSACTION_INGEST_HTTP_POLL_MAX_BLOCKS_PER_CYCLE..=crate::MAX_RAW_TRANSACTION_INGEST_HTTP_POLL_MAX_BLOCKS_PER_CYCLE)
+            .contains(&max_discovered_blocks_per_cycle)
+        {
+            return std::result::Result::Err(crate::runtime_error("runtime_resources.http_block_polling_limit_invalid"));
+        }
+        let profile = validate_http_block_polling_profile(&http_pool, &polling_role);
+        let (network, profile_fingerprint) = match profile {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        let source_key = http_block_polling_live_source_key(&network, &polling_role, commitment, &profile_fingerprint);
+        return std::result::Result::Ok(Self {
+            http_pool,
+            polling_role,
+            commitment,
+            network,
+            poll_interval,
+            max_discovered_blocks_per_cycle,
+            profile_fingerprint,
+            source_key,
+        });
+    }
+
+    /// Runs one productive HTTP live block polling source until cooperative stop or one safe terminal source failure.
+    pub(crate) async fn run(
+        self,
+        settings: crate::RawTransactionIngestSettings,
+        mut stop_receiver: tokio::sync::watch::Receiver<bool>,
+        admission_sender: tokio::sync::mpsc::Sender<crate::RawTransactionIngress>,
+        processing_frontier_sender: tokio::sync::watch::Sender<crate::RawTransactionIngestProcessingFrontierProjection>,
+    ) -> ksp_core_lib::Result<()> {
+        let context_config = ksp_onchain_transport_lib::SolanaContextConfig::new(std::option::Option::Some(self.commitment), std::option::Option::None);
+        let get_block_config = ksp_onchain_transport_lib::SolanaGetBlockConfig::new(
+            std::option::Option::Some(self.commitment),
+            std::option::Option::Some(ksp_onchain_transport_lib::SolanaTransactionEncoding::Base64),
+            std::option::Option::Some(ksp_onchain_transport_lib::SolanaTransactionDetails::Full),
+            std::option::Option::Some(1),
+            std::option::Option::Some(false),
+        );
+        let start_slot = tokio::select! {
+            biased;
+            _ = stop_receiver.changed() => {
+                return std::result::Result::Ok(());
+            }
+            result = self.http_pool.get_slot(&self.polling_role, std::option::Option::Some(&context_config)) => result,
+        };
+        let start_slot = match start_slot {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(source_transport_error(error.code())),
+        };
+        let mut next_scan_slot = start_slot;
+        let mut processing_frontier = RawTransactionIngestProcessingFrontierReporter::new(processing_frontier_sender);
+        processing_frontier.set_source_state(crate::RawTransactionIngestSourceState::Active);
+        let mut fault = std::option::Option::None;
+        'source: loop {
+            if *stop_receiver.borrow() {
+                break;
+            }
+            let current_tip = tokio::select! {
+                biased;
+                _ = stop_receiver.changed() => {
+                    break;
+                }
+                result = self.http_pool.get_slot(&self.polling_role, std::option::Option::Some(&context_config)) => result,
+            };
+            let current_tip = match current_tip {
+                std::result::Result::Ok(value) => value,
+                std::result::Result::Err(error) => {
+                    fault = std::option::Option::Some(source_transport_error(error.code()));
+                    break;
+                },
+            };
+            if next_scan_slot <= current_tip {
+                let discovered = tokio::select! {
+                    biased;
+                    _ = stop_receiver.changed() => {
+                        break;
+                    }
+                    result = self.http_pool.get_blocks_with_limit(
+                        &self.polling_role,
+                        next_scan_slot,
+                        u64::from(self.max_discovered_blocks_per_cycle),
+                        std::option::Option::Some(&context_config),
+                    ) => result,
+                };
+                let discovered = match discovered {
+                    std::result::Result::Ok(value) => value,
+                    std::result::Result::Err(error) => {
+                        fault = std::option::Option::Some(source_transport_error(error.code()));
+                        break;
+                    },
+                };
+                let validated = validate_http_block_polling_discovery(next_scan_slot, discovered.as_slice());
+                if let std::result::Result::Err(error) = validated {
+                    fault = std::option::Option::Some(error);
+                    break;
+                }
+                let discovered_count = discovered.len();
+                let mut blocked_by_null = false;
+                for slot in discovered {
+                    let observed = tokio::select! {
+                        biased;
+                        _ = stop_receiver.changed() => {
+                            break 'source;
+                        }
+                        result = self.http_pool.get_block_observed(&self.polling_role, slot, std::option::Option::Some(&get_block_config)) => result,
+                    };
+                    let observed = match observed {
+                        std::result::Result::Ok(value) => value,
+                        std::result::Result::Err(error) => {
+                            fault = std::option::Option::Some(source_transport_error(error.code()));
+                            break 'source;
+                        },
+                    };
+                    let block = match observed.value() {
+                        std::option::Option::Some(value) => value,
+                        std::option::Option::None => {
+                            next_scan_slot = slot;
+                            blocked_by_null = true;
+                            break;
+                        },
+                    };
+                    let received_at = match current_raw_timestamp() {
+                        std::result::Result::Ok(value) => value,
+                        std::result::Result::Err(error) => {
+                            fault = std::option::Option::Some(error);
+                            break 'source;
+                        },
+                    };
+                    let ingresses = project_http_block_polling_ingresses(&self, &settings, slot, block, &observed, received_at);
+                    let ingresses = match ingresses {
+                        std::result::Result::Ok(value) => value,
+                        std::result::Result::Err(error) => {
+                            fault = std::option::Option::Some(error);
+                            break 'source;
+                        },
+                    };
+                    for ingress in ingresses {
+                        let sent = tokio::select! {
+                            biased;
+                            _ = stop_receiver.changed() => {
+                                break 'source;
+                            }
+                            result = admission_sender.send(ingress) => result,
+                        };
+                        if sent.is_err() {
+                            if *stop_receiver.borrow() {
+                                break 'source;
+                            }
+                            fault = std::option::Option::Some(crate::runtime_error("source.admission_closed"));
+                            break 'source;
+                        }
+                    }
+                    if let std::result::Result::Err(error) = processing_frontier.observe_settled(slot) {
+                        fault = std::option::Option::Some(error);
+                        break 'source;
+                    }
+                    next_scan_slot = match slot.checked_add(1) {
+                        std::option::Option::Some(value) => value,
+                        std::option::Option::None => {
+                            fault = std::option::Option::Some(crate::runtime_error("source.http_block_polling_slot_exhausted"));
+                            break 'source;
+                        },
+                    };
+                }
+                if !blocked_by_null && discovered_count < usize::from(self.max_discovered_blocks_per_cycle) && next_scan_slot <= current_tip {
+                    next_scan_slot = match current_tip.checked_add(1) {
+                        std::option::Option::Some(value) => value,
+                        std::option::Option::None => {
+                            fault = std::option::Option::Some(crate::runtime_error("source.http_block_polling_slot_exhausted"));
+                            break;
+                        },
+                    };
+                }
+            }
+            tokio::select! {
+                biased;
+                _ = stop_receiver.changed() => {
+                    break;
+                }
+                _ = tokio::time::sleep(self.poll_interval) => {},
+            }
+        }
+        processing_frontier.discard_all_pending();
+        processing_frontier.set_source_state(crate::RawTransactionIngestSourceState::Closing);
+        if let std::option::Option::Some(error) = fault {
+            processing_frontier.set_source_state(crate::RawTransactionIngestSourceState::Failed);
+            return std::result::Result::Err(error);
+        }
+        processing_frontier.set_source_state(crate::RawTransactionIngestSourceState::Closed);
+        return std::result::Result::Ok(());
+    }
+}
+
+impl std::fmt::Debug for crate::RawTransactionIngestHttpBlockPollingSource {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        return formatter
+            .debug_struct("RawTransactionIngestHttpBlockPollingSource")
+            .field("polling_role", &self.polling_role.as_str())
+            .field("network", &self.network.as_str())
+            .field("commitment", &self.commitment)
+            .field("poll_interval_ms", &self.poll_interval.as_millis())
+            .field("max_discovered_blocks_per_cycle", &self.max_discovered_blocks_per_cycle)
+            .field("http_endpoint_count", &self.http_pool.snapshot().endpoint_count())
+            .field("profile_fingerprint_bytes", &self.profile_fingerprint.len())
+            .field("source_key_bytes", &self.source_key.len())
+            .finish();
+    }
+}
+
 /// Validated standard Solana `blockSubscribe` direct RAW source owned by the continuous RAW transaction ingest Worker.
 ///
 /// The caller provides one Transport-owned standard WebSocket endpoint, one block filter and a Confirmed/Finalized commitment. The source requests Full/Base64
@@ -1332,8 +1610,9 @@ impl std::fmt::Debug for crate::RawTransactionIngestYellowstoneSource {
 
 /// Caller-composed runtime resources accepted by the continuous RAW transaction ingest Worker.
 ///
-/// The aggregate owns a bounded collection of capability-specific live-source contracts. `pre.005` supports Yellowstone, standard Solana logs/block and
-/// Helius transaction sources while deliberately keeping simultaneous multi-source supervision gated until the dedicated supervisor tranche.
+/// The aggregate owns a bounded collection of capability-specific live-source contracts. `pre.006` supports Yellowstone, standard Solana logs/block, Helius
+/// transaction and HTTP live block polling sources while deliberately keeping simultaneous multi-source supervision gated until the dedicated supervisor
+/// tranche.
 pub struct RawTransactionIngestRuntimeResources {
     sources: std::vec::Vec<RawTransactionIngestLiveSource>,
 }
@@ -1351,6 +1630,12 @@ impl crate::RawTransactionIngestRuntimeResources {
         return Self { sources: std::vec![RawTransactionIngestLiveSource::HeliusTransaction(source)] };
     }
 
+    /// Creates one runtime-resource aggregate from one standard Solana HTTP live block polling source contract.
+    #[must_use]
+    pub fn from_http_block_polling_source(source: crate::RawTransactionIngestHttpBlockPollingSource) -> Self {
+        return Self { sources: std::vec![RawTransactionIngestLiveSource::HttpBlockPolling(source)] };
+    }
+
     /// Creates one runtime-resource aggregate from one standard Solana `blockSubscribe` direct RAW source contract.
     #[must_use]
     pub fn from_standard_block_source(source: crate::RawTransactionIngestStandardBlockSource) -> Self {
@@ -1411,6 +1696,27 @@ impl crate::RawTransactionIngestRuntimeResources {
         return std::result::Result::Ok(());
     }
 
+    /// Adds one validated HTTP live block polling source while preserving the global 1..32 bound, one-network invariant and unique logical source identity.
+    pub fn try_push_http_block_polling_source(&mut self, source: crate::RawTransactionIngestHttpBlockPollingSource) -> 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::HttpBlockPolling(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(());
+    }
+
     /// Adds one validated standard Solana block source while preserving the global 1..32 bound, one-network invariant and unique logical source identity.
     pub fn try_push_standard_block_source(&mut self, source: crate::RawTransactionIngestStandardBlockSource) -> ksp_core_lib::Result<()> {
         if self.sources.len() >= crate::MAX_RAW_TRANSACTION_INGEST_LIVE_SOURCES {
@@ -1522,6 +1828,277 @@ impl std::hash::Hasher for RawTransactionIngestSourceKeyHashWriter<'_> {
     }
 }
 
+fn http_block_polling_live_source_key(
+    network: &ksp_store_lib::RawNetworkId,
+    polling_role: &ksp_onchain_transport_lib::HttpRoleName,
+    commitment: ksp_onchain_transport_lib::SolanaCommitment,
+    profile_fingerprint: &[u8; 32],
+) -> [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"http_block_polling");
+    hash_live_source_key_component(&mut hasher, network.as_str().as_bytes());
+    hash_live_source_key_component(&mut hasher, polling_role.as_str().as_bytes());
+    hash_live_source_key_component(&mut hasher, commitment.as_str().as_bytes());
+    hash_live_source_key_component(&mut hasher, profile_fingerprint);
+    return hasher.finalize().into();
+}
+
+fn validate_http_block_polling_profile(
+    http_pool: &ksp_onchain_transport_lib::HttpTransportPool,
+    polling_role: &ksp_onchain_transport_lib::HttpRoleName,
+) -> ksp_core_lib::Result<(ksp_store_lib::RawNetworkId, [u8; 32])> {
+    let get_block = match ksp_onchain_transport_lib::find_http_rpc_method("getBlock") {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return std::result::Result::Err(crate::runtime_error("runtime_resources.http_block_polling_get_block_missing")),
+    };
+    let get_blocks = match ksp_onchain_transport_lib::find_http_rpc_method("getBlocksWithLimit") {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return std::result::Result::Err(crate::runtime_error("runtime_resources.http_block_polling_get_blocks_missing")),
+    };
+    let get_slot = match ksp_onchain_transport_lib::find_http_rpc_method("getSlot") {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return std::result::Result::Err(crate::runtime_error("runtime_resources.http_block_polling_get_slot_missing")),
+    };
+    let mut has_get_block = false;
+    let mut has_get_blocks = false;
+    let mut has_get_slot = false;
+    let mut expected_cluster: std::option::Option<&str> = std::option::Option::None;
+    let mut routes = std::vec::Vec::new();
+    let snapshot = http_pool.snapshot();
+    for endpoint in snapshot.endpoints() {
+        if !endpoint.enabled() {
+            continue;
+        }
+        for role in endpoint.roles() {
+            if !role.enabled() || role.role() != polling_role.as_str() {
+                continue;
+            }
+            let supports_get_block = http_role_supports_request_kind(role, get_block.request_kind());
+            let supports_get_blocks = http_role_supports_request_kind(role, get_blocks.request_kind());
+            let supports_get_slot = http_role_supports_request_kind(role, get_slot.request_kind());
+            if !supports_get_block && !supports_get_blocks && !supports_get_slot {
+                continue;
+            }
+            match expected_cluster {
+                std::option::Option::Some(cluster) if cluster != endpoint.cluster() => {
+                    return std::result::Result::Err(crate::runtime_error("runtime_resources.transport_network_mismatch"));
+                },
+                std::option::Option::Some(_) => {},
+                std::option::Option::None => expected_cluster = std::option::Option::Some(endpoint.cluster()),
+            }
+            if ksp_store_lib::RawProvenanceCode::new(endpoint.provider()).is_err() {
+                return std::result::Result::Err(crate::runtime_error("runtime_resources.http_block_polling_provider_unrepresentable"));
+            }
+            if ksp_store_lib::RawProvenanceCode::new(endpoint.name()).is_err() {
+                return std::result::Result::Err(crate::runtime_error("runtime_resources.http_block_polling_endpoint_unrepresentable"));
+            }
+            has_get_block |= supports_get_block;
+            has_get_blocks |= supports_get_blocks;
+            has_get_slot |= supports_get_slot;
+            routes.push((
+                endpoint.cluster().to_owned(),
+                endpoint.provider().to_owned(),
+                endpoint.name().to_owned(),
+                supports_get_block,
+                supports_get_blocks,
+                supports_get_slot,
+            ));
+        }
+    }
+    if !has_get_slot || !has_get_blocks || !has_get_block {
+        return std::result::Result::Err(crate::runtime_error("runtime_resources.http_block_polling_role_unsupported"));
+    }
+    let cluster = match expected_cluster {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return std::result::Result::Err(crate::runtime_error("runtime_resources.http_block_polling_role_unsupported")),
+    };
+    let network = match ksp_store_lib::RawNetworkId::new(cluster) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => {
+            return std::result::Result::Err(crate::runtime_error("runtime_resources.http_block_polling_network_unrepresentable"));
+        },
+    };
+    routes.sort_unstable();
+    let mut hasher = sha2::Sha256::new();
+    hasher.update(RAW_TRANSACTION_INGEST_HTTP_BLOCK_POLLING_PROFILE_FINGERPRINT_DOMAIN);
+    hash_live_source_key_component(&mut hasher, polling_role.as_str().as_bytes());
+    hash_live_source_key_component(&mut hasher, &(routes.len() as u64).to_be_bytes());
+    for (cluster, provider, endpoint, supports_get_block, supports_get_blocks, supports_get_slot) in routes {
+        hash_live_source_key_component(&mut hasher, cluster.as_bytes());
+        hash_live_source_key_component(&mut hasher, provider.as_bytes());
+        hash_live_source_key_component(&mut hasher, endpoint.as_bytes());
+        hash_live_source_key_component(&mut hasher, if supports_get_block { b"1" } else { b"0" });
+        hash_live_source_key_component(&mut hasher, if supports_get_blocks { b"1" } else { b"0" });
+        hash_live_source_key_component(&mut hasher, if supports_get_slot { b"1" } else { b"0" });
+    }
+    return std::result::Result::Ok((network, hasher.finalize().into()));
+}
+
+fn http_role_supports_request_kind(role: &ksp_onchain_transport_lib::HttpEndpointRoleSnapshot, request_kind: &str) -> bool {
+    return role.request_kinds().iter().any(|kind| return kind.as_str() == "*" || kind.as_str() == request_kind);
+}
+
+fn validate_http_block_polling_discovery(next_scan_slot: u64, discovered: &[u64]) -> ksp_core_lib::Result<()> {
+    let mut previous = std::option::Option::None;
+    for slot in discovered {
+        if *slot < next_scan_slot {
+            return std::result::Result::Err(crate::runtime_error("source.http_block_polling_discovery_before_frontier"));
+        }
+        if let std::option::Option::Some(previous) = previous
+            && *slot <= previous
+        {
+            return std::result::Result::Err(crate::runtime_error("source.http_block_polling_discovery_not_strictly_increasing"));
+        }
+        previous = std::option::Option::Some(*slot);
+    }
+    return std::result::Result::Ok(());
+}
+
+fn project_http_block_polling_ingresses(
+    source: &crate::RawTransactionIngestHttpBlockPollingSource,
+    settings: &crate::RawTransactionIngestSettings,
+    slot: u64,
+    block: &ksp_onchain_transport_lib::SolanaConfirmedBlock,
+    observed: &ksp_onchain_transport_lib::HttpObservedValue<std::option::Option<ksp_onchain_transport_lib::SolanaConfirmedBlock>>,
+    received_at: ksp_store_lib::RawTimestamp,
+) -> ksp_core_lib::Result<std::vec::Vec<crate::RawTransactionIngress>> {
+    let transactions = match block.transactions() {
+        ksp_onchain_transport_lib::SolanaWireField::Value(value) => value,
+        ksp_onchain_transport_lib::SolanaWireField::Omitted | ksp_onchain_transport_lib::SolanaWireField::Null => {
+            return std::result::Result::Err(crate::runtime_error("source.http_block_polling_transactions_missing"));
+        },
+    };
+    let provenance = build_http_block_polling_provenance(source, settings, observed.endpoint_name(), observed.provider(), received_at);
+    let provenance = match provenance {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let mut ingresses = std::vec::Vec::with_capacity(transactions.len());
+    for (position, transaction) in transactions.iter().enumerate() {
+        let material = build_http_block_polling_material_from_transaction(&source.network, slot, block.block_time(), transaction, position);
+        let material = match material {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        ingresses.push(crate::RawTransactionIngress {
+            material,
+            network: source.network.clone(),
+            provenance: provenance.clone(),
+            source_key: source.source_key,
+        });
+    }
+    return std::result::Result::Ok(ingresses);
+}
+
+fn build_http_block_polling_material_from_transaction(
+    network: &ksp_store_lib::RawNetworkId,
+    slot: u64,
+    block_time: std::option::Option<i64>,
+    transaction: &ksp_onchain_transport_lib::SolanaBlockTransaction,
+    position: usize,
+) -> ksp_core_lib::Result<ksp_raw_transaction_lib::RawTransactionMaterial> {
+    let transaction_data = match transaction.transaction() {
+        ksp_onchain_transport_lib::SolanaEncodedTransaction::Binary { data, encoding }
+            if *encoding == ksp_onchain_transport_lib::SolanaTransactionBinaryEncoding::Base64 =>
+        {
+            data.as_str()
+        },
+        ksp_onchain_transport_lib::SolanaEncodedTransaction::Binary { .. }
+        | ksp_onchain_transport_lib::SolanaEncodedTransaction::LegacyBinary(_)
+        | ksp_onchain_transport_lib::SolanaEncodedTransaction::Json(_) => {
+            return std::result::Result::Err(crate::runtime_error("source.http_block_polling_transaction_encoding_invalid"));
+        },
+    };
+    let version = qualify_http_block_polling_version(transaction.version());
+    let version = match version {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let transaction_index = match u32::try_from(position) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("source.http_block_polling_transaction_index_invalid")),
+    };
+    let material = ksp_raw_transaction_lib::RawTransactionMaterial::binary_base64_with_embedded_signature(
+        network.clone(),
+        slot,
+        block_time,
+        transaction_data,
+        map_hydration_wire_field(transaction.meta(), |value| return value.clone()),
+        version,
+        ksp_raw_transaction_lib::RawTransactionWireField::Value(transaction_index),
+    );
+    return match material {
+        std::result::Result::Ok(value) => std::result::Result::Ok(value),
+        std::result::Result::Err(_) => std::result::Result::Err(crate::runtime_error("source.http_block_polling_transaction_signature_invalid")),
+    };
+}
+
+fn qualify_http_block_polling_version(
+    version: &ksp_onchain_transport_lib::SolanaWireField<ksp_onchain_transport_lib::SolanaTransactionVersion>,
+) -> ksp_core_lib::Result<ksp_raw_transaction_lib::RawTransactionWireField<ksp_raw_transaction_lib::RawTransactionVersion>> {
+    return match version {
+        ksp_onchain_transport_lib::SolanaWireField::Value(ksp_onchain_transport_lib::SolanaTransactionVersion::Legacy) => {
+            std::result::Result::Ok(ksp_raw_transaction_lib::RawTransactionWireField::Value(ksp_raw_transaction_lib::RawTransactionVersion::Legacy))
+        },
+        ksp_onchain_transport_lib::SolanaWireField::Value(ksp_onchain_transport_lib::SolanaTransactionVersion::Number(0)) => {
+            std::result::Result::Ok(ksp_raw_transaction_lib::RawTransactionWireField::Value(ksp_raw_transaction_lib::RawTransactionVersion::Number(0)))
+        },
+        ksp_onchain_transport_lib::SolanaWireField::Value(ksp_onchain_transport_lib::SolanaTransactionVersion::Number(1)) => {
+            std::result::Result::Ok(ksp_raw_transaction_lib::RawTransactionWireField::Value(ksp_raw_transaction_lib::RawTransactionVersion::Number(1)))
+        },
+        ksp_onchain_transport_lib::SolanaWireField::Value(ksp_onchain_transport_lib::SolanaTransactionVersion::Number(_)) => {
+            std::result::Result::Err(crate::runtime_error("source.http_block_polling_transaction_version_unsupported"))
+        },
+        ksp_onchain_transport_lib::SolanaWireField::Omitted | ksp_onchain_transport_lib::SolanaWireField::Null => {
+            std::result::Result::Err(crate::runtime_error("source.http_block_polling_transaction_version_unqualified"))
+        },
+    };
+}
+
+fn build_http_block_polling_provenance(
+    source: &crate::RawTransactionIngestHttpBlockPollingSource,
+    settings: &crate::RawTransactionIngestSettings,
+    endpoint_name: &str,
+    provider_name: &ksp_onchain_transport_lib::HttpProviderName,
+    received_at: ksp_store_lib::RawTimestamp,
+) -> ksp_core_lib::Result<ksp_store_lib::RawAcquisitionProvenance> {
+    let provider = match ksp_store_lib::RawProvenanceCode::new(provider_name.as_str()) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => {
+            return std::result::Result::Err(crate::runtime_error("source.http_block_polling_provider_unrepresentable"));
+        },
+    };
+    let endpoint_id = match ksp_store_lib::RawProvenanceCode::new(endpoint_name) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => {
+            return std::result::Result::Err(crate::runtime_error("source.http_block_polling_endpoint_unrepresentable"));
+        },
+    };
+    let protocol = match ksp_store_lib::RawProvenanceCode::new(RAW_TRANSACTION_INGEST_HTTP_BLOCK_POLLING_PROTOCOL) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("source.http_block_polling_protocol_unrepresentable")),
+    };
+    let acquisition_method = match ksp_store_lib::RawProvenanceCode::new(RAW_TRANSACTION_INGEST_HTTP_BLOCK_POLLING_ACQUISITION_METHOD) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("source.http_block_polling_method_unrepresentable")),
+    };
+    let capture_session = match ksp_store_lib::RawProvenanceCode::new(settings.worker_id().as_str()) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("source.http_block_polling_capture_session_unrepresentable")),
+    };
+    let commitment = match ksp_store_lib::RawProvenanceCode::new(source.commitment.as_str()) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("source.http_block_polling_commitment_unrepresentable")),
+    };
+    return std::result::Result::Ok(
+        ksp_store_lib::RawAcquisitionProvenance::new(provider, protocol, acquisition_method, ksp_store_lib::RawAcquisitionOrigin::Live, received_at)
+            .with_endpoint_id(endpoint_id)
+            .with_commitment(commitment)
+            .with_capture_session_id(capture_session),
+    );
+}
+
 fn helius_transaction_filter_fingerprint(filter: &ksp_onchain_transport_lib::HeliusTransactionSubscribeFilter) -> [u8; 32] {
     let mut hasher = sha2::Sha256::new();
     hasher.update(RAW_TRANSACTION_INGEST_HELIUS_TRANSACTION_FILTER_FINGERPRINT_DOMAIN);