From f68832517080ed2a46757949761771eff9b7a814 Mon Sep 17 00:00:00 2001 From: SinuS Von SifriduS Date: Thu, 10 Sep 2026 16:05:52 +0200 Subject: [PATCH] v0.3.13-pre.006 --- Cargo.toml | 4 +- README.md | 4 +- .../README.md | 29 +- .../USAGE.md | 35 +- .../src/lib.rs | 20 +- .../src/runtime_resources.rs | 583 +++++++++++++++++- .../tests/dependency_boundary.rs | 69 ++- .../tests/hardening.rs | 58 +- .../tests/public_api.rs | 39 +- .../tests/release_completeness.rs | 12 +- .../unit_tests/runtime_resources.rs | 230 ++++++- deltas/0.3.13/pre.006.md | 322 ++++++++++ ...3_13_MULTI_SOURCE_LIVE_CONVERGENCE_PLAN.md | 104 +++- ...0-V0_3_13_MULTI_SOURCE_LIVE_CONVERGENCE.md | 91 ++- 14 files changed, 1549 insertions(+), 51 deletions(-) create mode 100644 deltas/0.3.13/pre.006.md diff --git a/Cargo.toml b/Cargo.toml index 1b23d6e..90b04cc 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,12 +1,12 @@ # file: Cargo.toml -# version: 549 +# version: 550 [workspace] resolver = "3" members = ["crates/ksp-app-backfill-desk", "crates/ksp-app-config-desk", "crates/ksp-app-solprices-desk", "crates/ksp-app-store-desk", "crates/ksp-app-wallet-desk", "crates/ksp-config-lib", "crates/ksp-core-lib", "crates/ksp-interface-lib", "crates/ksp-job-api", "crates/ksp-job-backfill-lib", "crates/ksp-logging-lib", "crates/ksp-offchain-transport-lib", "crates/ksp-onchain-transport-lib", "crates/ksp-program-api", "crates/ksp-raw-transaction-lib", "crates/ksp-store-api", "crates/ksp-store-lib", "crates/ksp-store-postgres-lib", "crates/ksp-wallet-lib", "crates/ksp-worker-api", "crates/ksp-worker-raw-transaction-ingest-lib"] [workspace.package] -version = "0.3.13-pre.5.fix.1" +version = "0.3.13-pre.6" edition = "2024" license = "MIT" repository = "https://git.sasedev.com/Sasedev/khadhroony-solana-project" diff --git a/README.md b/README.md index 94f8218..0cad423 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,5 @@ - + # Khadhroony Solana Project @@ -57,7 +57,7 @@ La couche RAW dispose d'une façade Store backend-neutral et d'un premier job hi `ksp-worker-api` fournit désormais la fondation générique des services continus : identité Worker, lifecycle borné, health/activity, intention de stop coopératif et observation latest-value. Cette API reste Core-only, runtime-neutral et sans connaissance Solana, Transport, Store ou Job. Elle ne démarre ni n'arrête elle-même un runtime concret. -`ksp-worker-raw-transaction-ingest-lib` est le premier Worker concret distinct du Job Backfill. Il conserve sa fondation Start/Stop, admission bornée, Common RAW, Store backend-neutral, shutdown borné et snapshots latest-value, et possède maintenant une première source productive Yellowstone + hydration HTTP `getTransaction`. Les signaux `Transaction`, `TransactionStatus` et transactions de `Block` sont coalescés puis hydratés avant admission RAW ; `BlockMeta` et `Slot` restent des signaux de continuité. Le reconnect/replay reste propriétaire de Transport, la frontier de traitement reste run-local et un gap de rétention prouvé provoque un fault au lieu de lancer une campagne historique. `ksp-job-backfill-lib` conserve son rôle historique paramétré et borné ; les deux producteurs restent indépendants et convergent uniquement vers les mêmes contrats RAW/Store. +`ksp-worker-raw-transaction-ingest-lib` est le premier Worker concret distinct du Job Backfill. Il conserve sa fondation Start/Stop, admission bornée, Common RAW, Store backend-neutral, shutdown borné et snapshots latest-value. Ses sources productives couvrent maintenant Yellowstone + hydration HTTP `getTransaction`, Standard WS `logsSubscribe` + hydration, Standard WS `blockSubscribe` RAW-direct, Helius `transactionSubscribe` + hydration et Solana HTTP live block polling. Le polling HTTP démarre au `getSlot` observé du run, découvre uniquement les blocs live via `getBlocksWithLimit` puis matérialise les slots listés via `getBlock observed`; il n'est jamais utilisé comme Backfill implicite. Les signaux hydratés de même `(network, signature, commitment)` convergent vers le même coordinator, tandis que les voies block RAW-direct passent par la même admission Common RAW. Le reconnect/replay reste propriétaire de Transport, la frontier de traitement reste run-local et un gap de rétention prouvé provoque un fault au lieu de lancer une campagne historique. `ksp-job-backfill-lib` conserve son rôle historique paramétré et borné ; les deux producteurs restent indépendants et convergent uniquement vers les mêmes contrats RAW/Store. ## Points d'entrée diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/README.md b/crates/ksp-worker-raw-transaction-ingest-lib/README.md index e49a9e7..1de4a11 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/README.md +++ b/crates/ksp-worker-raw-transaction-ingest-lib/README.md @@ -1,5 +1,5 @@ - + # ksp-worker-raw-transaction-ingest-lib @@ -13,7 +13,7 @@ RawTransactionIngestWorker::start RawTransactionIngestWorker::start_with_runtime_resources -> même runtime + une source productive supervisée - Yellowstone, WS standard logsSubscribe, WS standard blockSubscribe ou Helius transactionSubscribe + Yellowstone, WS standard logsSubscribe, WS standard blockSubscribe, Helius transactionSubscribe ou HTTP block polling + hydration HTTP getTransaction lorsque la source produit une référence ``` @@ -45,6 +45,12 @@ Helius transactionSubscribe -> coalescence bornée par (network, signature, commitment) -> HTTP getTransaction observed avant Common RAW +Solana HTTP live block polling + -> getSlot borne le run courant + -> getBlocksWithLimit découvre les blocs live + -> getBlock observed matérialise Full/Base64 Legacy/V0/V1 + -> matériau Common RAW direct par transaction du bloc + les chemins productifs -> ksp-raw-transaction-lib -> admission centrale bornée @@ -76,7 +82,7 @@ La construction est sans I/O et refuse notamment : - une provenance provider/endpoint non représentable ; - l'absence d'une route HTTP compatible pour `getTransaction` sur le même réseau. -Le runtime-resource aggregate public accepte une collection validée de 1 à 32 sources logiques. `pre.005` sait exécuter une source unique Yellowstone, Standard Logs, Standard Block ou Helius Transaction ; une activation simultanée de plusieurs sources reste rejetée avant spawn jusqu'au supervisor dédié. La collection interne, les `source_key`, les URLs, les filtres et les clients inférieurs ne sont pas exposés. +Le runtime-resource aggregate public accepte une collection validée de 1 à 32 sources logiques. `pre.006` sait exécuter une source unique Yellowstone, Standard Logs, Standard Block, Helius Transaction ou HTTP Block Polling ; une activation simultanée de plusieurs sources reste rejetée avant spawn jusqu'au supervisor dédié. La collection interne, les `source_key`, les URLs, les filtres et les clients inférieurs ne sont pas exposés. ## Contrat de source Standard Logs + HTTP @@ -128,6 +134,22 @@ Au runtime, `HeliusLaserStreamWsSession::connect` puis `transaction_subscribe` s La clé logique Helius inclut réseau, identités provider/endpoint sûres, commitment et empreinte privée du filtre. Les listes de pubkeys du filtre sont normalisées avant hash afin que leur ordre ne crée pas artificiellement deux sources logiques ; le rôle HTTP d'hydration reste exclu de l'identité live. +## Contrat de source HTTP Block Polling + +`RawTransactionIngestHttpBlockPollingSource::new` reçoit : + +```text +HttpTransportPool +HttpRoleName de polling +SolanaCommitment Confirmed ou Finalized +``` + +La construction est sans I/O et vérifie que le rôle HTTP possède, sur un seul réseau, les capacités `getSlot`, `getBlocksWithLimit` et `getBlock`. La cadence est bornée entre 100 ms et 30 s, avec 1 s par défaut ; la découverte est bornée entre 1 et 1024 blocs par cycle, avec 128 par défaut. Ces réglages de cadence ne font pas partie de l'identité logique de la source. + +Au démarrage, le premier `getSlot` fixe la borne inférieure du run. Le Worker ne demande aucun slot antérieur. Chaque cycle relit le tip, découvre les blocs disponibles avec `getBlocksWithLimit`, puis matérialise chaque slot listé par `getBlock observed` en `Full + Base64 + maxSupportedTransactionVersion = 1 + showRewards = false`. Seules les transactions Legacy/V0/V1 explicitement qualifiées entrent directement dans Common RAW. + +Un slot listé dont `getBlock` retourne `null` reste la tête de reprise du cycle suivant ; il n'est ni considéré vide ni marqué settled. Les slots absents de la liste de découverte sont traités comme non produits/skipped pour ce run. Le slot n'est settled qu'après admission réussie de toutes ses transactions. Le polling reste run-local : aucun checkpoint durable, aucun scan avant la borne initiale et aucun Backfill implicite ne sont créés. Les retries/reroutages HTTP restent possédés par `ksp-onchain-transport-lib`. + ## Runtime et lifecycle Le Worker s'exécute sur le runtime Tokio courant du caller. Il ne crée pas de runtime global et n'expose aucun `JoinHandle` public. @@ -264,7 +286,6 @@ La crate ne dépend pas de Config, Job, `ksp-store-api` directement, backend Sto La verticale actuelle ne possède pas : - plusieurs sources productives simultanées dans `RawTransactionIngestRuntimeResources` ; -- source HTTP live polling Worker ; - sélection Config interne au Worker ; - checkpoint persistent de processing frontier ; - campagne de réparation historique automatique ; diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/USAGE.md b/crates/ksp-worker-raw-transaction-ingest-lib/USAGE.md index 4e4ab21..7f0045d 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/USAGE.md +++ b/crates/ksp-worker-raw-transaction-ingest-lib/USAGE.md @@ -1,5 +1,5 @@ - + # Utilisation de ksp-worker-raw-transaction-ingest-lib @@ -101,7 +101,7 @@ let handle = match ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestWo }; ``` -Les deux entrées exigent un runtime Tokio courant et un `Store` portant exactement le même `RawNetworkId` que les settings. Le démarrage avec ressources exige également que chaque source composée cible ce même réseau. Une exécution productive accepte une source unique Yellowstone, Standard Logs, Standard Block ou Helius Transaction ; une collection de plusieurs sources est validable à la composition mais reste rejetée avant spawn tant que la supervision simultanée n'est pas disponible. +Les deux entrées exigent un runtime Tokio courant et un `Store` portant exactement le même `RawNetworkId` que les settings. Le démarrage avec ressources exige également que chaque source composée cible ce même réseau. Une exécution productive accepte une source unique Yellowstone, Standard Logs, Standard Block, Helius Transaction ou HTTP Block Polling ; une collection de plusieurs sources est validable à la composition mais reste rejetée avant spawn tant que la supervision simultanée n'est pas disponible. ### Source Standard Logs productive @@ -181,6 +181,32 @@ fn helius_transaction_runtime_resources( Le Worker demande la forme Helius `Full` avec `Base64`, `showRewards = false` et `maxSupportedTransactionVersion = 1`, mais ne fait pas confiance au nested payload pour construire directement le Common RAW. Il conserve seulement signature/slot/index et hydrate par `getTransaction observed`. Une notification d'une autre forme est fail-closed. +### Source HTTP Block Polling productive + +Une source HTTP live peut être composée sans WebSocket ni gRPC : + +```rust +fn http_block_polling_runtime_resources( + http_pool: ksp_onchain_transport_lib::HttpTransportPool, + polling_role: ksp_onchain_transport_lib::HttpRoleName, + commitment: ksp_onchain_transport_lib::SolanaCommitment, +) -> ksp_core_lib::Result { + let source = match ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestHttpBlockPollingSource::new( + http_pool, + polling_role, + commitment, + ) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + return std::result::Result::Ok(ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestRuntimeResources::from_http_block_polling_source(source)); +} +``` + +Le rôle HTTP doit disposer, sur un même réseau, de `getSlot`, `getBlocksWithLimit` et `getBlock`. Le commitment est limité à `Confirmed`/`Finalized`. Par défaut, le Worker interroge toutes les secondes et borne la découverte à 128 blocs par cycle. `new_with_limits` permet de choisir une cadence entre 100 ms et 30 s et une limite entre 1 et 1024 blocs par cycle ; les limites de débit physiques restent celles de Transport. + +Au démarrage du run, la première valeur `getSlot(commitment)` devient la borne inférieure stricte du poller. Il ne demande jamais de slot antérieur. `getBlocksWithLimit` détermine les slots réellement matérialisables, puis `getBlock observed` produit directement le Common RAW en Full/Base64 pour Legacy/V0/V1. Un slot listé dont `getBlock` retourne `null` reste la tête de reprise du prochain cycle et n'est jamais transformé en progression silencieuse. + ## Préparer la source Yellowstone La `YellowstoneSubscribeRequest` doit : @@ -205,6 +231,7 @@ Yellowstone Block -> un signal par transaction -> HTTP getTransactio Standard WS logsSubscribe -> context.slot + signature -> HTTP getTransaction -> Common RAW -> admission Standard WS blockSubscribe -> Full/Base64 Legacy|V0|V1 -> Common RAW direct par transaction -> admission Helius transactionSubscribe -> Full envelope -> signature/slot/index -> HTTP getTransaction -> Common RAW -> admission +HTTP live block polling -> getSlot -> getBlocksWithLimit -> getBlock observed -> Common RAW direct Legacy|V0|V1 -> admission Yellowstone BlockMeta -> continuity-only Yellowstone Slot -> continuity-only Yellowstone Account/Ping/Pong/Entry -> sans RAW Transaction dans cette verticale @@ -347,8 +374,8 @@ Le pattern attendu est : ```text Config / application / service owner -> résout endpoints, credentials et rôles - -> construit une source Transport Yellowstone, Standard Logs, Standard Block ou Helius Transaction - -> construit HttpTransportPool + hydration role seulement pour les sources hydratées + -> construit une source Transport Yellowstone, Standard Logs, Standard Block, Helius Transaction ou HTTP Block Polling + -> construit HttpTransportPool + rôle HTTP adapté à la source -> construit le Store -> construit RawTransactionIngestRuntimeResources -> construit RawTransactionIngestSettings diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/src/lib.rs b/crates/ksp-worker-raw-transaction-ingest-lib/src/lib.rs index e175233..a082ec9 100644 --- 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: 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. diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs b/crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs index e0de9af..8c7053c 100644 --- 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,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 { + 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 { + 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, + admission_sender: tokio::sync::mpsc::Sender, + processing_frontier_sender: tokio::sync::watch::Sender, + ) -> 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, } @@ -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>, + received_at: ksp_store_lib::RawTimestamp, +) -> ksp_core_lib::Result> { + 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, + transaction: &ksp_onchain_transport_lib::SolanaBlockTransaction, + position: usize, +) -> ksp_core_lib::Result { + 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_core_lib::Result> { + 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 { + 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); diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/tests/dependency_boundary.rs b/crates/ksp-worker-raw-transaction-ingest-lib/tests/dependency_boundary.rs index 140cdaa..8f0dd7c 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/tests/dependency_boundary.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/tests/dependency_boundary.rs @@ -1,5 +1,5 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/tests/dependency_boundary.rs -// version: 21 +// version: 22 //! Dependency firewall canaries for the RAW transaction ingest Worker foundation. @@ -85,7 +85,6 @@ fn v0_3_12_pre_003_transaction_and_status_adapters_are_private_and_transport_fac "pub struct RawTransactionIngestSourceSignal", "pub(crate) struct RawTransactionIngestSourceSignal", "pub use self::runtime_resources::RawTransactionIngestSourceSignal", - "get_block_observed", "ksp_store_postgres_lib::", "reqwest::", "tonic::", @@ -127,15 +126,7 @@ fn v0_3_12_pre_004_hydration_contract_uses_only_transport_facade_common_raw_and_ } assert!(resources.contains("fn finalize_hydration")); assert!(resources.contains("fn fetch_hydration")); - for forbidden in [ - "get_block_observed", - "ksp_config_lib::", - "ksp_job_backfill_lib::", - "ksp_store_postgres_lib::", - "reqwest::", - "tonic::", - "yellowstone_grpc_proto::", - ] { + for forbidden in ["ksp_config_lib::", "ksp_job_backfill_lib::", "ksp_store_postgres_lib::", "reqwest::", "tonic::", "yellowstone_grpc_proto::"] { assert!(!resources.contains(forbidden) && !root.contains(forbidden), "pre.004 crossed a forbidden boundary: {forbidden}"); } return; @@ -157,15 +148,7 @@ fn v0_3_12_pre_005_block_and_continuity_adapters_remain_private_and_transport_fa ] { assert!(resources.contains(required), "required pre.005 private adapter contract missing: {required}"); } - for forbidden in [ - "get_block_observed", - "ksp_config_lib::", - "ksp_job_backfill_lib::", - "ksp_store_postgres_lib::", - "reqwest::", - "tonic::", - "yellowstone_grpc_proto::", - ] { + for forbidden in ["ksp_config_lib::", "ksp_job_backfill_lib::", "ksp_store_postgres_lib::", "reqwest::", "tonic::", "yellowstone_grpc_proto::"] { assert!(!resources.contains(forbidden) && !root.contains(forbidden), "pre.005 crossed a forbidden boundary: {forbidden}"); } return; @@ -200,7 +183,6 @@ fn v0_3_12_pre_006_productive_source_uses_transport_session_bounded_coalescence_ assert!(resources.contains(required), "required pre.006 productive-source contract missing: {required}"); } for forbidden in [ - "get_block_observed", "ksp_config_lib::", "ksp_job_backfill_lib::", "ksp_store_postgres_lib::", @@ -214,6 +196,49 @@ fn v0_3_12_pre_006_productive_source_uses_transport_session_bounded_coalescence_ return; } +#[test] +fn v0_3_13_pre_006_http_block_polling_reuses_transport_facades_without_becoming_backfill() { + let root = include_str!("../src/lib.rs"); + let resources = include_str!("../src/runtime_resources.rs"); + for required in [ + "RawTransactionIngestHttpBlockPollingSource", + "get_slot(&self.polling_role", + "get_blocks_with_limit(", + "get_block_observed(&self.polling_role", + "let mut next_scan_slot = start_slot", + "next_scan_slot = slot", + "tokio::time::sleep(self.poll_interval)", + "RawAcquisitionOrigin::Live", + "block_polling_get_block", + "SolanaTransactionEncoding::Base64", + "SolanaTransactionDetails::Full", + "std::option::Option::Some(1)", + ] { + assert!(resources.contains(required), "required pre.006 HTTP polling contract missing: {required}"); + } + let source_impl = match resources.split_once("impl crate::RawTransactionIngestHttpBlockPollingSource {") { + std::option::Option::Some((_, tail)) => match tail.split_once("impl std::fmt::Debug for crate::RawTransactionIngestHttpBlockPollingSource") { + std::option::Option::Some((value, _)) => value, + std::option::Option::None => "", + }, + std::option::Option::None => "", + }; + assert_eq!(source_impl.matches("get_block_observed(").count(), 1); + for forbidden in [ + "ksp_config_lib::", + "ksp_job_backfill_lib::", + "ksp_store_postgres_lib::", + "reqwest::", + "tokio_tungstenite::", + "tonic::", + "yellowstone_grpc_proto::", + "unbounded_channel", + ] { + assert!(!source_impl.contains(forbidden) && !root.contains(forbidden), "pre.006 HTTP polling crossed a forbidden boundary: {forbidden}"); + } + return; +} + #[test] fn v0_3_13_pre_003_standard_logs_source_reuses_transport_facades_and_common_hydration_only() { let root = include_str!("../src/lib.rs"); @@ -237,7 +262,6 @@ fn v0_3_13_pre_003_standard_logs_source_reuses_transport_facades_and_common_hydr } assert!(runtime.contains("run_single_live_source")); for forbidden in [ - "get_block_observed", "ksp_config_lib::", "ksp_job_backfill_lib::", "ksp_store_postgres_lib::", @@ -278,7 +302,6 @@ fn v0_3_13_pre_004_standard_block_source_is_direct_raw_and_transport_facade_only assert!(resources.contains(required), "required pre.004 standard block contract missing: {required}"); } for forbidden in [ - "get_block_observed", "ksp_config_lib::", "ksp_job_backfill_lib::", "ksp_store_postgres_lib::", diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/tests/hardening.rs b/crates/ksp-worker-raw-transaction-ingest-lib/tests/hardening.rs index 5d82ead..94caa89 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/tests/hardening.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/tests/hardening.rs @@ -1,5 +1,5 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/tests/hardening.rs -// version: 14 +// version: 15 //! External public, security, redaction and release-boundary hardening canaries for `pre.010`. @@ -400,7 +400,7 @@ fn v0_3_12_pre_006_runtime_resource_contract_opens_one_supervised_transport_sour } assert!(runtime.contains("start_with_runtime_resources")); assert!(runtime.contains("run_single_live_source(source_settings, stop_receiver, admission_sender, processing_frontier_sender)")); - for forbidden in ["get_block_observed", "ksp_config_lib::", "ksp_store_postgres_lib::", "reqwest::", "tonic::", "yellowstone_grpc_proto::"] { + for forbidden in ["ksp_config_lib::", "ksp_store_postgres_lib::", "reqwest::", "tonic::", "yellowstone_grpc_proto::"] { assert!(!resources.contains(forbidden) && !runtime.contains(forbidden), "pre.006 runtime source crossed a forbidden boundary: {forbidden}"); } return; @@ -652,6 +652,48 @@ fn v0_3_13_pre_005_helius_transaction_redaction_full_reference_and_tier_neutrali return; } +#[test] +fn v0_3_13_pre_006_http_block_polling_is_bounded_run_local_stop_preemptible_and_redacted() { + let resources = include_str!("../src/runtime_resources.rs"); + for required in [ + "DEFAULT_RAW_TRANSACTION_INGEST_HTTP_POLL_INTERVAL", + "DEFAULT_RAW_TRANSACTION_INGEST_HTTP_POLL_MAX_BLOCKS_PER_CYCLE", + "MIN_RAW_TRANSACTION_INGEST_HTTP_POLL_INTERVAL", + "MAX_RAW_TRANSACTION_INGEST_HTTP_POLL_INTERVAL", + "MIN_RAW_TRANSACTION_INGEST_HTTP_POLL_MAX_BLOCKS_PER_CYCLE", + "MAX_RAW_TRANSACTION_INGEST_HTTP_POLL_MAX_BLOCKS_PER_CYCLE", + "let mut next_scan_slot = start_slot", + "next_scan_slot = slot", + "tokio::time::sleep(self.poll_interval)", + "processing_frontier.observe_settled(slot)", + "source.http_block_polling_transaction_version_unqualified", + "source.http_block_polling_transaction_version_unsupported", + ] { + assert!(resources.contains(required), "required pre.006 HTTP polling hardening guard missing: {required}"); + } + let source_impl = match resources.split_once("impl crate::RawTransactionIngestHttpBlockPollingSource {") { + std::option::Option::Some((_, tail)) => match tail.split_once("impl std::fmt::Debug for crate::RawTransactionIngestHttpBlockPollingSource") { + std::option::Option::Some((value, _)) => value, + std::option::Option::None => "", + }, + std::option::Option::None => "", + }; + assert_eq!(source_impl.matches("get_block_observed(").count(), 1); + assert!(!source_impl.contains("tokio::spawn"), "HTTP polling must not spawn one task per tick"); + assert!(!source_impl.contains("observe_pending("), "HTTP block polling is direct RAW and must not inflate hydration_pending"); + let source_struct = match resources.split_once("pub struct RawTransactionIngestHttpBlockPollingSource {") { + std::option::Option::Some((_, tail)) => match tail.split_once("impl crate::RawTransactionIngestHttpBlockPollingSource") { + std::option::Option::Some((value, _)) => value, + std::option::Option::None => "", + }, + std::option::Option::None => "", + }; + for forbidden in ["url:", "api_key", "credential", "secret", "transaction:", "payload:", "tier:"] { + assert!(!source_struct.contains(forbidden), "sensitive/raw material stored in HTTP polling source: {forbidden}"); + } + return; +} + #[test] fn v0_3_12_pre_009_hydration_retry_ownership_and_no_orphan_cleanup_are_explicit() { let resources = include_str!("../src/runtime_resources.rs"); @@ -665,8 +707,16 @@ fn v0_3_12_pre_009_hydration_retry_ownership_and_no_orphan_cleanup_are_explicit( ] { assert!(resources.contains(required), "required pre.009 no-orphan/backpressure guard missing: {required}"); } - for forbidden in ["tokio::time::sleep", "tokio::time::interval", "get_block_observed", "unbounded_channel"] { - assert!(!resources.contains(forbidden), "Worker introduced forbidden retry/unbounded behavior: {forbidden}"); + let hydration = match resources.split_once("struct RawTransactionIngestHydrationCoordinator {") { + std::option::Option::Some((_, tail)) => match tail.split_once("fn hydration_method_code(") { + std::option::Option::Some((value, _)) => value, + std::option::Option::None => "", + }, + std::option::Option::None => "", + }; + for forbidden in ["tokio::time::sleep", "tokio::time::interval", "get_block_observed"] { + assert!(!hydration.contains(forbidden), "hydration coordinator introduced forbidden retry behavior: {forbidden}"); } + assert!(!resources.contains("unbounded_channel"), "Worker introduced an unbounded channel"); return; } diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/tests/public_api.rs b/crates/ksp-worker-raw-transaction-ingest-lib/tests/public_api.rs index 0313952..4b0070d 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/tests/public_api.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/tests/public_api.rs @@ -1,5 +1,5 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/tests/public_api.rs -// version: 15 +// version: 16 //! External public-surface proofs for the RAW transaction ingest Worker foundation. @@ -276,6 +276,43 @@ fn v0_3_13_pre_004_standard_block_runtime_resource_surface_is_typed_and_transpor return; } +#[test] +fn v0_3_13_pre_006_http_block_polling_runtime_resource_surface_is_bounded_and_transport_owned() { + let _source_new: fn( + ksp_onchain_transport_lib::HttpTransportPool, + ksp_onchain_transport_lib::HttpRoleName, + ksp_onchain_transport_lib::SolanaCommitment, + ) -> ksp_core_lib::Result = + ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestHttpBlockPollingSource::new; + let _source_new_with_limits: fn( + ksp_onchain_transport_lib::HttpTransportPool, + ksp_onchain_transport_lib::HttpRoleName, + ksp_onchain_transport_lib::SolanaCommitment, + std::time::Duration, + u16, + ) -> ksp_core_lib::Result = + ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestHttpBlockPollingSource::new_with_limits; + let _resources_from_polling: fn( + ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestHttpBlockPollingSource, + ) -> ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestRuntimeResources = + ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestRuntimeResources::from_http_block_polling_source; + let _push_polling: fn( + &mut ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestRuntimeResources, + ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestHttpBlockPollingSource, + ) -> ksp_core_lib::Result<()> = ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestRuntimeResources::try_push_http_block_polling_source; + assert_eq!(ksp_worker_raw_transaction_ingest_lib::DEFAULT_RAW_TRANSACTION_INGEST_HTTP_POLL_INTERVAL, std::time::Duration::from_secs(1)); + assert_eq!(ksp_worker_raw_transaction_ingest_lib::DEFAULT_RAW_TRANSACTION_INGEST_HTTP_POLL_MAX_BLOCKS_PER_CYCLE, 128); + assert_eq!(ksp_worker_raw_transaction_ingest_lib::MIN_RAW_TRANSACTION_INGEST_HTTP_POLL_INTERVAL, std::time::Duration::from_millis(100)); + assert_eq!(ksp_worker_raw_transaction_ingest_lib::MAX_RAW_TRANSACTION_INGEST_HTTP_POLL_INTERVAL, std::time::Duration::from_secs(30)); + assert_eq!(ksp_worker_raw_transaction_ingest_lib::MIN_RAW_TRANSACTION_INGEST_HTTP_POLL_MAX_BLOCKS_PER_CYCLE, 1); + assert_eq!(ksp_worker_raw_transaction_ingest_lib::MAX_RAW_TRANSACTION_INGEST_HTTP_POLL_MAX_BLOCKS_PER_CYCLE, 1024); + let resources = include_str!("../src/runtime_resources.rs"); + for forbidden in ["pub fn http_pool(", "pub fn polling_role(", "pub fn source_key(", "pub fn profile_fingerprint("] { + assert!(!resources.contains(forbidden), "HTTP polling source implementation escape hatch present: {forbidden}"); + } + return; +} + #[test] fn v0_3_12_pre_007_processing_frontier_snapshot_getters_are_public_and_processing_only() { let _hydration_pending: fn(&ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot) -> usize = diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/tests/release_completeness.rs b/crates/ksp-worker-raw-transaction-ingest-lib/tests/release_completeness.rs index 447a512..e42ce72 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/tests/release_completeness.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/tests/release_completeness.rs @@ -1,5 +1,5 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/tests/release_completeness.rs -// version: 13 +// version: 14 //! Release-completeness canaries through the `pre.010` public/release/security hardening tranche. @@ -60,6 +60,8 @@ fn pre_010_public_root_export_inventory_is_exact() { exports, std::vec![ "DEFAULT_RAW_TRANSACTION_INGEST_ADMISSION_QUEUE_CAPACITY", + "DEFAULT_RAW_TRANSACTION_INGEST_HTTP_POLL_INTERVAL", + "DEFAULT_RAW_TRANSACTION_INGEST_HTTP_POLL_MAX_BLOCKS_PER_CYCLE", "DEFAULT_RAW_TRANSACTION_INGEST_PERSISTENCE_CONCURRENCY", "DEFAULT_RAW_TRANSACTION_INGEST_SHUTDOWN_DRAIN_TIMEOUT", "ERROR_CODE_RAW_TRANSACTION_INGEST_CONTENT_CONFLICT", @@ -70,15 +72,20 @@ fn pre_010_public_root_export_inventory_is_exact() { "ERROR_CODE_RAW_TRANSACTION_INGEST_SOURCE_FAILED", "ERROR_CODE_RAW_TRANSACTION_INGEST_STORE_FAILED", "MAX_RAW_TRANSACTION_INGEST_ADMISSION_QUEUE_CAPACITY", + "MAX_RAW_TRANSACTION_INGEST_HTTP_POLL_INTERVAL", + "MAX_RAW_TRANSACTION_INGEST_HTTP_POLL_MAX_BLOCKS_PER_CYCLE", "MAX_RAW_TRANSACTION_INGEST_LIVE_SOURCES", "MAX_RAW_TRANSACTION_INGEST_PERSISTENCE_CONCURRENCY", "MAX_RAW_TRANSACTION_INGEST_SHUTDOWN_DRAIN_TIMEOUT", "MIN_RAW_TRANSACTION_INGEST_ADMISSION_QUEUE_CAPACITY", + "MIN_RAW_TRANSACTION_INGEST_HTTP_POLL_INTERVAL", + "MIN_RAW_TRANSACTION_INGEST_HTTP_POLL_MAX_BLOCKS_PER_CYCLE", "MIN_RAW_TRANSACTION_INGEST_PERSISTENCE_CONCURRENCY", "MIN_RAW_TRANSACTION_INGEST_SHUTDOWN_DRAIN_TIMEOUT", "RAW_TRANSACTION_INGEST_WORKER_KIND_CODE", "RawTransactionIngestHandle", "RawTransactionIngestHeliusTransactionSource", + "RawTransactionIngestHttpBlockPollingSource", "RawTransactionIngestRuntimeResources", "RawTransactionIngestSettings", "RawTransactionIngestSnapshot", @@ -119,6 +126,7 @@ fn pre_010_external_hardening_suite_is_present_and_scoped() { "v0_3_13_pre_003_standard_logs_redaction_and_reference_only_contract_are_explicit", "v0_3_13_pre_004_standard_block_redaction_version_and_null_guards_are_explicit", "v0_3_13_pre_005_helius_transaction_redaction_full_reference_and_tier_neutrality_are_explicit", + "v0_3_13_pre_006_http_block_polling_is_bounded_run_local_stop_preemptible_and_redacted", ] { assert!(hardening.contains(required), "required pre.010 hardening canary missing: {required}"); } @@ -135,6 +143,7 @@ fn pre_010_external_hardening_suite_is_present_and_scoped() { assert!(dependency_boundary.contains("v0_3_13_pre_003_standard_logs_source_reuses_transport_facades_and_common_hydration_only")); assert!(dependency_boundary.contains("v0_3_13_pre_004_standard_block_source_is_direct_raw_and_transport_facade_only")); assert!(dependency_boundary.contains("v0_3_13_pre_005_helius_transaction_source_reuses_transport_facade_and_common_hydration_only")); + assert!(dependency_boundary.contains("v0_3_13_pre_006_http_block_polling_reuses_transport_facades_without_becoming_backfill")); let public_api = include_str!("public_api.rs"); assert!(public_api.contains("pre_003_kind_code_and_settings_are_consumable_from_crate_root")); assert!(public_api.contains("pre_004_start_handle_and_terminal_future_are_consumable_without_public_join_handle")); @@ -144,6 +153,7 @@ fn pre_010_external_hardening_suite_is_present_and_scoped() { assert!(public_api.contains("v0_3_13_pre_002_multi_source_runtime_resource_surface_is_bounded_and_source_neutral")); assert!(public_api.contains("v0_3_13_pre_003_standard_logs_runtime_resource_surface_is_typed_and_transport_owned")); assert!(public_api.contains("v0_3_13_pre_004_standard_block_runtime_resource_surface_is_typed_and_transport_owned")); + assert!(public_api.contains("v0_3_13_pre_006_http_block_polling_runtime_resource_surface_is_bounded_and_transport_owned")); assert!(public_api.contains("v0_3_12_pre_007_processing_frontier_snapshot_getters_are_public_and_processing_only")); assert!(public_api.contains("pre_008_snapshot_surface_and_common_projection_are_public_and_stable")); return; diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime_resources.rs b/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime_resources.rs index bb54aa5..758afb6 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime_resources.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime_resources.rs @@ -1,5 +1,5 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime_resources.rs -// version: 17 +// version: 18 fn grpc_endpoint(cluster: &str) -> std::option::Option { return grpc_endpoint_with_identity(cluster, "yellowstone-fixture", "fixture-provider"); @@ -62,6 +62,54 @@ fn http_pool(cluster: &str, role_name: &str, request_kind: &str) -> std::option: }; } +fn http_polling_pool_with_identity( + cluster: &str, + role_name: &str, + endpoint_name: &str, + provider: &str, + request_kinds: &[&str], +) -> std::option::Option { + let url = match ksp_onchain_transport_lib::HttpEndpointUrl::parse("https://fixture.invalid/poll?token=HTTP-POLL-SECRET-CANARY") { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return std::option::Option::None, + }; + let role = ksp_onchain_transport_lib::HttpEndpointRoleSettings::new( + ksp_onchain_transport_lib::HttpRoleName::new(role_name), + true, + request_kinds.iter().map(|value| return ksp_onchain_transport_lib::HttpRequestKind::new(*value)).collect(), + 10, + ksp_onchain_transport_lib::HttpRoleLimits::new( + std::option::Option::None, + std::option::Option::None, + std::option::Option::None, + std::option::Option::None, + ), + ); + let endpoint = ksp_onchain_transport_lib::HttpEndpointSettings::new( + endpoint_name, + true, + ksp_onchain_transport_lib::HttpProviderName::new(provider), + ksp_onchain_transport_lib::HttpClusterName::new(cluster), + url, + std::time::Duration::from_secs(1), + std::time::Duration::from_secs(2), + std::option::Option::Some(4), + std::vec![role], + ); + let settings = ksp_onchain_transport_lib::HttpTransportSettings::new( + std::vec![endpoint], + ksp_onchain_transport_lib::HttpRetrySettings::new(1, std::time::Duration::from_millis(10), std::time::Duration::from_millis(20)), + ); + return match ksp_onchain_transport_lib::HttpTransportPool::new(settings) { + std::result::Result::Ok(value) => std::option::Option::Some(value), + std::result::Result::Err(_) => std::option::Option::None, + }; +} + +fn http_polling_pool(cluster: &str) -> std::option::Option { + return http_polling_pool_with_identity(cluster, "live-poll", "poll-fixture", "fixture-provider", &["get_block", "get_blocks_with_limit", "get_slot"]); +} + fn ws_endpoint( cluster: &str, endpoint_name: &str, @@ -2670,3 +2718,183 @@ async fn pre_009_abort_joins_in_flight_hydration_and_clears_pending_projection() assert_eq!(cleaned.processing_frontier_slot(), std::option::Option::None); return; } + +#[test] +fn v0_3_13_pre_006_http_block_polling_source_validates_bounds_capabilities_and_logical_identity() { + let pool = match http_polling_pool("devnet") { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let source = match crate::RawTransactionIngestHttpBlockPollingSource::new( + pool.clone(), + ksp_onchain_transport_lib::HttpRoleName::new("live-poll"), + ksp_onchain_transport_lib::SolanaCommitment::Confirmed, + ) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return, + }; + let same_identity_different_timing = match crate::RawTransactionIngestHttpBlockPollingSource::new_with_limits( + pool, + ksp_onchain_transport_lib::HttpRoleName::new("live-poll"), + ksp_onchain_transport_lib::SolanaCommitment::Confirmed, + std::time::Duration::from_secs(2), + 256, + ) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return, + }; + assert_eq!(source.source_key, same_identity_different_timing.source_key); + assert_eq!(source.network.as_str(), "devnet"); + assert_eq!(source.poll_interval, crate::DEFAULT_RAW_TRANSACTION_INGEST_HTTP_POLL_INTERVAL); + assert_eq!(source.max_discovered_blocks_per_cycle, crate::DEFAULT_RAW_TRANSACTION_INGEST_HTTP_POLL_MAX_BLOCKS_PER_CYCLE); + let different_endpoint_pool = match http_polling_pool_with_identity( + "devnet", + "live-poll", + "poll-fixture-other", + "fixture-provider", + &["get_block", "get_blocks_with_limit", "get_slot"], + ) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let different_endpoint = match crate::RawTransactionIngestHttpBlockPollingSource::new( + different_endpoint_pool, + ksp_onchain_transport_lib::HttpRoleName::new("live-poll"), + ksp_onchain_transport_lib::SolanaCommitment::Confirmed, + ) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return, + }; + assert_ne!(source.source_key, different_endpoint.source_key); + let processed = crate::RawTransactionIngestHttpBlockPollingSource::new( + match http_polling_pool("devnet") { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }, + ksp_onchain_transport_lib::HttpRoleName::new("live-poll"), + ksp_onchain_transport_lib::SolanaCommitment::Processed, + ); + assert!(processed.is_err()); + let too_fast = crate::RawTransactionIngestHttpBlockPollingSource::new_with_limits( + match http_polling_pool("devnet") { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }, + ksp_onchain_transport_lib::HttpRoleName::new("live-poll"), + ksp_onchain_transport_lib::SolanaCommitment::Confirmed, + std::time::Duration::from_millis(99), + 128, + ); + assert!(too_fast.is_err()); + let too_slow = crate::RawTransactionIngestHttpBlockPollingSource::new_with_limits( + match http_polling_pool("devnet") { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }, + ksp_onchain_transport_lib::HttpRoleName::new("live-poll"), + ksp_onchain_transport_lib::SolanaCommitment::Confirmed, + std::time::Duration::from_millis(30_001), + 128, + ); + assert!(too_slow.is_err()); + for invalid_limit in [0_u16, 1025_u16] { + let invalid = crate::RawTransactionIngestHttpBlockPollingSource::new_with_limits( + match http_polling_pool("devnet") { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }, + ksp_onchain_transport_lib::HttpRoleName::new("live-poll"), + ksp_onchain_transport_lib::SolanaCommitment::Confirmed, + std::time::Duration::from_secs(1), + invalid_limit, + ); + assert!(invalid.is_err()); + } + let missing_capability_pool = match http_polling_pool_with_identity("devnet", "live-poll", "poll-missing", "fixture-provider", &["get_block", "get_slot"]) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let missing_capability = crate::RawTransactionIngestHttpBlockPollingSource::new( + missing_capability_pool, + ksp_onchain_transport_lib::HttpRoleName::new("live-poll"), + ksp_onchain_transport_lib::SolanaCommitment::Confirmed, + ); + assert!(missing_capability.is_err()); + let debug = std::format!("{source:?}"); + assert!(!debug.contains("HTTP-POLL-SECRET-CANARY")); + assert!(!debug.contains("fixture.invalid")); + return; +} + +#[test] +fn v0_3_13_pre_006_http_block_polling_discovery_never_moves_before_run_frontier_and_requires_strict_order() { + assert!(super::validate_http_block_polling_discovery(100, &[]).is_ok()); + assert!(super::validate_http_block_polling_discovery(100, &[100, 102, 105]).is_ok()); + assert!(super::validate_http_block_polling_discovery(100, &[99]).is_err()); + assert!(super::validate_http_block_polling_discovery(100, &[100, 100]).is_err()); + assert!(super::validate_http_block_polling_discovery(100, &[102, 101]).is_err()); + return; +} + +#[test] +fn v0_3_13_pre_006_http_block_polling_qualifies_legacy_v0_v1_and_rejects_ambiguous_or_future_versions() { + let cases = [ + (ksp_onchain_transport_lib::SolanaTransactionVersion::Legacy, ksp_raw_transaction_lib::RawTransactionVersion::Legacy), + (ksp_onchain_transport_lib::SolanaTransactionVersion::Number(0), ksp_raw_transaction_lib::RawTransactionVersion::Number(0)), + (ksp_onchain_transport_lib::SolanaTransactionVersion::Number(1), ksp_raw_transaction_lib::RawTransactionVersion::Number(1)), + ]; + for (wire, expected) in cases { + let qualified = super::qualify_http_block_polling_version(&ksp_onchain_transport_lib::SolanaWireField::Value(wire)); + let qualified = match qualified { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return, + }; + let qualified = match qualified { + ksp_raw_transaction_lib::RawTransactionWireField::Value(value) => value, + _ => return, + }; + assert_eq!(qualified, expected); + } + assert!( + super::qualify_http_block_polling_version(&ksp_onchain_transport_lib::SolanaWireField::Value( + ksp_onchain_transport_lib::SolanaTransactionVersion::Number(2), + )) + .is_err() + ); + assert!(super::qualify_http_block_polling_version(&ksp_onchain_transport_lib::SolanaWireField::Omitted).is_err()); + assert!(super::qualify_http_block_polling_version(&ksp_onchain_transport_lib::SolanaWireField::Null).is_err()); + return; +} + +#[test] +fn v0_3_13_pre_006_runtime_resources_reject_duplicate_http_polling_identity_even_when_timing_differs() { + let first = match crate::RawTransactionIngestHttpBlockPollingSource::new( + match http_polling_pool("devnet") { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }, + ksp_onchain_transport_lib::HttpRoleName::new("live-poll"), + ksp_onchain_transport_lib::SolanaCommitment::Confirmed, + ) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return, + }; + let duplicate = match crate::RawTransactionIngestHttpBlockPollingSource::new_with_limits( + match http_polling_pool("devnet") { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }, + ksp_onchain_transport_lib::HttpRoleName::new("live-poll"), + ksp_onchain_transport_lib::SolanaCommitment::Confirmed, + std::time::Duration::from_secs(2), + 256, + ) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return, + }; + let mut resources = crate::RawTransactionIngestRuntimeResources::from_http_block_polling_source(first); + let pushed = resources.try_push_http_block_polling_source(duplicate); + assert!(pushed.is_err()); + assert_eq!(resources.source_count(), 1); + return; +} diff --git a/deltas/0.3.13/pre.006.md b/deltas/0.3.13/pre.006.md new file mode 100644 index 0000000..8790c6e --- /dev/null +++ b/deltas/0.3.13/pre.006.md @@ -0,0 +1,322 @@ + + + +# Delta 0.3.13-pre.006 — HTTP live block polling run-local + +## Base requise + +```text +livraison précédente : 0.3.13-pre.005-fix.001 +Cargo base : 0.3.13-pre.5.fix.1 +delta base : deltas/0.3.13/pre.005-fix.001.md +archive delta base : ksp-general-0.3.13-pre.005-fix.001.zip +``` + +Le gate opérateur reçu le 10 septembre 2026 sur cette base est entièrement vert pour les commandes exécutées : `cargo fmt`, `cargo fmt --check`, audits Rust/Markdown, `cargo check --workspace`, Clippy strict et toutes les suites de `ksp-worker-raw-transaction-ingest-lib` (`84` unit, `4` cross-layer, `12` dependency-boundary, `19` hardening, `13` public-api, `4` release-completeness, `0` doc-test). Cette exécution ne contenait ni le test Transport ni les commandes `cargo tree`; ils ne sont donc pas déclarés PASS pour ce gate précis. + +## Objectif + +Ajouter la source HTTP live prévue par la tranche sans la transformer en Backfill : + +```text +getSlot(commitment) + -> borne inférieure du run +getBlocksWithLimit(next_scan_slot, bounded_limit, commitment) + -> découverte des blocs disponibles +getBlock observed(slot, Full/Base64/maxV1/rewards=false) + -> Common RAW direct Legacy/V0/V1 + -> admission centrale Worker + -> Store +``` + +Le polling ne remonte jamais avant la borne observée au démarrage et ne crée aucun checkpoint durable inter-run. + +## Version + +```text +livraison : 0.3.13-pre.006 +workspace.package.version : 0.3.13-pre.6 +archive : ksp-general-0.3.13-pre.006.zip +``` + +`Cargo.toml` passe également de la version de fichier `549` à `550`. + +## Source HTTP Block Polling + +Nouvelle surface publique Worker : + +```text +RawTransactionIngestHttpBlockPollingSource +RawTransactionIngestRuntimeResources::from_http_block_polling_source +RawTransactionIngestRuntimeResources::try_push_http_block_polling_source +DEFAULT_RAW_TRANSACTION_INGEST_HTTP_POLL_INTERVAL +MIN_RAW_TRANSACTION_INGEST_HTTP_POLL_INTERVAL +MAX_RAW_TRANSACTION_INGEST_HTTP_POLL_INTERVAL +DEFAULT_RAW_TRANSACTION_INGEST_HTTP_POLL_MAX_BLOCKS_PER_CYCLE +MIN_RAW_TRANSACTION_INGEST_HTTP_POLL_MAX_BLOCKS_PER_CYCLE +MAX_RAW_TRANSACTION_INGEST_HTTP_POLL_MAX_BLOCKS_PER_CYCLE +``` + +La construction caller-owned reçoit un `HttpTransportPool`, un `HttpRoleName` et un commitment `Confirmed` ou `Finalized`. Elle est sans I/O et vérifie statiquement que le rôle possède des routes compatibles `getSlot`, `getBlocksWithLimit` et `getBlock` sur un même réseau. + +Bornes publiques : + +```text +poll interval : défaut 1 s ; min 100 ms ; max 30 s +max discovered blocks/cycle : défaut 128 ; min 1 ; max 1024 +``` + +Les limitations physiques, la sélection d'endpoint, le rate-limit, les retries et le backoff HTTP restent possédés par `ksp-onchain-transport-lib`. + +## Borne de run et progression + +Au démarrage : + +```text +start_slot = getSlot(commitment) +next_scan_slot = start_slot +``` + +Le Worker ne demande aucun slot inférieur à `start_slot`. Pour chaque cycle, il relit le tip courant, puis appelle `getBlocksWithLimit` depuis `next_scan_slot`. La réponse doit être strictement croissante et ne peut contenir aucun slot inférieur à la frontier de scan. + +Chaque slot listé est matérialisé par `getBlock observed`. Si le résultat vaut `null` : + +```text +next_scan_slot = slot +aucun settled +aucun Common RAW +reprise au cycle suivant +``` + +Le cas `null` n'est donc ni assimilé à un bloc vide ni converti en progression silencieuse. + +Les slots non listés par `getBlocksWithLimit` sont traités comme non produits/skipped dans le run courant. Lorsque la liste reçue ne remplit pas le batch demandé et qu'aucun slot n'est bloqué par `null`, la frontier peut rejoindre le tip observé du cycle sans lancer de scan historique implicite. + +## RAW-direct et version + +Le poller demande exactement : + +```text +encoding = Base64 +transactionDetails = Full +maxSupportedTransactionVersion = 1 +showRewards = false +commitment = Confirmed | Finalized +``` + +Chaque transaction qualifiée produit directement `RawTransactionMaterial` avec : + +```text +signature embarquée dans le wire Base64 +slot +blockTime +meta +version Legacy | 0 | 1 +transactionIndex dérivé de la position dans le bloc +``` + +Les cas suivants sont fail-closed : transaction non Base64, transactions omises/nulles, version omise/nulle, version `>1`, signature invalide ou index non représentable en `u32`. + +La voie HTTP Block Polling est RAW-direct et n'utilise pas le coordinateur d'hydration. Le slot n'est marqué `settled` qu'après l'envoi réussi de toutes les transactions du bloc vers l'admission centrale ; `hydration_pending` n'est jamais artificiellement incrémenté. + +## Provenance observée + +La provenance de chaque transaction reprend l'endpoint et le provider réellement gagnants de `getBlock observed` : + +```text +origin = Live +protocol = solana_http +method = block_polling_get_block +endpoint = endpoint observé +provider = provider observé +commitment = source commitment +capture session = WorkerId +``` + +Aucun URL, header HTTP ou corps brut n'est exposé par `HttpObservedValue` ou les diagnostics Worker. + +## Identité logique + +Le `source_key` HTTP polling est dérivé de : + +```text +domain KSP live-source v1 +family = http_block_polling +network +polling role +commitment +private route-profile fingerprint +SHA-256 -> [u8; 32] +``` + +L'empreinte de profil est construite à partir des identités sûres cluster/provider/endpoint et des capacités pertinentes de chaque route active, après tri déterministe. La cadence et le batch de découverte n'entrent volontairement pas dans l'identité logique : modifier un réglage de timing ne crée pas une deuxième source conceptuelle. + +Le bound global `1..32`, l'invariant réseau unique et le rejet transactionnel des doublons de `pre.002` s'appliquent à cette famille. + +## Activation runtime + +Le contrat reste volontairement mono-source avant `pre.007` : + +```text +1 source Yellowstone : productif +1 source Standard Logs : productif +1 source Standard Block : productif +1 source Helius Transaction : productif +1 source HTTP Block Polling : productif +2..32 sources : composition validable +start 2..32 : runtime_resources.multi_source_activation_pending avant spawn +``` + +Aucune source configurée n'est ignorée silencieusement. + +## Preuves ajoutées + +Unit tests Worker : + +```text +bornes cadence/batch et commitment +présence exacte des capacités HTTP requises +source_key stable malgré changement de timing et différent si route logique différente +redaction URL sensible dans Debug +validation stricte de la découverte croissante +qualification Legacy/V0/V1 et rejet version ambiguë/future +rejet transactionnel d'une identité HTTP polling dupliquée +``` + +Canaris externes : + +```text +public_api : constructeurs/from/push et bornes publiques disponibles depuis crate root +dependency_boundary : Transport facade uniquement, aucun Config/Backfill/backend/direct HTTP client +hardening : run-local, timer borné, getBlock observed unique, aucun task par tick, aucun hydration_pending direct +release_completeness : exports et canaris pre.006 obligatoires +``` + +Les anciens canaris qui interdisaient globalement `get_block_observed` ou `tokio::time::sleep` sont resserrés sur le coordinateur d'hydration : ces primitives restent interdites dans l'hydration mais sont désormais légitimes, explicites et bornées dans la seule source HTTP polling. + +## Documentation + +`README.md`, le README du crate, `USAGE.md`, le plan et le journal de validation sont alignés sur la cinquième famille productive. `USAGE.md` reste version-neutral : il décrit la construction et le comportement sans journaliser le numéro de prerelease. + +Une incohérence du README de crate, qui classait encore le live HTTP polling parmi les éléments hors périmètre, est corrigée dans cette tranche. + +## Fichiers modifiés + +```text +Cargo.toml +README.md +crates/ksp-worker-raw-transaction-ingest-lib/README.md +crates/ksp-worker-raw-transaction-ingest-lib/USAGE.md +crates/ksp-worker-raw-transaction-ingest-lib/src/lib.rs +crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs +crates/ksp-worker-raw-transaction-ingest-lib/tests/dependency_boundary.rs +crates/ksp-worker-raw-transaction-ingest-lib/tests/hardening.rs +crates/ksp-worker-raw-transaction-ingest-lib/tests/public_api.rs +crates/ksp-worker-raw-transaction-ingest-lib/tests/release_completeness.rs +crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime_resources.rs +docs/plans/034-V0_3_13_MULTI_SOURCE_LIVE_CONVERGENCE_PLAN.md +docs/validation/030-V0_3_13_MULTI_SOURCE_LIVE_CONVERGENCE.md +``` + +## Fichiers ajoutés + +```text +deltas/0.3.13/pre.006.md +``` + +## Fichiers supprimés + +```text +aucun +``` + +## Non-claims + +`pre.006` ne prétend pas : + +```text +supervision simultanée de plusieurs sources +checkpoint durable du polling +Backfill implicite +scan de slots antérieurs au start_slot du run +retry/backoff HTTP implémenté dans Worker +client reqwest direct dans Worker +Config lu dans Worker +getBlock=null assimilé à un bloc vide +transaction version >1 ou ambiguë acceptée +modification de ksp-onchain-transport-lib +nouvelle dépendance externe +``` + +## Validation dans l'environnement d'assemblage + +Le toolchain Rust/Cargo/Rustfmt n'est pas disponible dans l'environnement d'assemblage. Les commandes Cargo post-`pre.006` restent donc `NON EXÉCUTÉ LOCAL`. Les audits statiques KSP, le scan normatif, le contrôle des versions de fichiers et le diff exhaustif sont exécutés avant packaging. + +## Gate opérateur requis avant pre.007 + +```bash +cargo fmt --all +cargo fmt --all -- --check +python3 scripts/audit_rust_workspace_rules.py +python3 scripts/audit_markdown_tables.py README.md RULES.md ROADMAP.md CHANGELOG.md docs prompts crates deltas +cargo check --workspace +cargo clippy --workspace --all-targets --all-features -- -D warnings +cargo test -p ksp-onchain-transport-lib +cargo test -p ksp-worker-raw-transaction-ingest-lib +cargo tree -p ksp-worker-raw-transaction-ingest-lib --edges normal +cargo tree -p ksp-worker-raw-transaction-ingest-lib -e features +cargo tree --duplicates +``` + +`pre.007` ne doit pas commencer si ce gate révèle une anomalie. + +## Résultats statiques finaux de l'assemblage + +Les valeurs ci-dessous sont mises à jour après la dernière passe d'audits sur l'état exact livré. + +```text +General Rust rule audit: clean +Rust export completeness audit: 0 candidate(s) +KSP workspace Rust rule audit: clean +Markdown table audit: clean (340 tables, 839 files) +Normative rule definitions: 489 +Unique normative IDs: 489 +Duplicates: 0 +``` + +## Contrôle exhaustif avant packaging + +Comparaison avec la base `0.3.13-pre.005-fix.001` : + +```text +added : 1 + deltas/0.3.13/pre.006.md +modified : 13 + Cargo.toml + README.md + crates/ksp-worker-raw-transaction-ingest-lib/README.md + crates/ksp-worker-raw-transaction-ingest-lib/USAGE.md + crates/ksp-worker-raw-transaction-ingest-lib/src/lib.rs + crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs + crates/ksp-worker-raw-transaction-ingest-lib/tests/dependency_boundary.rs + crates/ksp-worker-raw-transaction-ingest-lib/tests/hardening.rs + crates/ksp-worker-raw-transaction-ingest-lib/tests/public_api.rs + crates/ksp-worker-raw-transaction-ingest-lib/tests/release_completeness.rs + crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime_resources.rs + docs/plans/034-V0_3_13_MULTI_SOURCE_LIVE_CONVERGENCE_PLAN.md + docs/validation/030-V0_3_13_MULTI_SOURCE_LIVE_CONVERGENCE.md +deleted : 0 +``` + +Les `13/13` fichiers existants modifiés portent une version de fichier exactement `+1`. Le diff de `Cargo.toml` est limité au header `549 -> 550` et à `workspace.package.version = "0.3.13-pre.6"`; aucun membre workspace ni aucune dépendance n'est modifié. + +La revue statique ciblée confirme dans le code Worker : + +```text +get_block_observed( : 1 occurrence, source HTTP polling uniquement +get_blocks_with_limit( : 1 occurrence, source HTTP polling uniquement +tokio::time::sleep : 1 occurrence, cadence du poller uniquement +aucun reqwest/tokio-tungstenite/tonic/yellowstone-grpc-proto direct +aucun ksp-config-lib/ksp-job-backfill-lib/ksp-store-postgres-lib direct +aucun unbounded_channel +``` diff --git a/docs/plans/034-V0_3_13_MULTI_SOURCE_LIVE_CONVERGENCE_PLAN.md b/docs/plans/034-V0_3_13_MULTI_SOURCE_LIVE_CONVERGENCE_PLAN.md index de2335e..7096fe1 100644 --- a/docs/plans/034-V0_3_13_MULTI_SOURCE_LIVE_CONVERGENCE_PLAN.md +++ b/docs/plans/034-V0_3_13_MULTI_SOURCE_LIVE_CONVERGENCE_PLAN.md @@ -1,5 +1,5 @@ - + # Plan v0.3.13 — WS standard / Helius / HTTP live + convergence multi-source RawTransaction @@ -1042,4 +1042,106 @@ cargo tree --duplicates ``` `pre.006` ne commence pas tant qu'une anomalie de format, règle, compilation, test ou frontière de dépendances subsiste. +## 50. Gate opérateur pre.005-fix.001 reçu avant pre.006 + +Le gate opérateur communiqué le 10 septembre 2026 est entièrement vert après `pre.005-fix.001` : + +```text +cargo fmt --all + --check : PASS +audit Rust général : clean +export completeness : 0 candidate +workspace Rust rules : clean +Markdown : clean, 340 tables / 838 files +cargo check --workspace : PASS +cargo clippy --workspace --all-targets --all-features -- -D warnings : PASS +worker unit : 84/84 PASS +cross_layer_completeness : 4/4 PASS +dependency_boundary : 12/12 PASS +hardening : 19/19 PASS +public_api : 13/13 PASS +release_completeness : 4/4 PASS +doc-tests Worker : 0/0 PASS +``` + +Cette preuve ferme la tranche Helius et autorise le polling HTTP live. + +## 51. Implémentation pre.006 — HTTP live block polling + +`pre.006` ajoute `RawTransactionIngestHttpBlockPollingSource` à l'agrégat borné. La construction caller-owned reçoit : + +```text +HttpTransportPool +HttpRoleName polling +SolanaCommitment Confirmed ou Finalized +poll_interval borné +max_discovered_blocks_per_cycle borné +``` + +Le profil doit exposer `getSlot`, `getBlocksWithLimit` et `getBlock` sur un réseau unique. Le Worker ne construit pas de client HTTP, ne lit pas Config et ne possède aucun retry HTTP : sélection d'endpoint, rate limit, retry et backoff restent dans Transport. + +Bornes retenues : + +```text +poll_interval : défaut 1 s, min 100 ms, max 30 s +max_discovered_blocks_per_cycle : défaut 128, min 1, max 1024 +``` + +## 52. Borne de run et algorithme de progression + +Au démarrage de la source : + +```text +start_slot = getSlot(commitment) +next_scan_slot = start_slot +``` + +Aucun appel de découverte ou matérialisation ne doit viser un slot inférieur à `start_slot`. Chaque cycle exécute ensuite : + +```text +current_tip = getSlot(commitment) +getBlocksWithLimit(next_scan_slot, bounded_limit, commitment) +pour chaque slot listé : getBlock observed(slot, Full/Base64/maxV1/rewards=false) +``` + +Les slots absents de `getBlocksWithLimit` sont traités comme skipped/non produits et ne déclenchent aucun faux gap. Si un slot explicitement listé retourne `getBlock = null`, ce slot reste `next_scan_slot`; la source le retente au cycle suivant et ne progresse jamais silencieusement au-delà. + +La cadence entre cycles appartient au Worker via un timer stop-preemptible. Aucun task par tick n'est créé. + +## 53. RAW-direct, version et provenance HTTP + +Comme Standard Block, le polling HTTP produit directement le Common RAW depuis chaque transaction `getBlock` : + +```text +encoding = Base64 +transactionDetails = Full +maxSupportedTransactionVersion = 1 +showRewards = false +Legacy / V0 / V1 uniquement +``` + +Version omise/nulle, version `>1`, transaction non Base64 ou champ transactions omis/nul sont fail-closed. La provenance utilise l'endpoint/provider réellement gagnant de `getBlock observed`, `protocol = solana_http`, `method = block_polling_get_block`, `origin = Live`. + +Le slot n'est déclaré settled qu'après admission réussie de toutes ses transactions. La voie directe ne détourne pas `hydration_pending`. + +## 54. Identité logique HTTP polling + +Le `source_key` dérive de : + +```text +domain KSP live source key +family = http_block_polling +network +polling role +commitment +private safe route-profile fingerprint +SHA-256 -> [u8; 32] +``` + +L'empreinte de profil inclut les identités sûres endpoint/provider et les capacités des routes pertinentes. La cadence et la limite de découverte sont volontairement exclues : changer un réglage de timing ne doit pas permettre de contourner la détection de doublon d'une même source logique. Les états de rate-limit/health dynamiques n'entrent pas non plus dans cette identité. + +## 55. Frontières et report pre.007 + +`pre.006` rend HTTP Block Polling productif en mode source unique. L'agrégat `1..32` accepte cette nouvelle famille et rejette les doublons/cross-network avant mutation, mais une collection `>1` continue d'échouer fermée avant spawn. + +La supervision simultanée, le lifecycle par source et l'inventaire multi-source restent exclusivement `pre.007`. `pre.006` ne transforme pas le poller HTTP en Backfill et ne crée aucun checkpoint durable inter-run. diff --git a/docs/validation/030-V0_3_13_MULTI_SOURCE_LIVE_CONVERGENCE.md b/docs/validation/030-V0_3_13_MULTI_SOURCE_LIVE_CONVERGENCE.md index 31be913..7c2ffee 100644 --- a/docs/validation/030-V0_3_13_MULTI_SOURCE_LIVE_CONVERGENCE.md +++ b/docs/validation/030-V0_3_13_MULTI_SOURCE_LIVE_CONVERGENCE.md @@ -1,11 +1,11 @@ - + # Validation v0.3.13 — WS standard / Helius / HTTP live + convergence multi-source ## 1. Rôle -Ce document enregistre les preuves réellement exécutées et les non-claims de `0.3.13`. `pre.001` ferme l'audit/planification, `pre.002` matérialise la composition runtime multi-source, `pre.003` ajoute Standard Solana `logsSubscribe` + hydration HTTP observée, `pre.004` ajoute Standard Solana `blockSubscribe` RAW-direct qualifié Legacy/V0/V1 et `pre.005` ajoute Helius `transactionSubscribe` + hydration commune, sans activer encore la supervision simultanée. +Ce document enregistre les preuves réellement exécutées et les non-claims de `0.3.13`. `pre.001` ferme l'audit/planification, `pre.002` matérialise la composition runtime multi-source, `pre.003` ajoute Standard Solana `logsSubscribe` + hydration HTTP observée, `pre.004` ajoute Standard Solana `blockSubscribe` RAW-direct qualifié Legacy/V0/V1, `pre.005` ajoute Helius `transactionSubscribe` + hydration commune et `pre.006` ajoute le live HTTP block polling run-local, sans activer encore la supervision simultanée. ## 2. Archive stable contrôlée @@ -1067,4 +1067,91 @@ cargo tree -p ksp-worker-raw-transaction-ingest-lib --edges normal cargo tree -p ksp-worker-raw-transaction-ingest-lib -e features cargo tree --duplicates ``` +## 51. Gate opérateur pre.005-fix.001 reçu + +Preuve communiquée le 10 septembre 2026 : + +```text +fmt : PASS +audits Rust : clean / export completeness 0 +Markdown : clean, 340 tables / 838 files +cargo check --workspace : PASS +clippy workspace all-targets all-features -D warnings : PASS +worker unit : 84/84 PASS +cross-layer : 4/4 PASS +dependency boundary : 12/12 PASS +hardening : 19/19 PASS +public API : 13/13 PASS +release completeness : 4/4 PASS +``` + +## 52. Surface pre.006 matérialisée + +```text +RawTransactionIngestHttpBlockPollingSource +RawTransactionIngestRuntimeResources::from_http_block_polling_source +RawTransactionIngestRuntimeResources::try_push_http_block_polling_source +getSlot(commitment) au démarrage et à chaque cycle +getBlocksWithLimit(next_scan_slot, bounded_limit, commitment) +getBlock observed sur les slots listés +Full/Base64/maxV1/rewards=false +RAW-direct Legacy/V0/V1 +timer Worker stop-preemptible +``` + +Aucun manifeste de crate ni aucune dépendance Transport/Config supplémentaire n'est requis. + +## 53. Invariants statiques vérifiés pre.006 + +La tranche encode explicitement : + +```text +start_slot = premier getSlot du run +next_scan_slot initial = start_slot +aucune découverte avant next_scan_slot +découverte strictement croissante +getBlock=null sur slot listé => next_scan_slot reste ce slot +skipped slot non listé => pas de faux gap +slot settled seulement après admission de tout le bloc +aucun hydration_pending pour la voie RAW-direct +version Legacy/V0/V1 qualifiée, ambiguë/>1 rejetée +provenance endpoint/provider réellement observé sur getBlock +source_key indépendant du poll interval et du batch size +``` + +Les anciens canaris de `0.3.12` qui interdisaient globalement `get_block_observed`/`tokio::time::sleep` ont été resserrés plutôt que supprimés : le coordinator d'hydration continue de ne posséder ni polling ni retry, tandis que ces primitives sont explicitement permises et bornées dans la seule source HTTP polling. + +## 54. Validation locale d'assemblage pre.006 + +Le toolchain Cargo/Rustfmt n'est pas disponible dans l'environnement d'assemblage. Les commandes Cargo post-modification restent donc `NON EXÉCUTÉ LOCAL`. Les audits statiques exécutés sur l'état final sont consignés dans le delta `pre.006`. + +## 55. Non-claims pre.006 + +```text +pas de scan historique avant le start_slot du run +pas de checkpoint durable de polling +pas de Backfill implicite +pas de multi-source simultané avant pre.007 +pas de retry/backoff HTTP implémenté dans le Worker +pas de client reqwest direct +pas de Config lu dans le Worker +pas de version transaction >1 ou ambiguë acceptée +pas de getBlock=null assimilé à un slot vide +``` + +## 56. Gate opérateur requis avant pre.007 + +```bash +cargo fmt --all +cargo fmt --all -- --check +python3 scripts/audit_rust_workspace_rules.py +python3 scripts/audit_markdown_tables.py README.md RULES.md ROADMAP.md CHANGELOG.md docs prompts crates deltas +cargo check --workspace +cargo clippy --workspace --all-targets --all-features -- -D warnings +cargo test -p ksp-onchain-transport-lib +cargo test -p ksp-worker-raw-transaction-ingest-lib +cargo tree -p ksp-worker-raw-transaction-ingest-lib --edges normal +cargo tree -p ksp-worker-raw-transaction-ingest-lib -e features +cargo tree --duplicates +```