diff --git a/Cargo.toml b/Cargo.toml index 81a2548..a2c2f59 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,12 +1,12 @@ # file: Cargo.toml -# version: 541 +# version: 542 [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.2.fix.1" +version = "0.3.13-pre.3" edition = "2024" license = "MIT" repository = "https://git.sasedev.com/Sasedev/khadhroony-solana-project" diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/README.md b/crates/ksp-worker-raw-transaction-ingest-lib/README.md index 8f71bc4..42c9970 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 @@ -12,7 +12,8 @@ RawTransactionIngestWorker::start -> fondation source-neutral, sans source productive RawTransactionIngestWorker::start_with_runtime_resources - -> même runtime + une source Yellowstone productive supervisée + -> même runtime + une source productive supervisée + Yellowstone ou WS standard logsSubscribe + hydration HTTP getTransaction ``` @@ -20,14 +21,20 @@ Le Worker reste indépendant de Config et de tout backend Store physique. Le cal ## Pipeline productif actuel -La première verticale live est : +Les verticales live productives sont : ```text Yellowstone standard subscribe -> Transaction / TransactionStatus / Block -> signal source-neutral (network, signature, slot, commitment, provenance sûre) + +Solana standard WS logsSubscribe + -> context.slot + signature + -> signal source-neutral (network, signature, slot, commitment, provenance sûre) + +les deux chemins -> coalescence bornée par (network, signature, commitment) - -> HTTP getTransaction observed + -> HTTP getTransaction observed, Base64, maxSupportedTransactionVersion=1 -> ksp-raw-transaction-lib -> admission centrale bornée -> ksp-store-lib @@ -58,7 +65,23 @@ 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 contient exactement une source Yellowstone validée. Il n'expose ni enum provider, ni collection de sources, ni callback, ni queue d'enqueue, ni client inférieur. +Le runtime-resource aggregate public accepte une collection validée de 1 à 32 sources logiques. `pre.003` sait exécuter une source unique Yellowstone ou une source unique Standard Logs ; 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 + +`RawTransactionIngestStandardLogsSource::new` reçoit : + +```text +WsEndpointSettings kind solana_standard +SolanaLogsSubscribeFilter +SolanaCommitment Confirmed ou Finalized +HttpTransportPool +HttpRoleName d'hydration +``` + +La construction est sans I/O. Elle refuse un endpoint WS invalide ou non standard, `Processed`, un réseau/provenance non représentable et l'absence de route HTTP `getTransaction` compatible sur le même réseau. Le filtre `All`, `AllWithVotes` ou `Mentions(pubkey)` participe uniquement à une empreinte privée ; le pubkey d'un filtre `Mentions` n'est pas recopié dans `Debug`, snapshot ou provenance textuelle. + +Au runtime, le Worker ouvre `SolanaStandardWsSession::connect`, puis `logs_subscribe`. Les lignes de logs et `err` restent dans Transport et ne sont jamais stockées dans le signal Worker. Seuls `context.slot` et `signature` sont projetés vers l'hydration commune. Reconnect, resubscribe et backpressure de la subscription restent possédés par Transport. ## Runtime et lifecycle @@ -77,11 +100,11 @@ Le shutdown est borné par `shutdown_drain_timeout`. Les tâches source, hydrati La queue centrale est un `tokio::sync::mpsc` privé borné par `admission_queue_capacity`. Les sources internes subissent la backpressure ; aucune queue non bornée ni silent drop n'est autorisé. -La source Yellowstone possède également un coordinateur d'hydration borné : +Le Worker possède un coordinateur d'hydration source-neutral borné, réutilisé par Yellowstone et Standard Logs : ```text in-flight hydration <= persistence_concurrency -pending source signals <= 65_536 +pending source signals <= admission_queue_capacity ``` Les signaux partageant le même `(network, signature, commitment)` sont coalescés avant le fan-out HTTP. Les provenances utiles restent néanmoins conservées pour les ingress produits après hydration. @@ -196,7 +219,7 @@ 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 WS/Helius-specific ou HTTP polling Worker ; +- source `blockSubscribe`, Helius `transactionSubscribe` ou HTTP 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 96206b1..785637c 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,35 @@ 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 la source Yellowstone/HTTP cible ce même réseau. +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 ou Standard Logs ; 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 + +Une source standard Solana WS se compose ainsi : + +```rust +fn standard_logs_runtime_resources( + ws_endpoint: ksp_onchain_transport_lib::WsEndpointSettings, + filter: ksp_onchain_transport_lib::SolanaLogsSubscribeFilter, + commitment: ksp_onchain_transport_lib::SolanaCommitment, + http_pool: ksp_onchain_transport_lib::HttpTransportPool, + hydration_role: ksp_onchain_transport_lib::HttpRoleName, +) -> ksp_core_lib::Result { + let source = match ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestStandardLogsSource::new( + ws_endpoint, + filter, + commitment, + http_pool, + hydration_role, + ) { + 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_standard_logs_source(source)); +} +``` + +`ws_endpoint` doit être un endpoint Transport valide de kind `solana_standard`. Le commitment doit être explicitement `Confirmed` ou `Finalized`. Le pool HTTP doit exposer `getTransaction` via le rôle indiqué sur le même réseau. `All`, `AllWithVotes` et `Mentions(pubkey)` sont acceptés par le contrat Transport ; la valeur du filtre reste privée dans le Worker. ## Préparer la source Yellowstone @@ -118,15 +146,16 @@ Le caller ne passe pas de signature, `program_id`, plage de slots ou limite hist ## Comprendre la pipeline live -Les familles Yellowstone sont traitées ainsi : +Les familles productives sont traitées ainsi : ```text -Transaction -> signal -> HTTP getTransaction -> Common RAW -> admission -TransactionStatus -> signal -> HTTP getTransaction -> Common RAW -> admission -Block -> un signal par transaction -> HTTP getTransaction -> Common RAW -> admission -BlockMeta -> continuity-only -Slot -> continuity-only -Account/Ping/Pong/Entry -> sans RAW Transaction dans cette verticale +Yellowstone Transaction -> signal -> HTTP getTransaction -> Common RAW -> admission +Yellowstone TransactionStatus -> signal -> HTTP getTransaction -> Common RAW -> admission +Yellowstone Block -> un signal par transaction -> HTTP getTransaction -> Common RAW -> admission +Standard WS logsSubscribe -> context.slot + signature -> HTTP getTransaction -> Common RAW -> admission +Yellowstone BlockMeta -> continuity-only +Yellowstone Slot -> continuity-only +Yellowstone Account/Ping/Pong/Entry -> sans RAW Transaction dans cette verticale ``` Les signaux de même `(network, signature, commitment)` sont coalescés avant l'hydration HTTP. Le Worker ne possède pas une boucle de retry HTTP : reroutage/retry/backoff restent dans `ksp-onchain-transport-lib`. @@ -266,7 +295,7 @@ Le pattern attendu est : ```text Config / application / service owner -> résout endpoints, credentials et rôles - -> construit YellowstoneGrpcChannel + YellowstoneSubscribeRequest + -> construit une source Transport Yellowstone ou Standard Logs -> construit HttpTransportPool + hydration role -> construit le Store -> construit RawTransactionIngestRuntimeResources 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 a526684..06849ac 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: 19 +// version: 20 #![warn(missing_docs)] #![deny(unreachable_pub)] @@ -10,12 +10,11 @@ //! 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.002` adds a bounded 1..32 caller-composed live-source -//! aggregate with deterministic logical source identities while deliberately keeping productive multi-source activation gated until the supervisor tranche; the -//! validated Yellowstone/HTTP runtime-resource contract still drives the only productive supervised source task. -//! Transaction/TransactionStatus/Block updates feed bounded HTTP `getTransaction` hydration and the existing central -//! admission path; BlockMeta/Slot remain continuity-only signals. Hydration is coalesced by network/signature/commitment -//! under bounded in-flight and pending budgets. A bounded run-local processing frontier projects hydration pending, +//! Store persistence in normal mode plus concrete latest-value snapshots projected onto Worker API. `pre.003` keeps the bounded 1..32 caller-composed +//! aggregate and adds a productive standard Solana `logsSubscribe` + observed HTTP `getTransaction` source beside Yellowstone while simultaneous multi-source +//! activation remains gated until the dedicated supervisor tranche. Both live source families converge into one source-neutral hydration coordinator contract and +//! the existing central admission path. Yellowstone Transaction/TransactionStatus/Block and standard logs notifications become signature/slot references; +//! BlockMeta/Slot remain continuity-only signals. Hydration is coalesced by network/signature/commitment under bounded in-flight and pending budgets. A bounded run-local processing frontier projects hydration pending, //! oldest pending slot and highest unblocked actually observed slot. The productive source now also projects safe Transport reconnect/replay state and faults //! conservatively when Transport proves a replay-retention continuity gap; history remediation remains outside this crate. @@ -52,8 +51,10 @@ pub use self::runtime::RawTransactionIngestTerminalFuture; pub use self::runtime::RawTransactionIngestWorker; /// Maximum number of logical live sources accepted by one runtime-resource aggregate. pub use self::runtime_resources::MAX_RAW_TRANSACTION_INGEST_LIVE_SOURCES; -/// Caller-composed runtime resources for the first Yellowstone + HTTP source family. +/// Caller-composed bounded runtime resources for supported continuous RAW transaction live-source families. pub use self::runtime_resources::RawTransactionIngestRuntimeResources; +/// Validated standard Solana `logsSubscribe` + HTTP hydration source contract owned by the continuous RAW transaction ingest Worker. +pub use self::runtime_resources::RawTransactionIngestStandardLogsSource; /// Validated Yellowstone + HTTP source contract owned by the continuous RAW transaction ingest Worker. pub use self::runtime_resources::RawTransactionIngestYellowstoneSource; /// Default bounded admission queue capacity for one RAW transaction ingest Worker. diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/src/runtime.rs b/crates/ksp-worker-raw-transaction-ingest-lib/src/runtime.rs index 4cdbfb2..e8e634e 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/src/runtime.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/src/runtime.rs @@ -1,5 +1,5 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/src/runtime.rs -// version: 12 +// version: 13 type PersistencePort = std::sync::Arc; type PersistenceTasks = tokio::task::JoinSet>; @@ -99,8 +99,8 @@ impl crate::RawTransactionIngestWorker { /// Starts one Worker with caller-composed runtime resources. /// - /// `pre.002` accepts a validated multi-source aggregate but intentionally activates only the existing single-Yellowstone supervisor path; aggregates with - /// more than one source fail closed before any source task is spawned. + /// `pre.003` activates either the existing Yellowstone source or one standard Solana `logsSubscribe` source through the same bounded Worker foundation. + /// Aggregates with more than one source still fail closed before any source task is spawned until the dedicated simultaneous supervisor tranche. pub fn start_with_runtime_resources( settings: crate::RawTransactionIngestSettings, store: std::sync::Arc, @@ -117,10 +117,9 @@ impl crate::RawTransactionIngestWorker { if let std::result::Result::Err(error) = runtime_resources.validate_network(settings.network()) { return std::result::Result::Err(error); } - let source = match runtime_resources.into_yellowstone_source() { - std::result::Result::Ok(value) => value, - std::result::Result::Err(error) => return std::result::Result::Err(error), - }; + if let std::result::Result::Err(error) = runtime_resources.validate_single_source_activation() { + return std::result::Result::Err(error); + } let source_settings = settings.clone(); let (processing_frontier_sender, processing_frontier_receiver) = tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty()); @@ -132,7 +131,7 @@ impl crate::RawTransactionIngestWorker { std::option::Option::Some(processing_frontier_receiver), move |children, stop_receiver, admission_sender| { let _abort_handle = children.spawn(async move { - return source.run(source_settings, stop_receiver, admission_sender, processing_frontier_sender).await; + return runtime_resources.run_single_live_source(source_settings, stop_receiver, admission_sender, processing_frontier_sender).await; }); }, ); 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 f8cac1d..ff2362d 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,5 +1,5 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs -// version: 13 +// version: 14 use sha2::Digest; // rust-rules: trait-import @@ -7,6 +7,9 @@ use sha2::Digest; // rust-rules: trait-import pub const MAX_RAW_TRANSACTION_INGEST_LIVE_SOURCES: usize = 32; const RAW_TRANSACTION_INGEST_LIVE_SOURCE_KEY_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.live_source.source_key.v1\0"; +const RAW_TRANSACTION_INGEST_STANDARD_LOGS_FILTER_FINGERPRINT_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.standard_logs.filter.v1\0"; +const RAW_TRANSACTION_INGEST_STANDARD_LOGS_HTTP_PROTOCOL: &str = "solana_ws_http"; +const RAW_TRANSACTION_INGEST_STANDARD_LOGS_HTTP_SOURCE_KEY_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.standard_logs_http.source_key.v1\0"; const RAW_TRANSACTION_INGEST_YELLOWSTONE_FILTER_FINGERPRINT_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.yellowstone.filters.v1\0"; const RAW_TRANSACTION_INGEST_YELLOWSTONE_HTTP_PROTOCOL: &str = "yellowstone_http"; const RAW_TRANSACTION_INGEST_YELLOWSTONE_HTTP_SOURCE_KEY_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.yellowstone_http.source_key.v1\0"; @@ -14,26 +17,43 @@ const RAW_TRANSACTION_INGEST_YELLOWSTONE_HTTP_SOURCE_KEY_DOMAIN: &[u8] = b"ksp.r #[derive(Clone, Copy, Debug, Eq, PartialEq)] enum RawTransactionIngestSourceFamily { Block, + Logs, Transaction, TransactionStatus, } enum RawTransactionIngestLiveSource { + StandardLogs(crate::RawTransactionIngestStandardLogsSource), Yellowstone(crate::RawTransactionIngestYellowstoneSource), } impl RawTransactionIngestLiveSource { fn network(&self) -> &ksp_store_lib::RawNetworkId { return match self { + Self::StandardLogs(source) => &source.network, Self::Yellowstone(source) => &source.network, }; } fn source_key(&self) -> [u8; 32] { return match self { + Self::StandardLogs(source) => source.source_key, Self::Yellowstone(source) => source.source_key, }; } + + async fn run( + self, + settings: crate::RawTransactionIngestSettings, + stop_receiver: tokio::sync::watch::Receiver, + admission_sender: tokio::sync::mpsc::Sender, + processing_frontier_sender: tokio::sync::watch::Sender, + ) -> ksp_core_lib::Result<()> { + return match self { + 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, + }; + } } #[derive(Clone, Copy, Debug, Eq, PartialEq)] @@ -321,6 +341,18 @@ impl std::convert::From<(&crate::RawTransactionIngestYellowstoneSource, &ksp_onc } } +#[derive(Clone)] +struct RawTransactionIngestHydrationContext { + commitment: ksp_onchain_transport_lib::SolanaCommitment, + http_pool: ksp_onchain_transport_lib::HttpTransportPool, + hydration_role: ksp_onchain_transport_lib::HttpRoleName, + network: ksp_store_lib::RawNetworkId, + protocol: &'static str, + route: RawTransactionIngestSourceRoute, + route_prefix: &'static str, + source_key_domain: &'static [u8], +} + /// Validated Yellowstone plus HTTP runtime source owned by the continuous RAW transaction ingest Worker. /// /// The Transport-owned channel, subscribe request, HTTP pool and hydration role remain private. Construction validates deterministic source-composition @@ -373,7 +405,7 @@ impl crate::RawTransactionIngestYellowstoneSource { std::option::Option::Some(value) => value, std::option::Option::None => return std::result::Result::Err(crate::runtime_error("runtime_resources.hydration_method_missing")), }; - let compatible_http_routes = compatible_http_route_count(&http_pool, &hydration_role, method.request_kind(), network.as_str(), &route); + let compatible_http_routes = compatible_http_route_count(&http_pool, &hydration_role, method.request_kind(), network.as_str(), &route, "ys"); let compatible_http_routes = match compatible_http_routes { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), @@ -389,6 +421,23 @@ impl crate::RawTransactionIngestYellowstoneSource { return std::result::Result::Ok(Self { yellowstone_channel, subscribe_request, http_pool, hydration_role, network, route, source_key }); } + fn hydration_context(&self) -> RawTransactionIngestHydrationContext { + let commitment = match self.subscribe_request.commitment() { + std::option::Option::Some(value) => value, + std::option::Option::None => ksp_onchain_transport_lib::SolanaCommitment::Confirmed, + }; + return RawTransactionIngestHydrationContext { + commitment, + http_pool: self.http_pool.clone(), + hydration_role: self.hydration_role.clone(), + network: self.network.clone(), + protocol: RAW_TRANSACTION_INGEST_YELLOWSTONE_HTTP_PROTOCOL, + route: self.route.clone(), + route_prefix: "ys", + source_key_domain: RAW_TRANSACTION_INGEST_YELLOWSTONE_HTTP_SOURCE_KEY_DOMAIN, + }; + } + /// Runs the productive Yellowstone source task until cooperative stop or one safe terminal source failure. pub(crate) async fn run( self, @@ -408,6 +457,7 @@ impl crate::RawTransactionIngestYellowstoneSource { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(source_transport_error(error.code())), }; + let hydration = self.hydration_context(); let mut coordinator = RawTransactionIngestHydrationCoordinator::new(&settings); let mut processing_frontier = RawTransactionIngestProcessingFrontierReporter::new(processing_frontier_sender); let snapshot_source = session.snapshot_source(); @@ -422,7 +472,7 @@ impl crate::RawTransactionIngestYellowstoneSource { if *stop_receiver.borrow() { break; } - if let std::result::Result::Err(error) = coordinator.start_hydrations(&self) { + if let std::result::Result::Err(error) = coordinator.start_hydrations(&hydration) { fault = std::option::Option::Some(error); break; } @@ -460,7 +510,7 @@ impl crate::RawTransactionIngestYellowstoneSource { let handled = coordinator .handle_joined( joined, - &self, + &hydration, &settings, &admission_sender, &mut stop_receiver, @@ -490,6 +540,7 @@ impl crate::RawTransactionIngestYellowstoneSource { }; if let std::result::Result::Err(error) = route_yellowstone_update( &self, + &hydration, &mut coordinator, &mut processing_frontier, update, @@ -520,6 +571,258 @@ impl crate::RawTransactionIngestYellowstoneSource { } } +/// Validated standard Solana `logsSubscribe` plus HTTP hydration source owned by the continuous RAW transaction ingest Worker. +/// +/// The caller provides one Transport-owned standard WebSocket endpoint, one standard logs filter, a Confirmed/Finalized commitment and one HTTP hydration +/// pool/role. The source keeps the WebSocket URL and filter private, never copies remote logs/errors into Worker state, and emits only signature/slot +/// references into the source-neutral hydration coordinator. +pub struct RawTransactionIngestStandardLogsSource { + ws_endpoint: ksp_onchain_transport_lib::WsEndpointSettings, + filter: ksp_onchain_transport_lib::SolanaLogsSubscribeFilter, + commitment: ksp_onchain_transport_lib::SolanaCommitment, + http_pool: ksp_onchain_transport_lib::HttpTransportPool, + hydration_role: ksp_onchain_transport_lib::HttpRoleName, + network: ksp_store_lib::RawNetworkId, + route: RawTransactionIngestSourceRoute, + filter_fingerprint: [u8; 32], + source_key: [u8; 32], +} + +impl crate::RawTransactionIngestStandardLogsSource { + /// Creates one validated standard Solana logs source without opening WebSocket or HTTP network I/O. + pub fn new( + ws_endpoint: ksp_onchain_transport_lib::WsEndpointSettings, + filter: ksp_onchain_transport_lib::SolanaLogsSubscribeFilter, + commitment: ksp_onchain_transport_lib::SolanaCommitment, + http_pool: ksp_onchain_transport_lib::HttpTransportPool, + hydration_role: ksp_onchain_transport_lib::HttpRoleName, + ) -> ksp_core_lib::Result { + let ws_settings = ksp_onchain_transport_lib::WsTransportSettings::new(std::vec![ws_endpoint.clone()]); + if ws_settings.validate().is_err() { + return std::result::Result::Err(crate::runtime_error("runtime_resources.standard_logs_endpoint_invalid")); + } + if ws_endpoint.protocol() != ksp_onchain_transport_lib::WsProtocolKind::SolanaStandard { + return std::result::Result::Err(crate::runtime_error("runtime_resources.standard_logs_protocol_invalid")); + } + 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.hydration_commitment_invalid")); + }, + } + let network = match ksp_store_lib::RawNetworkId::new(ws_endpoint.cluster().as_str()) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("runtime_resources.standard_logs_network_unrepresentable")), + }; + let provider = match ksp_store_lib::RawProvenanceCode::new(ws_endpoint.provider().as_str()) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("runtime_resources.standard_logs_provider_unrepresentable")), + }; + let endpoint_id = match ksp_store_lib::RawProvenanceCode::new(ws_endpoint.name()) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("runtime_resources.standard_logs_endpoint_unrepresentable")), + }; + let route = RawTransactionIngestSourceRoute { endpoint_id, provider }; + let method = match ksp_onchain_transport_lib::find_http_rpc_method("getTransaction") { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::result::Result::Err(crate::runtime_error("runtime_resources.hydration_method_missing")), + }; + let compatible_http_routes = compatible_http_route_count(&http_pool, &hydration_role, method.request_kind(), network.as_str(), &route, "ws"); + let compatible_http_routes = match compatible_http_routes { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + if compatible_http_routes == 0 { + return std::result::Result::Err(crate::runtime_error("runtime_resources.hydration_role_unsupported")); + } + let filter_fingerprint = standard_logs_filter_fingerprint(&filter); + let source_key = standard_logs_live_source_key(&network, &route, commitment, &filter_fingerprint); + return std::result::Result::Ok(Self { + ws_endpoint, + filter, + commitment, + http_pool, + hydration_role, + network, + route, + filter_fingerprint, + source_key, + }); + } + + fn hydration_context(&self) -> RawTransactionIngestHydrationContext { + return RawTransactionIngestHydrationContext { + commitment: self.commitment, + http_pool: self.http_pool.clone(), + hydration_role: self.hydration_role.clone(), + network: self.network.clone(), + protocol: RAW_TRANSACTION_INGEST_STANDARD_LOGS_HTTP_PROTOCOL, + route: self.route.clone(), + route_prefix: "ws", + source_key_domain: RAW_TRANSACTION_INGEST_STANDARD_LOGS_HTTP_SOURCE_KEY_DOMAIN, + }; + } + + /// Runs one productive standard `logsSubscribe` 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 connected = tokio::select! { + biased; + _ = stop_receiver.changed() => { + return std::result::Result::Ok(()); + } + result = ksp_onchain_transport_lib::SolanaStandardWsSession::connect(self.ws_endpoint.clone()) => result, + }; + let session = match connected { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(source_transport_error(error.code())), + }; + let config = ksp_onchain_transport_lib::SolanaCommitmentConfig::new(std::option::Option::Some(self.commitment)); + let subscribed = tokio::select! { + biased; + _ = stop_receiver.changed() => { + let _closed = session.close().await; + return std::result::Result::Ok(()); + } + result = session.logs_subscribe(&self.filter, std::option::Option::Some(&config)) => result, + }; + let mut subscription = match subscribed { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + let _closed = session.close().await; + return std::result::Result::Err(source_transport_error(error.code())); + }, + }; + let hydration = self.hydration_context(); + let mut coordinator = RawTransactionIngestHydrationCoordinator::new(&settings); + let mut processing_frontier = RawTransactionIngestProcessingFrontierReporter::new(processing_frontier_sender); + processing_frontier.set_source_state(crate::RawTransactionIngestSourceState::Active); + let mut fault = std::option::Option::None; + loop { + if *stop_receiver.borrow() { + break; + } + if let std::result::Result::Err(error) = coordinator.start_hydrations(&hydration) { + fault = std::option::Option::Some(error); + break; + } + let can_receive = coordinator.can_receive(); + if !can_receive && coordinator.tasks.is_empty() { + fault = std::option::Option::Some(crate::runtime_error("source.hydration_stalled")); + break; + } + tokio::select! { + biased; + _ = stop_receiver.changed() => { + break; + } + joined = coordinator.tasks.join_next(), if !coordinator.tasks.is_empty() => { + let joined = match joined { + std::option::Option::Some(value) => value, + std::option::Option::None => { + fault = std::option::Option::Some(crate::runtime_error("source.hydration_join_missing")); + break; + }, + }; + let handled = coordinator + .handle_joined( + joined, + &hydration, + &settings, + &admission_sender, + &mut stop_receiver, + &mut processing_frontier, + ) + .await; + match handled { + std::result::Result::Ok(true) => {}, + std::result::Result::Ok(false) => break, + std::result::Result::Err(error) => { + fault = std::option::Option::Some(error); + break; + }, + } + } + notification = subscription.recv(), if can_receive => { + let notification = match notification { + std::option::Option::Some(std::result::Result::Ok(value)) => value, + std::option::Option::Some(std::result::Result::Err(error)) => { + fault = std::option::Option::Some(source_transport_error(error.code())); + break; + }, + std::option::Option::None => { + let error = match subscription.terminal_error_code() { + std::option::Option::Some(code) => source_transport_error(code), + std::option::Option::None => crate::runtime_error("source.standard_logs_subscription_closed"), + }; + fault = std::option::Option::Some(error); + 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; + }, + }; + let signal = match project_standard_logs_signal(&self, ¬ification) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + fault = std::option::Option::Some(error); + break; + }, + }; + if let std::result::Result::Err(error) = coordinator.queue_signal(&hydration, signal, received_at, &mut processing_frontier) { + fault = std::option::Option::Some(error); + break; + } + } + } + } + coordinator.abort_all(&mut processing_frontier).await; + processing_frontier.set_source_state(crate::RawTransactionIngestSourceState::Closing); + let closed = session.close().await; + if let std::option::Option::Some(error) = fault { + processing_frontier.set_source_state(crate::RawTransactionIngestSourceState::Failed); + return std::result::Result::Err(error); + } + return match closed { + std::result::Result::Ok(()) => { + processing_frontier.set_source_state(crate::RawTransactionIngestSourceState::Closed); + std::result::Result::Ok(()) + }, + std::result::Result::Err(error) => { + processing_frontier.set_source_state(crate::RawTransactionIngestSourceState::Failed); + std::result::Result::Err(source_transport_error(error.code())) + }, + }; + } +} + +impl std::fmt::Debug for crate::RawTransactionIngestStandardLogsSource { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let http_snapshot = self.http_pool.snapshot(); + return formatter + .debug_struct("RawTransactionIngestStandardLogsSource") + .field("ws_endpoint_name", &self.ws_endpoint.name()) + .field("ws_provider", &self.ws_endpoint.provider().as_str()) + .field("network", &self.network.as_str()) + .field("commitment", &self.commitment) + .field("filter_kind", &standard_logs_filter_kind(&self.filter)) + .field("filter_fingerprint_bytes", &self.filter_fingerprint.len()) + .field("hydration_role", &self.hydration_role.as_str()) + .field("http_endpoint_count", &http_snapshot.endpoint_count()) + .field("source_key_bytes", &self.source_key.len()) + .finish(); + } +} + impl std::fmt::Debug for crate::RawTransactionIngestYellowstoneSource { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { let http_snapshot = self.http_pool.snapshot(); @@ -545,8 +848,8 @@ 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.002` validates composition and stable logical identities but -/// deliberately keeps productive multi-source supervision gated until the dedicated supervisor tranche. +/// The aggregate owns a bounded collection of capability-specific live-source contracts. `pre.003` supports Yellowstone and standard Solana logs sources while +/// deliberately keeping simultaneous multi-source supervision gated until the dedicated supervisor tranche. pub struct RawTransactionIngestRuntimeResources { sources: std::vec::Vec, } @@ -558,6 +861,12 @@ impl crate::RawTransactionIngestRuntimeResources { return Self { sources: std::vec![RawTransactionIngestLiveSource::Yellowstone(yellowstone_source)] }; } + /// Creates one runtime-resource aggregate from one standard Solana `logsSubscribe` plus HTTP hydration source contract. + #[must_use] + pub fn from_standard_logs_source(source: crate::RawTransactionIngestStandardLogsSource) -> Self { + return Self { sources: std::vec![RawTransactionIngestLiveSource::StandardLogs(source)] }; + } + /// Returns the number of validated logical live sources currently owned by this aggregate. #[must_use] pub fn source_count(&self) -> usize { @@ -585,6 +894,27 @@ impl crate::RawTransactionIngestRuntimeResources { return std::result::Result::Ok(()); } + /// Adds one validated standard Solana logs source while preserving the global 1..32 bound, one-network invariant and unique logical source identity. + pub fn try_push_standard_logs_source(&mut self, source: crate::RawTransactionIngestStandardLogsSource) -> 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::StandardLogs(source); + let expected_network = match self.sources.first() { + std::option::Option::Some(source) => source.network(), + std::option::Option::None => return std::result::Result::Err(crate::runtime_error("runtime_resources.source_collection_empty")), + }; + if candidate.network() != expected_network { + return std::result::Result::Err(crate::runtime_error("runtime_resources.source_network_mismatch")); + } + let source_key = candidate.source_key(); + if self.sources.iter().any(|source| return source.source_key() == source_key) { + return std::result::Result::Err(crate::runtime_error("runtime_resources.duplicate_source_identity")); + } + self.sources.push(candidate); + return std::result::Result::Ok(()); + } + /// Validates that caller-owned Worker settings target the same logical network as every composed live source. pub(crate) fn validate_network(&self, network: &ksp_store_lib::RawNetworkId) -> ksp_core_lib::Result<()> { if self.sources.is_empty() || self.sources.len() > crate::MAX_RAW_TRANSACTION_INGEST_LIVE_SOURCES { @@ -602,22 +932,34 @@ impl crate::RawTransactionIngestRuntimeResources { return std::result::Result::Ok(()); } - /// Consumes the aggregate into the single Yellowstone source supported by the current productive supervisor wiring. - /// - /// Multi-source activation is intentionally deferred to the dedicated supervisor/source-inventory tranche; a collection larger than one therefore fails - /// closed rather than silently dropping sources or starting only a subset. - pub(crate) fn into_yellowstone_source(self) -> ksp_core_lib::Result { + /// Validates that the current supervisor tranche can activate the aggregate without silently dropping configured sources. + pub(crate) fn validate_single_source_activation(&self) -> ksp_core_lib::Result<()> { if self.sources.is_empty() { return std::result::Result::Err(crate::runtime_error("runtime_resources.source_collection_empty")); } if self.sources.len() != 1 { return std::result::Result::Err(crate::runtime_error("runtime_resources.multi_source_activation_pending")); } + return std::result::Result::Ok(()); + } + + /// Runs the already validated single live source through the common Worker source-task contract. + pub(crate) async fn run_single_live_source( + self, + settings: crate::RawTransactionIngestSettings, + stop_receiver: tokio::sync::watch::Receiver, + admission_sender: tokio::sync::mpsc::Sender, + processing_frontier_sender: tokio::sync::watch::Sender, + ) -> ksp_core_lib::Result<()> { let mut sources = self.sources.into_iter(); - return match sources.next() { - std::option::Option::Some(RawTransactionIngestLiveSource::Yellowstone(source)) => std::result::Result::Ok(source), - std::option::Option::None => std::result::Result::Err(crate::runtime_error("runtime_resources.source_collection_empty")), + let source = match sources.next() { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::result::Result::Err(crate::runtime_error("runtime_resources.source_collection_empty")), }; + if sources.next().is_some() { + return std::result::Result::Err(crate::runtime_error("runtime_resources.multi_source_activation_pending")); + } + return source.run(settings, stop_receiver, admission_sender, processing_frontier_sender).await; } } @@ -642,6 +984,45 @@ impl std::hash::Hasher for RawTransactionIngestSourceKeyHashWriter<'_> { } } +fn standard_logs_filter_fingerprint(filter: &ksp_onchain_transport_lib::SolanaLogsSubscribeFilter) -> [u8; 32] { + let mut hasher = sha2::Sha256::new(); + hasher.update(RAW_TRANSACTION_INGEST_STANDARD_LOGS_FILTER_FINGERPRINT_DOMAIN); + match filter { + ksp_onchain_transport_lib::SolanaLogsSubscribeFilter::All => hash_live_source_key_component(&mut hasher, b"all"), + ksp_onchain_transport_lib::SolanaLogsSubscribeFilter::AllWithVotes => hash_live_source_key_component(&mut hasher, b"allWithVotes"), + ksp_onchain_transport_lib::SolanaLogsSubscribeFilter::Mentions(pubkey) => { + hash_live_source_key_component(&mut hasher, b"mentions"); + hash_live_source_key_component(&mut hasher, &pubkey.to_bytes()); + }, + } + return hasher.finalize().into(); +} + +fn standard_logs_filter_kind(filter: &ksp_onchain_transport_lib::SolanaLogsSubscribeFilter) -> &'static str { + return match filter { + ksp_onchain_transport_lib::SolanaLogsSubscribeFilter::All => "all", + ksp_onchain_transport_lib::SolanaLogsSubscribeFilter::AllWithVotes => "all_with_votes", + ksp_onchain_transport_lib::SolanaLogsSubscribeFilter::Mentions(_) => "mentions", + }; +} + +fn standard_logs_live_source_key( + network: &ksp_store_lib::RawNetworkId, + route: &RawTransactionIngestSourceRoute, + commitment: ksp_onchain_transport_lib::SolanaCommitment, + filter_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"standard_logs"); + hash_live_source_key_component(&mut hasher, network.as_str().as_bytes()); + hash_live_source_key_component(&mut hasher, route.provider.as_str().as_bytes()); + hash_live_source_key_component(&mut hasher, route.endpoint_id.as_str().as_bytes()); + hash_live_source_key_component(&mut hasher, commitment.as_str().as_bytes()); + hash_live_source_key_component(&mut hasher, filter_fingerprint); + return hasher.finalize().into(); +} + fn yellowstone_live_source_key( network: &ksp_store_lib::RawNetworkId, route: &RawTransactionIngestSourceRoute, @@ -675,7 +1056,8 @@ fn compatible_http_route_count( hydration_role: &ksp_onchain_transport_lib::HttpRoleName, request_kind: &str, expected_cluster: &str, - yellowstone_route: &RawTransactionIngestSourceRoute, + source_route: &RawTransactionIngestSourceRoute, + route_prefix: &str, ) -> ksp_core_lib::Result { let snapshot = pool.snapshot(); let mut compatible = 0_usize; @@ -694,7 +1076,7 @@ fn compatible_http_route_count( if endpoint.cluster() != expected_cluster { return std::result::Result::Err(crate::runtime_error("runtime_resources.transport_network_mismatch")); } - let composite = composite_provenance_codes(yellowstone_route, endpoint.provider(), endpoint.name()); + let composite = composite_provenance_codes(source_route, route_prefix, endpoint.provider(), endpoint.name()); if let std::result::Result::Err(error) = composite { return std::result::Result::Err(error); } @@ -705,18 +1087,19 @@ fn compatible_http_route_count( } fn composite_provenance_codes( - yellowstone_route: &RawTransactionIngestSourceRoute, + source_route: &RawTransactionIngestSourceRoute, + route_prefix: &str, http_provider: &str, http_endpoint: &str, ) -> ksp_core_lib::Result<(ksp_store_lib::RawProvenanceCode, ksp_store_lib::RawProvenanceCode)> { - let provider = std::format!("ys.{}:http.{}", yellowstone_route.provider.as_str(), http_provider); + let provider = std::format!("{}.{}:http.{}", route_prefix, source_route.provider.as_str(), http_provider); let provider = match ksp_store_lib::RawProvenanceCode::new(provider) { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => { return std::result::Result::Err(crate::runtime_error("runtime_resources.composite_provider_unrepresentable")); }, }; - let endpoint = std::format!("ys.{}:http.{}", yellowstone_route.endpoint_id.as_str(), http_endpoint); + let endpoint = std::format!("{}.{}:http.{}", route_prefix, source_route.endpoint_id.as_str(), http_endpoint); let endpoint = match ksp_store_lib::RawProvenanceCode::new(endpoint) { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => { @@ -849,8 +1232,48 @@ fn project_yellowstone_signal( }; } +trait RawTransactionIngestStandardLogsView { + fn signature(&self) -> &str; + + fn slot(&self) -> u64; +} + +impl RawTransactionIngestStandardLogsView for ksp_onchain_transport_lib::SolanaRpcResponse { + fn signature(&self) -> &str { + return self.value().signature(); + } + + fn slot(&self) -> u64 { + return self.context().slot(); + } +} + +fn project_standard_logs_signal( + source: &crate::RawTransactionIngestStandardLogsSource, + response: &T, +) -> ksp_core_lib::Result { + let signature = ksp_raw_transaction_lib::parse_raw_transaction_signature(response.signature()); + let signature = match signature { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("source.standard_logs_signature_invalid")), + }; + return std::result::Result::Ok(RawTransactionIngestSourceSignal { + created_at: std::option::Option::None, + family: RawTransactionIngestSourceFamily::Logs, + matched_filter_count: 1, + matched_filter_fingerprint: source.filter_fingerprint, + matched_filter_id: std::option::Option::None, + network: source.network.clone(), + route: source.route.clone(), + signature, + slot: response.slot(), + transaction_index: std::option::Option::None, + }); +} + fn route_yellowstone_update( source: &crate::RawTransactionIngestYellowstoneSource, + hydration: &RawTransactionIngestHydrationContext, coordinator: &mut RawTransactionIngestHydrationCoordinator, processing_frontier: &mut RawTransactionIngestProcessingFrontierReporter, update: ksp_onchain_transport_lib::YellowstoneSubscribeUpdate, @@ -861,14 +1284,14 @@ fn route_yellowstone_update( std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; - return coordinator.queue_signal(source, RawTransactionIngestSourceSignal::from((source, value.as_ref())), received_at, processing_frontier); + return coordinator.queue_signal(hydration, RawTransactionIngestSourceSignal::from((source, value.as_ref())), received_at, processing_frontier); }, ksp_onchain_transport_lib::YellowstoneSubscribeUpdate::TransactionStatus(value) => { let received_at = match current_raw_timestamp() { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; - return coordinator.queue_signal(source, RawTransactionIngestSourceSignal::from((source, &value)), received_at, processing_frontier); + return coordinator.queue_signal(hydration, RawTransactionIngestSourceSignal::from((source, &value)), received_at, processing_frontier); }, ksp_onchain_transport_lib::YellowstoneSubscribeUpdate::Block(value) => { let received_at = match current_raw_timestamp() { @@ -880,7 +1303,7 @@ fn route_yellowstone_update( std::result::Result::Err(error) => return std::result::Result::Err(error), }; for signal in signals { - if let std::result::Result::Err(error) = coordinator.queue_signal(source, signal, received_at, processing_frontier) { + if let std::result::Result::Err(error) = coordinator.queue_signal(hydration, signal, received_at, processing_frontier) { return std::result::Result::Err(error); } } @@ -935,7 +1358,7 @@ async fn wait_yellowstone_session_snapshot( } fn source_transport_error(code: ksp_core_lib::ErrorCode) -> ksp_core_lib::Error { - return ksp_core_lib::Error::new(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_SOURCE_FAILED, "RAW transaction ingest Yellowstone source transport failed") + return ksp_core_lib::Error::new(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_SOURCE_FAILED, "RAW transaction ingest source transport failed") .with_context("transport_domain", code.domain()) .with_context("transport_code", code.code()); } @@ -1251,7 +1674,7 @@ impl RawTransactionIngestHydrationCoordinator { fn new(settings: &crate::RawTransactionIngestSettings) -> Self { return Self { max_in_flight: settings.persistence_concurrency(), - max_pending_signals: crate::MAX_RAW_TRANSACTION_INGEST_ADMISSION_QUEUE_CAPACITY, + max_pending_signals: settings.admission_queue_capacity(), pending_signal_count: 0, pending: std::collections::BTreeMap::new(), tasks: RawTransactionIngestHydrationTasks::new(), @@ -1264,7 +1687,7 @@ impl RawTransactionIngestHydrationCoordinator { fn queue_signal( &mut self, - source: &crate::RawTransactionIngestYellowstoneSource, + hydration: &RawTransactionIngestHydrationContext, signal: RawTransactionIngestSourceSignal, received_at: ksp_store_lib::RawTimestamp, processing_frontier: &mut RawTransactionIngestProcessingFrontierReporter, @@ -1272,7 +1695,7 @@ impl RawTransactionIngestHydrationCoordinator { if self.pending_signal_count >= self.max_pending_signals { return std::result::Result::Err(crate::runtime_error("source.hydration_pending_saturated")); } - let key = match hydration_key(source, &signal) { + let key = match hydration_key(hydration, &signal) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; @@ -1295,10 +1718,10 @@ impl RawTransactionIngestHydrationCoordinator { if let std::result::Result::Err(error) = processing_frontier.observe_pending(signal_slot) { return std::result::Result::Err(error); } - return self.start_hydrations(source); + return self.start_hydrations(hydration); } - fn start_hydrations(&mut self, source: &crate::RawTransactionIngestYellowstoneSource) -> ksp_core_lib::Result<()> { + fn start_hydrations(&mut self, hydration: &RawTransactionIngestHydrationContext) -> ksp_core_lib::Result<()> { while self.tasks.len() < self.max_in_flight { let key = self.pending.iter().find_map(|(key, pending)| { if pending.in_flight { @@ -1310,21 +1733,18 @@ impl RawTransactionIngestHydrationCoordinator { std::option::Option::Some(value) => value, std::option::Option::None => break, }; - let commitment = match hydration_commitment(source) { - std::result::Result::Ok(value) => value, - std::result::Result::Err(error) => return std::result::Result::Err(error), - }; + let commitment = hydration.commitment; let pending = match self.pending.get_mut(&key) { std::option::Option::Some(value) => value, std::option::Option::None => return std::result::Result::Err(crate::runtime_error("source.hydration_pending_missing")), }; pending.in_flight = true; - let pool = source.http_pool.clone(); - let role = source.hydration_role.clone(); - let expected_network = source.network.clone(); + let pool = hydration.http_pool.clone(); + let role = hydration.hydration_role.clone(); + let expected_network = hydration.network.clone(); let task_key = key.clone(); let _abort_handle = self.tasks.spawn(async move { - return fetch_yellowstone_hydration(pool, role, expected_network, task_key, commitment).await; + return fetch_hydration(pool, role, expected_network, task_key, commitment).await; }); } return std::result::Result::Ok(()); @@ -1333,7 +1753,7 @@ impl RawTransactionIngestHydrationCoordinator { async fn handle_joined( &mut self, joined: std::result::Result, tokio::task::JoinError>, - source: &crate::RawTransactionIngestYellowstoneSource, + hydration: &RawTransactionIngestHydrationContext, settings: &crate::RawTransactionIngestSettings, admission_sender: &tokio::sync::mpsc::Sender, stop_receiver: &mut tokio::sync::watch::Receiver, @@ -1354,7 +1774,7 @@ impl RawTransactionIngestHydrationCoordinator { self.pending_signal_count -= pending.signals.len(); for pending_signal in pending.signals { let signal_slot = pending_signal.signal.slot; - let ingress = finalize_yellowstone_hydration(source, settings, pending_signal.signal, pending_signal.received_at, &fetched.observed); + let ingress = finalize_hydration(hydration, settings, pending_signal.signal, pending_signal.received_at, &fetched.observed); let ingress = match ingress { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), @@ -1385,7 +1805,7 @@ impl RawTransactionIngestHydrationCoordinator { return std::result::Result::Err(error); } } - if let std::result::Result::Err(error) = self.start_hydrations(source) { + if let std::result::Result::Err(error) = self.start_hydrations(hydration) { return std::result::Result::Err(error); } return std::result::Result::Ok(true); @@ -1401,39 +1821,21 @@ impl RawTransactionIngestHydrationCoordinator { } } -fn hydration_commitment(source: &crate::RawTransactionIngestYellowstoneSource) -> ksp_core_lib::Result { - return match source.subscribe_request.commitment() { - std::option::Option::Some(ksp_onchain_transport_lib::SolanaCommitment::Confirmed) => { - std::result::Result::Ok(ksp_onchain_transport_lib::SolanaCommitment::Confirmed) - }, - std::option::Option::Some(ksp_onchain_transport_lib::SolanaCommitment::Finalized) => { - std::result::Result::Ok(ksp_onchain_transport_lib::SolanaCommitment::Finalized) - }, - std::option::Option::Some(ksp_onchain_transport_lib::SolanaCommitment::Processed) | std::option::Option::None => { - std::result::Result::Err(crate::runtime_error("hydration.commitment_invalid")) - }, - }; -} - fn hydration_key( - source: &crate::RawTransactionIngestYellowstoneSource, + hydration: &RawTransactionIngestHydrationContext, signal: &RawTransactionIngestSourceSignal, ) -> ksp_core_lib::Result { - if signal.network != source.network { + if signal.network != hydration.network { return std::result::Result::Err(crate::runtime_error("hydration.network_mismatch")); } - let commitment = match hydration_commitment(source) { - std::result::Result::Ok(value) => value, - std::result::Result::Err(error) => return std::result::Result::Err(error), - }; return std::result::Result::Ok(RawTransactionIngestHydrationKey { - commitment: commitment.as_str(), + commitment: hydration.commitment.as_str(), network: signal.network.clone(), signature: signal.signature, }); } -async fn fetch_yellowstone_hydration( +async fn fetch_hydration( http_pool: ksp_onchain_transport_lib::HttpTransportPool, hydration_role: ksp_onchain_transport_lib::HttpRoleName, expected_network: ksp_store_lib::RawNetworkId, @@ -1449,7 +1851,7 @@ async fn fetch_yellowstone_hydration( let config = ksp_onchain_transport_lib::SolanaGetTransactionConfig::new( std::option::Option::Some(commitment), std::option::Option::Some(ksp_onchain_transport_lib::SolanaTransactionEncoding::Base64), - std::option::Option::Some(0), + std::option::Option::Some(1), ); let signature_text = ksp_raw_transaction_lib::format_raw_transaction_signature(&key.signature); let observed = http_pool.get_transaction_observed(&hydration_role, signature_text.as_str(), std::option::Option::Some(&config)).await; @@ -1460,23 +1862,20 @@ async fn fetch_yellowstone_hydration( return std::result::Result::Ok(RawTransactionIngestHydrationFetch { key, observed }); } -fn finalize_yellowstone_hydration( - source: &crate::RawTransactionIngestYellowstoneSource, +fn finalize_hydration( + hydration: &RawTransactionIngestHydrationContext, settings: &crate::RawTransactionIngestSettings, signal: RawTransactionIngestSourceSignal, received_at: ksp_store_lib::RawTimestamp, observed: &RawTransactionIngestObservedTransaction, ) -> ksp_core_lib::Result> { - if &signal.network != settings.network() || signal.network != source.network { + if &signal.network != settings.network() || signal.network != hydration.network { return std::result::Result::Err(crate::runtime_error("hydration.network_mismatch")); } - if signal.route != source.route { + if signal.route != hydration.route { return std::result::Result::Err(crate::runtime_error("hydration.source_route_mismatch")); } - let commitment = match hydration_commitment(source) { - std::result::Result::Ok(value) => value, - std::result::Result::Err(error) => return std::result::Result::Err(error), - }; + let commitment = hydration.commitment; let transaction = match observed.value().as_ref() { std::option::Option::Some(value) => value, std::option::Option::None => return std::result::Result::Ok(std::option::Option::None), @@ -1510,12 +1909,12 @@ fn finalize_yellowstone_hydration( if embedded_signature != signal.signature { return std::result::Result::Err(crate::runtime_error("hydration.signature_mismatch")); } - let provenance = build_hydration_provenance(settings, &signal, observed.provider().as_str(), observed.endpoint_name(), commitment, received_at); + let provenance = build_hydration_provenance(hydration, settings, &signal, observed.provider().as_str(), observed.endpoint_name(), commitment, received_at); let provenance = match provenance { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; - let source_key = hydration_source_key(&provenance); + let source_key = hydration_source_key(hydration.source_key_domain, &provenance); let material = ksp_raw_transaction_lib::RawTransactionMaterial::binary_base64( signal.network.clone(), signal.signature, @@ -1548,20 +1947,18 @@ async fn hydrate_yellowstone_signal( received_at: ksp_store_lib::RawTimestamp, ) -> ksp_core_lib::Result { let reference = ksp_store_lib::RawTransactionReference::new(signal.network.clone(), signal.signature); - let key = match hydration_key(source, &signal) { + let hydration = source.hydration_context(); + let key = match hydration_key(&hydration, &signal) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; - let commitment = match hydration_commitment(source) { - std::result::Result::Ok(value) => value, - std::result::Result::Err(error) => return std::result::Result::Err(error), - }; - let fetched = fetch_yellowstone_hydration(source.http_pool.clone(), source.hydration_role.clone(), source.network.clone(), key, commitment).await; + let commitment = hydration.commitment; + let fetched = fetch_hydration(hydration.http_pool.clone(), hydration.hydration_role.clone(), hydration.network.clone(), key, commitment).await; let fetched = match fetched { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; - let ingress = finalize_yellowstone_hydration(source, settings, signal, received_at, &fetched.observed); + let ingress = finalize_hydration(&hydration, settings, signal, received_at, &fetched.observed); return match ingress { std::result::Result::Ok(std::option::Option::Some(value)) => { std::result::Result::Ok(RawTransactionIngestHydrationOutcome::Available(std::boxed::Box::new(value))) @@ -1572,6 +1969,7 @@ async fn hydrate_yellowstone_signal( } fn build_hydration_provenance( + hydration: &RawTransactionIngestHydrationContext, settings: &crate::RawTransactionIngestSettings, signal: &RawTransactionIngestSourceSignal, http_provider: &str, @@ -1579,11 +1977,11 @@ fn build_hydration_provenance( commitment: ksp_onchain_transport_lib::SolanaCommitment, received_at: ksp_store_lib::RawTimestamp, ) -> ksp_core_lib::Result { - let (provider, endpoint_id) = match composite_provenance_codes(&signal.route, http_provider, http_endpoint) { + let (provider, endpoint_id) = match composite_provenance_codes(&signal.route, hydration.route_prefix, http_provider, http_endpoint) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; - let protocol = match ksp_store_lib::RawProvenanceCode::new(RAW_TRANSACTION_INGEST_YELLOWSTONE_HTTP_PROTOCOL) { + let protocol = match ksp_store_lib::RawProvenanceCode::new(hydration.protocol) { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("hydration.protocol_unrepresentable")), }; @@ -1624,6 +2022,7 @@ fn build_hydration_provenance( fn hydration_method_code(family: RawTransactionIngestSourceFamily) -> &'static str { return match family { RawTransactionIngestSourceFamily::Block => "block_get_transaction", + RawTransactionIngestSourceFamily::Logs => "logs_get_transaction", RawTransactionIngestSourceFamily::Transaction => "transaction_get_transaction", RawTransactionIngestSourceFamily::TransactionStatus => "status_get_transaction", }; @@ -1669,9 +2068,9 @@ fn representable_observed_at( return std::option::Option::Some(observed_at); } -fn hydration_source_key(provenance: &ksp_store_lib::RawAcquisitionProvenance) -> [u8; 32] { +fn hydration_source_key(source_key_domain: &[u8], provenance: &ksp_store_lib::RawAcquisitionProvenance) -> [u8; 32] { let mut hasher = sha2::Sha256::new(); - hasher.update(RAW_TRANSACTION_INGEST_YELLOWSTONE_HTTP_SOURCE_KEY_DOMAIN); + hasher.update(source_key_domain); hash_hydration_source_key(&mut hasher, provenance.provider().as_str().as_bytes()); hash_hydration_source_key(&mut hasher, provenance.protocol().as_str().as_bytes()); hash_hydration_source_key(&mut hasher, provenance.acquisition_method().as_str().as_bytes()); 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 2082d8f..e59c0ca 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: 16 +// version: 17 //! Dependency firewall canaries for the RAW transaction ingest Worker foundation. @@ -176,9 +176,9 @@ fn v0_3_12_pre_006_productive_source_uses_transport_session_bounded_coalescence_ let runtime = include_str!("../src/runtime.rs"); let resources = include_str!("../src/runtime_resources.rs"); for required in [ - "runtime_resources.into_yellowstone_source()", + "runtime_resources.validate_single_source_activation()", "children.spawn", - "source.run(source_settings, stop_receiver, admission_sender, processing_frontier_sender)", + ".run_single_live_source(source_settings, stop_receiver, admission_sender, processing_frontier_sender)", ] { assert!(runtime.contains(required), "required pre.006 supervisor wiring missing: {required}"); } @@ -214,6 +214,47 @@ fn v0_3_12_pre_006_productive_source_uses_transport_session_bounded_coalescence_ 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"); + let runtime = include_str!("../src/runtime.rs"); + let resources = include_str!("../src/runtime_resources.rs"); + for required in [ + "RawTransactionIngestStandardLogsSource", + "SolanaStandardWsSession::connect", + ".logs_subscribe(", + "SolanaLogsSubscribeFilter", + "SolanaCommitmentConfig::new", + "project_standard_logs_signal", + "RawTransactionIngestSourceFamily::Logs", + "logs_get_transaction", + "solana_ws_http", + "get_transaction_observed", + "max_pending_signals: settings.admission_queue_capacity()", + "std::option::Option::Some(1)", + ] { + assert!(resources.contains(required), "required pre.003 standard logs contract missing: {required}"); + } + assert!(runtime.contains("run_single_live_source")); + for forbidden in [ + "get_block_observed", + "ksp_config_lib::", + "ksp_job_backfill_lib::", + "ksp_store_postgres_lib::", + "reqwest::", + "tokio_tungstenite::", + "tonic::", + "yellowstone_grpc_proto::", + "unbounded_channel", + ] { + assert!( + !runtime.contains(forbidden) && !resources.contains(forbidden) && !root.contains(forbidden), + "pre.003 crossed a forbidden boundary: {forbidden}", + ); + } + return; +} + #[test] fn v0_3_12_pre_006_source_surface_hardens_shutdown_and_faults_without_backend_edges() { let root = include_str!("../src/lib.rs"); 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 9c7ba56..44efa74 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: 11 +// version: 12 //! External public, security, redaction and release-boundary hardening canaries for `pre.010`. @@ -159,7 +159,10 @@ fn pre_010_source_visibility_contract_uses_crate_root_for_shared_items() { (include_str!("../src/runtime.rs"), &["RawTransactionIngestHandle", "RawTransactionIngestWorker"]), (include_str!("../src/snapshot.rs"), &["RawTransactionIngestSnapshot", "RawTransactionIngestSnapshotSource"]), (include_str!("../src/persistence.rs"), &["RawTransactionIngestPersistenceOutcome", "RawTransactionIngestPersistencePort"]), - (include_str!("../src/runtime_resources.rs"), &["RawTransactionIngestYellowstoneSource", "RawTransactionIngestRuntimeResources"]), + ( + include_str!("../src/runtime_resources.rs"), + &["RawTransactionIngestStandardLogsSource", "RawTransactionIngestYellowstoneSource", "RawTransactionIngestRuntimeResources"], + ), ]; for (source, symbols) in source_contracts { for symbol in symbols { @@ -317,7 +320,8 @@ fn v0_3_12_pre_003_private_signal_debug_and_shape_do_not_expose_signature_filter fn v0_3_12_pre_004_hydration_provenance_and_remote_material_are_bounded_and_redacted() { let resources = include_str!("../src/runtime_resources.rs"); for required in [ - "ys.{}:http.{}", + "route_prefix", + "{}.{}:http.{}", "composite_provider_unrepresentable", "composite_endpoint_unrepresentable", "RawAcquisitionOrigin::Live", @@ -390,7 +394,7 @@ fn v0_3_12_pre_006_runtime_resource_contract_opens_one_supervised_transport_sour assert!(resources.contains(required), "productive runtime-resource source behavior missing: {required}"); } assert!(runtime.contains("start_with_runtime_resources")); - assert!(runtime.contains("source.run(source_settings, stop_receiver, admission_sender, processing_frontier_sender)")); + 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::"] { assert!(!resources.contains(forbidden) && !runtime.contains(forbidden), "pre.006 runtime source crossed a forbidden boundary: {forbidden}"); } @@ -402,7 +406,7 @@ fn v0_3_12_pre_006_source_coalescence_is_bounded_stop_preemptible_and_redacted() let resources = include_str!("../src/runtime_resources.rs"); for required in [ "max_in_flight: settings.persistence_concurrency()", - "max_pending_signals: crate::MAX_RAW_TRANSACTION_INGEST_ADMISSION_QUEUE_CAPACITY", + "max_pending_signals: settings.admission_queue_capacity()", "pending_signal_count", "stop_receiver.changed()", "tasks.abort_all()", @@ -523,6 +527,45 @@ fn v0_3_12_pre_008_reconnect_projection_is_source_neutral_bounded_and_contains_n return; } +#[test] +fn v0_3_13_pre_003_standard_logs_redaction_and_reference_only_contract_are_explicit() { + let resources = include_str!("../src/runtime_resources.rs"); + for required in [ + "RawTransactionIngestStandardLogsSource", + ".field(\"filter_kind\"", + ".field(\"filter_fingerprint_bytes\"", + ".field(\"source_key_bytes\"", + "project_standard_logs_signal", + "matched_filter_count: 1", + "transaction_index: std::option::Option::None", + "created_at: std::option::Option::None", + ] { + assert!(resources.contains(required), "required pre.003 redaction/reference guard missing: {required}"); + } + let source_struct = match resources.split_once("pub struct RawTransactionIngestStandardLogsSource {") { + std::option::Option::Some((_, tail)) => match tail.split_once("impl crate::RawTransactionIngestStandardLogsSource") { + std::option::Option::Some((value, _)) => value, + std::option::Option::None => "", + }, + std::option::Option::None => "", + }; + assert!(source_struct.contains("filter: ksp_onchain_transport_lib::SolanaLogsSubscribeFilter")); + for forbidden in ["logs:", "err:", "payload:", "body:", "url:"] { + assert!(!source_struct.contains(forbidden), "remote/sensitive material stored in standard logs source: {forbidden}"); + } + let projection = match resources.split_once("fn project_standard_logs_signal") { + std::option::Option::Some((_, tail)) => match tail.split_once("fn route_yellowstone_update") { + std::option::Option::Some((value, _)) => value, + std::option::Option::None => "", + }, + std::option::Option::None => "", + }; + for forbidden in [".logs()", ".err()"] { + assert!(!projection.contains(forbidden), "logs/error material copied into standard logs projection: {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"); 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 59e68ae..a59f1e8 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: 12 +// version: 13 //! External public-surface proofs for the RAW transaction ingest Worker foundation. @@ -197,6 +197,33 @@ fn v0_3_13_pre_002_multi_source_runtime_resource_surface_is_bounded_and_source_n return; } +#[test] +fn v0_3_13_pre_003_standard_logs_runtime_resource_surface_is_typed_and_transport_owned() { + let _source_new: fn( + ksp_onchain_transport_lib::WsEndpointSettings, + ksp_onchain_transport_lib::SolanaLogsSubscribeFilter, + ksp_onchain_transport_lib::SolanaCommitment, + ksp_onchain_transport_lib::HttpTransportPool, + ksp_onchain_transport_lib::HttpRoleName, + ) -> ksp_core_lib::Result = + ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestStandardLogsSource::new; + let _resources_from_logs: fn( + ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestStandardLogsSource, + ) -> ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestRuntimeResources = + ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestRuntimeResources::from_standard_logs_source; + let _push_logs: fn( + &mut ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestRuntimeResources, + ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestStandardLogsSource, + ) -> ksp_core_lib::Result<()> = ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestRuntimeResources::try_push_standard_logs_source; + let root = include_str!("../src/lib.rs"); + let resources = include_str!("../src/runtime_resources.rs"); + assert!(root.contains("RawTransactionIngestStandardLogsSource")); + for forbidden in ["pub fn ws_endpoint(", "pub fn filter(", "pub fn http_pool(", "pub fn source_key("] { + assert!(!resources.contains(forbidden), "standard logs 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 6f734ea..97d0bdb 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: 10 +// version: 11 //! Release-completeness canaries through the `pre.010` public/release/security hardening tranche. @@ -84,6 +84,7 @@ fn pre_010_public_root_export_inventory_is_exact() { "RawTransactionIngestSnapshotFuture", "RawTransactionIngestSnapshotSource", "RawTransactionIngestSourceState", + "RawTransactionIngestStandardLogsSource", "RawTransactionIngestTerminalFuture", "RawTransactionIngestWorker", "RawTransactionIngestYellowstoneSource", @@ -113,6 +114,7 @@ fn pre_010_external_hardening_suite_is_present_and_scoped() { "v0_3_12_pre_009_hydration_retry_ownership_and_no_orphan_cleanup_are_explicit", "pre_010_lower_layers_have_no_dependency_return_to_concrete_worker", "v0_3_12_pre_002_public_root_exposes_contract_types_without_transport_implementation_paths", + "v0_3_13_pre_003_standard_logs_redaction_and_reference_only_contract_are_explicit", ] { assert!(hardening.contains(required), "required pre.010 hardening canary missing: {required}"); } @@ -126,6 +128,7 @@ fn pre_010_external_hardening_suite_is_present_and_scoped() { assert!(dependency_boundary.contains("v0_3_12_pre_005_block_and_continuity_adapters_remain_private_and_transport_facade_only")); assert!(dependency_boundary.contains("v0_3_12_pre_007_processing_frontier_remains_run_local_and_backend_neutral_after_continuity_extension")); assert!(dependency_boundary.contains("v0_3_12_pre_008_worker_observes_transport_reconnect_replay_and_faults_only_on_proven_retention_gap")); + assert!(dependency_boundary.contains("v0_3_13_pre_003_standard_logs_source_reuses_transport_facades_and_common_hydration_only")); 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")); @@ -133,6 +136,7 @@ fn pre_010_external_hardening_suite_is_present_and_scoped() { assert!(public_api.contains("pre_009_source_and_drain_timeout_error_codes_are_public_and_stable")); assert!(public_api.contains("v0_3_12_pre_002_runtime_resource_types_are_consumable_without_client_escape_hatch")); 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_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 71ae2a6..15f4239 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: 12 +// version: 13 fn grpc_endpoint(cluster: &str) -> std::option::Option { return grpc_endpoint_with_identity(cluster, "yellowstone-fixture", "fixture-provider"); @@ -62,6 +62,69 @@ fn http_pool(cluster: &str, role_name: &str, request_kind: &str) -> std::option: }; } +fn ws_endpoint( + cluster: &str, + endpoint_name: &str, + provider: &str, + protocol: ksp_onchain_transport_lib::WsProtocolKind, +) -> std::option::Option { + let url = match ksp_onchain_transport_lib::WsEndpointUrl::parse("ws://127.0.0.1:10001/WS-SECRET-CANARY") { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return std::option::Option::None, + }; + return std::option::Option::Some(ksp_onchain_transport_lib::WsEndpointSettings::new( + endpoint_name, + true, + ksp_onchain_transport_lib::WsProviderName::new(provider), + ksp_onchain_transport_lib::WsClusterName::new(cluster), + protocol, + url, + ksp_onchain_transport_lib::WsSessionSettings::default(), + )); +} + +fn standard_logs_source( + cluster: &str, + endpoint_name: &str, + filter: ksp_onchain_transport_lib::SolanaLogsSubscribeFilter, + commitment: ksp_onchain_transport_lib::SolanaCommitment, + hydration_role: &str, +) -> std::option::Option { + let endpoint = match ws_endpoint(cluster, endpoint_name, "fixture-ws-provider", ksp_onchain_transport_lib::WsProtocolKind::SolanaStandard) { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::option::Option::None, + }; + let pool = match http_pool(cluster, hydration_role, "get_transaction") { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::option::Option::None, + }; + return match crate::RawTransactionIngestStandardLogsSource::new( + endpoint, + filter, + commitment, + pool, + ksp_onchain_transport_lib::HttpRoleName::new(hydration_role), + ) { + std::result::Result::Ok(value) => std::option::Option::Some(value), + std::result::Result::Err(_) => std::option::Option::None, + }; +} + +struct StandardLogsViewFixture { + signature: std::string::String, + slot: u64, +} + +impl super::RawTransactionIngestStandardLogsView for StandardLogsViewFixture { + fn signature(&self) -> &str { + return self.signature.as_str(); + } + + fn slot(&self) -> u64 { + return self.slot; + } +} + fn transaction_request( commitment: std::option::Option, ) -> std::option::Option { @@ -331,6 +394,30 @@ fn signal_source_for_http_url_with_commitment( }; } +fn standard_logs_source_for_http_url( + url: &str, + filter: ksp_onchain_transport_lib::SolanaLogsSubscribeFilter, +) -> std::option::Option { + let endpoint = match ws_endpoint("devnet", "standard-logs-fixture", "fixture-ws-provider", ksp_onchain_transport_lib::WsProtocolKind::SolanaStandard) { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::option::Option::None, + }; + let pool = match http_pool_for_url(url, "devnet", "http-hydration-fixture", "fixture-http-provider") { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::option::Option::None, + }; + return match crate::RawTransactionIngestStandardLogsSource::new( + endpoint, + filter, + ksp_onchain_transport_lib::SolanaCommitment::Confirmed, + pool, + ksp_onchain_transport_lib::HttpRoleName::new("hydration"), + ) { + std::result::Result::Ok(value) => std::option::Option::Some(value), + std::result::Result::Err(_) => std::option::Option::None, + }; +} + fn pre_004_settings() -> std::option::Option { let network = match ksp_store_lib::RawNetworkId::new("devnet") { std::result::Result::Ok(value) => value, @@ -1049,7 +1136,7 @@ fn pre_004_composite_provenance_rejects_overflow_without_truncation() { std::result::Result::Err(_) => return, }; let route = super::RawTransactionIngestSourceRoute { endpoint_id, provider }; - let result = super::composite_provenance_codes(&route, "h".repeat(120).as_str(), "http-endpoint"); + let result = super::composite_provenance_codes(&route, "ys", "h".repeat(120).as_str(), "http-endpoint"); let error = match result { std::result::Result::Ok(_) => return, std::result::Result::Err(error) => error, @@ -1100,7 +1187,7 @@ async fn pre_004_observed_get_transaction_closes_signal_to_common_raw_ingress_wi assert!(request_body.contains(PRE_004_ZERO_SIGNATURE_TEXT)); assert!(request_body.contains("\"commitment\":\"confirmed\"")); assert!(request_body.contains("\"encoding\":\"base64\"")); - assert!(request_body.contains("\"maxSupportedTransactionVersion\":0")); + assert!(request_body.contains("\"maxSupportedTransactionVersion\":1")); let ingress = match outcome { super::RawTransactionIngestHydrationOutcome::Available(value) => value, super::RawTransactionIngestHydrationOutcome::Missing(_) => return, @@ -1172,7 +1259,7 @@ async fn pre_004_finalized_hydration_preserves_request_and_provenance_commitment let request_body = http_request_body(request.as_str()); assert!(request_body.contains("\"commitment\":\"finalized\"")); assert!(request_body.contains("\"encoding\":\"base64\"")); - assert!(request_body.contains("\"maxSupportedTransactionVersion\":0")); + assert!(request_body.contains("\"maxSupportedTransactionVersion\":1")); let ingress = match outcome { super::RawTransactionIngestHydrationOutcome::Available(value) => value, super::RawTransactionIngestHydrationOutcome::Missing(_) => return, @@ -1421,11 +1508,12 @@ async fn pre_006_coalescence_key_merges_transaction_and_status_before_http_fanou std::option::Option::Some(value) => value, std::option::Option::None => return, }; - let transaction_key = match super::hydration_key(&source, &transaction) { + let hydration = source.hydration_context(); + let transaction_key = match super::hydration_key(&hydration, &transaction) { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return, }; - let status_key = match super::hydration_key(&source, &status) { + let status_key = match super::hydration_key(&hydration, &status) { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return, }; @@ -1437,10 +1525,10 @@ async fn pre_006_coalescence_key_merges_transaction_and_status_before_http_fanou let mut coordinator = super::RawTransactionIngestHydrationCoordinator::new(&settings); let (frontier_sender, _frontier_receiver) = tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty()); let mut processing_frontier = super::RawTransactionIngestProcessingFrontierReporter::new(frontier_sender); - if coordinator.queue_signal(&source, transaction, received_at, &mut processing_frontier).is_err() { + if coordinator.queue_signal(&hydration, transaction, received_at, &mut processing_frontier).is_err() { return; } - if coordinator.queue_signal(&source, status, received_at, &mut processing_frontier).is_err() { + if coordinator.queue_signal(&hydration, status, received_at, &mut processing_frontier).is_err() { return; } assert_eq!(coordinator.pending.len(), 1); @@ -1467,11 +1555,12 @@ fn pre_006_coalescence_key_separates_network_signature_and_commitment() { std::option::Option::Some(value) => value, std::option::Option::None => return, }; - let first_key = match super::hydration_key(&source, &first) { + let hydration = source.hydration_context(); + let first_key = match super::hydration_key(&hydration, &first) { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return, }; - let second_key = match super::hydration_key(&source, &second) { + let second_key = match super::hydration_key(&hydration, &second) { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return, }; @@ -1481,6 +1570,273 @@ fn pre_006_coalescence_key_separates_network_signature_and_commitment() { return; } +#[test] +fn v0_3_13_pre_003_standard_logs_source_validates_protocol_commitment_and_hydration_route() { + let accepted = standard_logs_source( + "devnet", + "standard-logs-a", + ksp_onchain_transport_lib::SolanaLogsSubscribeFilter::All, + ksp_onchain_transport_lib::SolanaCommitment::Confirmed, + "hydration", + ); + let accepted = match accepted { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let resources = crate::RawTransactionIngestRuntimeResources::from_standard_logs_source(accepted); + assert_eq!(resources.source_count(), 1); + let helius_endpoint = + match ws_endpoint("devnet", "standard-logs-helius", "fixture-ws-provider", ksp_onchain_transport_lib::WsProtocolKind::HeliusLaserStream) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let pool = match http_pool("devnet", "hydration", "get_transaction") { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let protocol_error = crate::RawTransactionIngestStandardLogsSource::new( + helius_endpoint, + ksp_onchain_transport_lib::SolanaLogsSubscribeFilter::All, + ksp_onchain_transport_lib::SolanaCommitment::Confirmed, + pool.clone(), + ksp_onchain_transport_lib::HttpRoleName::new("hydration"), + ); + let protocol_error = match protocol_error { + std::result::Result::Ok(_) => return, + std::result::Result::Err(value) => value, + }; + assert!(protocol_error.context().iter().any(|context| return context.value() == "runtime_resources.standard_logs_protocol_invalid")); + let standard_endpoint = + match ws_endpoint("devnet", "standard-logs-processed", "fixture-ws-provider", ksp_onchain_transport_lib::WsProtocolKind::SolanaStandard) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let commitment_error = crate::RawTransactionIngestStandardLogsSource::new( + standard_endpoint, + ksp_onchain_transport_lib::SolanaLogsSubscribeFilter::AllWithVotes, + ksp_onchain_transport_lib::SolanaCommitment::Processed, + pool, + ksp_onchain_transport_lib::HttpRoleName::new("hydration"), + ); + let commitment_error = match commitment_error { + std::result::Result::Ok(_) => return, + std::result::Result::Err(value) => value, + }; + assert!(commitment_error.context().iter().any(|context| return context.value() == "runtime_resources.hydration_commitment_invalid")); + let cross_network_endpoint = + match ws_endpoint("devnet", "standard-logs-cross-network", "fixture-ws-provider", ksp_onchain_transport_lib::WsProtocolKind::SolanaStandard) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let cross_network_pool = match http_pool("testnet", "hydration", "get_transaction") { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let cross_network = crate::RawTransactionIngestStandardLogsSource::new( + cross_network_endpoint, + ksp_onchain_transport_lib::SolanaLogsSubscribeFilter::All, + ksp_onchain_transport_lib::SolanaCommitment::Confirmed, + cross_network_pool, + ksp_onchain_transport_lib::HttpRoleName::new("hydration"), + ); + let cross_network = match cross_network { + std::result::Result::Ok(_) => return, + std::result::Result::Err(value) => value, + }; + assert!(cross_network.context().iter().any(|context| return context.value() == "runtime_resources.transport_network_mismatch")); + return; +} + +#[test] +fn v0_3_13_pre_003_standard_logs_source_key_is_filter_sensitive_hydration_role_neutral_and_redacted() { + let all_a = match standard_logs_source( + "devnet", + "standard-logs-key", + ksp_onchain_transport_lib::SolanaLogsSubscribeFilter::All, + ksp_onchain_transport_lib::SolanaCommitment::Confirmed, + "hydration", + ) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let all_b = match standard_logs_source( + "devnet", + "standard-logs-key", + ksp_onchain_transport_lib::SolanaLogsSubscribeFilter::All, + ksp_onchain_transport_lib::SolanaCommitment::Confirmed, + "secondary_hydration", + ) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let votes = match standard_logs_source( + "devnet", + "standard-logs-key", + ksp_onchain_transport_lib::SolanaLogsSubscribeFilter::AllWithVotes, + ksp_onchain_transport_lib::SolanaCommitment::Confirmed, + "hydration", + ) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + assert_eq!(all_a.source_key, all_b.source_key); + assert_ne!(all_a.source_key, votes.source_key); + let mention = ksp_core_lib::Pubkey::new_from_array([0x5a_u8; 32]); + let mention_text = mention.to_string(); + let mentioned = match standard_logs_source( + "devnet", + "standard-logs-key", + ksp_onchain_transport_lib::SolanaLogsSubscribeFilter::Mentions(mention), + ksp_onchain_transport_lib::SolanaCommitment::Confirmed, + "hydration", + ) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + assert_ne!(all_a.source_key, mentioned.source_key); + let debug = std::format!("{mentioned:?}"); + assert!(debug.contains("filter_kind")); + assert!(debug.contains("mentions")); + assert!(!debug.contains(mention_text.as_str())); + assert!(!debug.contains("WS-SECRET-CANARY")); + assert!(!debug.contains("source_key: [")); + return; +} + +#[test] +fn v0_3_13_pre_003_standard_logs_projection_keeps_only_reference_material() { + let source = match standard_logs_source( + "devnet", + "standard-logs-projection", + ksp_onchain_transport_lib::SolanaLogsSubscribeFilter::All, + ksp_onchain_transport_lib::SolanaCommitment::Finalized, + "hydration", + ) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let fixture = StandardLogsViewFixture { signature: PRE_004_ZERO_SIGNATURE_TEXT.to_owned(), slot: 73 }; + let signal = match super::project_standard_logs_signal(&source, &fixture) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return, + }; + assert_eq!(signal.family, super::RawTransactionIngestSourceFamily::Logs); + assert_eq!(signal.slot, 73); + assert_eq!(signal.transaction_index, std::option::Option::None); + assert_eq!(signal.created_at, std::option::Option::None); + assert_eq!(signal.matched_filter_count, 1); + assert_eq!(signal.matched_filter_id, std::option::Option::None); + assert_eq!(signal.signature, ksp_store_lib::RawTransactionSignature::new([0_u8; 64])); + let debug = std::format!("{signal:?}"); + assert!(!debug.contains(PRE_004_ZERO_SIGNATURE_TEXT)); + return; +} + +#[tokio::test(flavor = "current_thread")] +async fn v0_3_13_pre_003_standard_logs_reference_hydrates_once_to_exact_common_raw() { + let body = std::format!( + "{{\"jsonrpc\":\"2.0\",\"result\":{{\"slot\":42,\"blockTime\":1760000120,\"transaction\":[\"{}\",\"base64\"],\"meta\":{{\"err\":null,\"fee\":5000}},\"version\":\"legacy\",\"transactionIndex\":7}},\"id\":1}}", + PRE_004_ZERO_TRANSACTION_BASE64, + ); + let (url, server) = match serve_http_once(body) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return, + }; + let source = match standard_logs_source_for_http_url(url.as_str(), ksp_onchain_transport_lib::SolanaLogsSubscribeFilter::All) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let settings = match pre_004_settings() { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let fixture = StandardLogsViewFixture { signature: PRE_004_ZERO_SIGNATURE_TEXT.to_owned(), slot: 42 }; + let signal = match super::project_standard_logs_signal(&source, &fixture) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return, + }; + let received_at = match ksp_store_lib::RawTimestamp::from_unix_millis(1_760_000_200_000) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return, + }; + let hydration = source.hydration_context(); + let key = match super::hydration_key(&hydration, &signal) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return, + }; + let fetched = + super::fetch_hydration(hydration.http_pool.clone(), hydration.hydration_role.clone(), hydration.network.clone(), key, hydration.commitment).await; + let fetched = match fetched { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return, + }; + let ingress = super::finalize_hydration(&hydration, &settings, signal, received_at, &fetched.observed); + let ingress = match ingress { + std::result::Result::Ok(std::option::Option::Some(value)) => value, + std::result::Result::Ok(std::option::Option::None) | std::result::Result::Err(_) => return, + }; + assert_eq!(ingress.provenance.protocol().as_str(), "solana_ws_http"); + assert_eq!(ingress.provenance.acquisition_method().as_str(), "logs_get_transaction"); + assert_eq!(ingress.material.slot(), 42); + assert_eq!(ingress.material.signature(), &ksp_store_lib::RawTransactionSignature::new([0_u8; 64])); + let (mut admission, sender) = crate::RawTransactionAdmission::new(1); + if sender.send(ingress).await.is_err() { + return; + } + std::mem::drop(sender); + let acquisition = match admission.receive(settings.network()).await { + std::result::Result::Ok(std::option::Option::Some(value)) => value, + _ => return, + }; + assert_eq!(acquisition.transaction().reference().network().as_str(), "devnet"); + assert_eq!(acquisition.transaction().reference().signature().as_bytes(), &[0_u8; 64]); + assert_eq!(acquisition.transaction().slot(), 42); + assert_eq!(acquisition.transaction().block_time().map(|value| return value.unix_millis()), std::option::Option::Some(1_760_000_120_000)); + assert_eq!(acquisition.observation().provenance().protocol().as_str(), "solana_ws_http"); + assert_eq!(acquisition.observation().provenance().acquisition_method().as_str(), "logs_get_transaction"); + let request_body = match server.join() { + std::result::Result::Ok(std::result::Result::Ok(value)) => value, + std::result::Result::Ok(std::result::Result::Err(_)) | std::result::Result::Err(_) => return, + }; + assert!(request_body.contains("\"method\":\"getTransaction\"")); + assert!(request_body.contains("\"encoding\":\"base64\"")); + assert!(request_body.contains("\"maxSupportedTransactionVersion\":1")); + return; +} + +#[test] +fn v0_3_13_pre_003_runtime_resources_reject_duplicate_standard_logs_identity() { + let first = match standard_logs_source( + "devnet", + "standard-logs-duplicate", + ksp_onchain_transport_lib::SolanaLogsSubscribeFilter::AllWithVotes, + ksp_onchain_transport_lib::SolanaCommitment::Confirmed, + "hydration", + ) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let duplicate = match standard_logs_source( + "devnet", + "standard-logs-duplicate", + ksp_onchain_transport_lib::SolanaLogsSubscribeFilter::AllWithVotes, + ksp_onchain_transport_lib::SolanaCommitment::Confirmed, + "secondary_hydration", + ) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let mut resources = crate::RawTransactionIngestRuntimeResources::from_standard_logs_source(first); + let result = resources.try_push_standard_logs_source(duplicate); + let error = match result { + std::result::Result::Ok(()) => return, + std::result::Result::Err(value) => value, + }; + assert!(error.context().iter().any(|context| return context.value() == "runtime_resources.duplicate_source_identity")); + assert_eq!(resources.source_count(), 1); + return; +} + #[test] fn pre_007_processing_frontier_never_advances_through_pending_work() { let mut frontier = super::RawTransactionIngestProcessingFrontier::new(); @@ -1593,12 +1949,13 @@ async fn pre_009_duplicate_storm_is_bounded_coalesced_and_abort_leaves_no_orphan coordinator.max_pending_signals = 3; let (frontier_sender, frontier_receiver) = tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty()); let mut processing_frontier = super::RawTransactionIngestProcessingFrontierReporter::new(frontier_sender); + let hydration = source.hydration_context(); assert!(processing_frontier.observe_settled(9).is_ok()); assert!(processing_frontier.observe_settled(10).is_ok()); for _ in 0..3 { - assert!(coordinator.queue_signal(&source, signal.clone(), received_at, &mut processing_frontier).is_ok()); + assert!(coordinator.queue_signal(&hydration, signal.clone(), received_at, &mut processing_frontier).is_ok()); } - let saturated = coordinator.queue_signal(&source, signal, received_at, &mut processing_frontier); + let saturated = coordinator.queue_signal(&hydration, signal, received_at, &mut processing_frontier); let saturated = match saturated { std::result::Result::Ok(()) => return, std::result::Result::Err(value) => value, @@ -1678,7 +2035,8 @@ async fn pre_009_abort_joins_in_flight_hydration_and_clears_pending_projection() let mut coordinator = super::RawTransactionIngestHydrationCoordinator::new(&settings); let (frontier_sender, frontier_receiver) = tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty()); let mut processing_frontier = super::RawTransactionIngestProcessingFrontierReporter::new(frontier_sender); - assert!(coordinator.queue_signal(&source, signal, received_at, &mut processing_frontier).is_ok()); + let hydration = source.hydration_context(); + assert!(coordinator.queue_signal(&hydration, signal, received_at, &mut processing_frontier).is_ok()); assert_eq!(coordinator.pending_signal_count, 1); assert_eq!(coordinator.tasks.len(), 1); assert_eq!(frontier_receiver.borrow().hydration_pending(), 1); diff --git a/deltas/0.3.13/pre.003.md b/deltas/0.3.13/pre.003.md new file mode 100644 index 0000000..0e826b0 --- /dev/null +++ b/deltas/0.3.13/pre.003.md @@ -0,0 +1,260 @@ + + + +# Delta 0.3.13-pre.003 — Standard Solana logsSubscribe + hydration commune + +## Base requise + +```text +delivery précédente : 0.3.13-pre.002-fix.001 +Cargo base : 0.3.13-pre.2.fix.1 +delta base : deltas/0.3.13/pre.002-fix.001.md +``` + +Le gate opérateur de la base a été exécuté après `cargo clean` et est vert : `cargo fmt --all -- --check` silencieux, audits Rust/Markdown propres, `cargo check --workspace`, Clippy strict, 389 tests unitaires Transport, 52 tests public API Transport, 44 tests release Transport, 4 doc-tests Transport, puis 70 tests unitaires Worker et toutes ses suites d'intégration sans échec. Les smokes réseau opt-in sont restés ignorés comme prévu. + +## Objectif + +Ajouter la première nouvelle famille live de `0.3.13` sans dupliquer Transport ni l'admission RAW : + +```text +Solana standard WS logsSubscribe + -> context.slot + signature + -> hydration HTTP getTransaction observed + -> Common RAW + -> admission centrale Worker + -> Store +``` + +La tranche ne démarre toujours qu'une seule source productive à la fois. La supervision simultanée de plusieurs sources reste réservée à `pre.007`. + +## Version + +```text +livraison : 0.3.13-pre.003 +workspace.package.version : 0.3.13-pre.3 +archive attendue : ksp-general-0.3.13-pre.003.zip +``` + +## Source Standard Logs + +Nouvelle surface publique Worker : + +```text +RawTransactionIngestStandardLogsSource +``` + +Construction caller-composed : + +```text +WsEndpointSettings kind solana_standard +SolanaLogsSubscribeFilter +SolanaCommitment Confirmed ou Finalized +HttpTransportPool +HttpRoleName hydration +``` + +La construction ne réalise aucune I/O. Elle valide le protocole WebSocket, le commitment, le réseau, la représentation sûre provider/endpoint et l'existence d'une route HTTP `getTransaction` compatible sur le même cluster. + +Au runtime, la source utilise exclusivement : + +```text +SolanaStandardWsSession::connect +logs_subscribe +HttpTransportPool::get_transaction_observed +``` + +Le Worker ne construit aucun socket/client inférieur et ne dépend directement ni de `reqwest`, ni de `tokio-tungstenite`, ni de `tonic`, ni de `yellowstone-grpc-proto`. + +## Projection et redaction + +Une notification `logsSubscribe` est réduite à son matériau de référence nécessaire : + +```text +context.slot +signature +commitment caller-composed +identité source/filter privée +``` + +Les champs distants `logs` et `err` ne sont pas copiés dans le signal Worker, le snapshot, la provenance textuelle ou `Debug`. + +Les variantes : + +```text +All +AllWithVotes +Mentions(pubkey) +``` + +participent à une empreinte SHA-256 privée. La valeur du pubkey `Mentions` n'est pas exposée par le Worker. + +## Identité logique + +Le `source_key` Standard Logs est dérivé de : + +```text +domain KSP live-source v1 +family = standard_logs +network +safe provider identity +safe endpoint identity +commitment +private filter fingerprint +``` + +Le rôle et le pool HTTP d'hydration n'entrent pas dans cette identité : ils décrivent la stratégie d'enrichissement, pas une nouvelle source live. + +Le rejet transactionnel des doublons et le bound global `1..32` introduits en `pre.002` s'appliquent aussi à Standard Logs. + +## Hydration source-neutral + +Le coordinator privé précédemment utilisé par Yellowstone est généralisé en contrat d'hydration source-neutral et réutilisé par Standard Logs. + +Pour `pre.003`, chaque exécution mono-source possède encore sa propre instance. Le partage réellement global entre plusieurs tâches source sera introduit avec le supervisor `pre.007`. + +Bornes effectives : + +```text +hydration in-flight <= persistence_concurrency +pending source signals <= admission_queue_capacity +``` + +Le plafond pending est donc aligné sur les settings effectifs du Worker plutôt que sur le plafond absolu maximal. + +La requête commune `getTransaction observed` utilise maintenant : + +```text +encoding = base64 +maxSupportedTransactionVersion = 1 +``` + +Cette adaptation s'applique également à Yellowstone. Elle prépare l'hydration aux transactions Legacy/V0/V1 mais ne qualifie aucun nouveau chemin RAW-direct V1. + +## Activation runtime + +Le runtime-resource aggregate peut composer Yellowstone et Standard Logs dans la collection privée bornée, mais la barrière de sûreté reste : + +```text +1 source Yellowstone : productif +1 source Standard Logs : productif +2..32 sources : composition validable +start 2..32 : runtime_resources.multi_source_activation_pending avant spawn +``` + +Aucune source configurée n'est silencieusement ignorée. + +## Preuves ajoutées + +Unit tests Worker : + +```text +validation standard WS / commitment / route HTTP +source_key stable et filter-sensitive +source_key indépendant du hydration role +All / AllWithVotes / Mentions distingués +Mentions/URL/source-key bytes redacted +projection reference-only +fixture signature+slot -> une hydration HTTP -> matériau Common RAW exact +getTransaction base64 + maxSupportedTransactionVersion=1 +rejet duplicate Standard Logs identity +``` + +Canaris externes : + +```text +public_api : nouvelle construction/composition accessible depuis crate root +dependency_boundary : uniquement façades Transport + Common RAW/Store existants +hardening : aucune copie logs/err/payload/url et Debug borné +release_completeness : nouvelle surface et canaris pre.003 obligatoires +``` + +Reconnect, resubscribe, IDs distants et backpressure WebSocket restent couverts et possédés par `ksp-onchain-transport-lib`; le Worker réutilise la façade `SolanaStandardWsSession` au lieu de reproduire ces mécanismes. + +## Fichiers modifiés + +```text +Cargo.toml +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.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.003.md +``` + +## Fichiers supprimés + +```text +aucun +``` + +## Non-claims + +`pre.003` ne prétend pas : + +```text +blockSubscribe productif +Helius transactionSubscribe productif +HTTP live block polling +supervision simultanée de plusieurs sources +fanout Store d'observations multi-source +coordinator partagé entre plusieurs tâches source simultanées +gap repair/backfill +nouvelle dépendance externe +``` + +## Validation dans l'environnement d'assemblage + +Les contrôles réellement exécutés après finalisation de la tranche sont : + +```text +python3 scripts/audit_rust_workspace_rules.py +PASS : General Rust rule audit clean +PASS : Rust export completeness audit 0 candidate +PASS : KSP workspace Rust rule audit clean + +python3 scripts/audit_markdown_tables.py README.md RULES.md ROADMAP.md CHANGELOG.md docs prompts crates deltas +PASS : Markdown table audit clean (340 tables, 831 files) + +supplemental normative definition-id scan +PASS : 489 definitions, 489 unique, 0 duplicate definition id + +pre-packaging exhaustive diff scan +PASS : 13 fichiers modifiés, 1 ajouté, 0 supprimé +PASS : les 13 headers versionnés des fichiers modifiés sont incrémentés exactement de +1 +PASS : aucune dépendance Cargo nouvelle +``` + +Le toolchain Rust/Cargo/Rustfmt n'est pas disponible dans cet environnement. Les commandes Cargo post-modification sont donc explicitement `NON EXÉCUTÉ LOCAL`, jamais déclarées PASS. + +Les caches Python produits par les audits sont supprimés avant packaging et ne font pas partie du delta. + +## Gate opérateur requis avant pre.004 + +```bash +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.004` reste bloquée si ce gate révèle une anomalie de format, règle, compilation, API, dépendance ou test. 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 26c4fa2..a235a09 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 @@ -460,6 +460,26 @@ Hydration retry/rate limit : Transport HTTP. Preuve : fixture logs notification -> une seule hydration -> exact Common RAW golden ; mentions/all/allWithVotes ; stop/reconnect/backpressure ; redaction des logs. +### 13.1 Décisions matérialisées en pre.003 + +La tranche conserve la barrière d’activation simultanée de `pre.002` : une source unique Yellowstone ou Standard Logs est productive ; une collection `2..32` reste rejetée avant spawn jusqu’au supervisor `pre.007`. Il n’est donc pas créé de supervisor provisoire qui ignorerait une partie de la configuration. + +Le coordinateur d’hydration Yellowstone devient un contrat privé source-neutral réutilisé par Standard Logs. Pour cette tranche, chaque exécution mono-source en possède une instance ; la possession réellement globale entre plusieurs sources n’arrive qu’avec le supervisor `pre.007`. Cette distinction corrige le raccourci de wording « coordinator global » du sizing initial sans modifier l’objectif de convergence. + +Bornes retenues : + +```text +hydration in-flight <= persistence_concurrency +pending source signals <= admission_queue_capacity +aucune queue Worker non bornée +``` + +Le `getTransaction observed` commun passe à `maxSupportedTransactionVersion = 1` pour Yellowstone comme pour Standard Logs. Cela ne déclare aucun chemin RAW-direct V1 ; il s’agit uniquement de rendre l’hydration HTTP version-aware conformément au contrat `0.3.13`. + +Pour Standard Logs, l’identité logique inclut le réseau, la famille `standard_logs`, l’identité provider/endpoint sûre, le commitment et l’empreinte privée du filtre. Le rôle/pool HTTP d’hydration n’entre pas dans cette identité. Les variantes `All`, `AllWithVotes` et `Mentions(pubkey)` sont distinguées sans recopier le pubkey dans `Debug`, snapshot ou diagnostics. + +Les valeurs `logs` et `err` de la notification restent entièrement dans Transport : le Worker ne projette que `context.slot` et `signature`. + ## 14. Contrat source WS standard blockSubscribe Entrée caller-composed : @@ -776,7 +796,7 @@ Collection bornée 1..32, source key, discriminants capability-owned, validation ### pre.003 — WS standard logsSubscribe + hydration -Source logs, signal privé, intégration coordinator global et hydration observed. +Source logs, signal privé, réutilisation du coordinator d’hydration source-neutral et hydration observed ; activation simultanée toujours différée à pre.007. ### pre.004 — WS standard blockSubscribe 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 103b820..e3bc7e2 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 et l'identité logique sans activer encore la supervision simultanée ni une nouvelle famille de source productive. +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 et `pre.003` ajoute la source productive Standard Solana `logsSubscribe` + hydration HTTP observée, sans activer encore la supervision simultanée. ## 2. Archive stable contrôlée @@ -658,3 +658,156 @@ cargo tree --duplicates `pre.003` ne doit pas commencer si cette livraison révèle une anomalie de règles, API, version, archive ou dépendance non corrigée. +## 28. Gate opérateur pre.002-fix.001 reçu avant pre.003 + +Le correctif de format `0.3.13-pre.002-fix.001` a été validé par l’opérateur après `cargo clean` : + +```text +cargo fmt --all : PASS +cargo fmt --all -- --check : PASS, silencieux +General Rust rule audit : clean +Rust export completeness audit : 0 candidate +KSP workspace Rust rule audit : clean +Markdown table audit : clean (340 tables, 830 files) +cargo check --workspace : PASS +clippy workspace/all-targets/all-features -D warnings : PASS +ksp-onchain-transport-lib unit tests : 389 pass +transport public_api : 52 pass +transport release_completeness : 44 pass +transport doc-tests : 4 pass +live transport smokes : ignored opt-in, comme prévu +ksp-worker-raw-transaction-ingest-lib unit tests : 70 pass +cross_layer_completeness : 4 pass +dependency_boundary : 9 pass +hardening : 16 pass +public_api : 10 pass +release_completeness : 4 pass +worker doc-tests : 0 +``` + +Le gate `pre.002-fix.001 -> pre.003` est satisfait. + +## 29. Implémentation pre.003 — Standard Logs + hydration + +`pre.003` ajoute `RawTransactionIngestStandardLogsSource`, construite exclusivement à partir de surfaces Transport déjà publiques : + +```text +WsEndpointSettings kind solana_standard +SolanaLogsSubscribeFilter +SolanaCommitment Confirmed ou Finalized +HttpTransportPool +HttpRoleName hydration +``` + +La construction reste sans I/O et valide avant spawn : endpoint/protocole, commitment, réseau, provenance sûre et existence d’une route `getTransaction` compatible sur le même cluster. Aucun endpoint URL, filtre brut, credential ou client inférieur n’est exposé par l’API Worker. + +Au runtime : + +```text +SolanaStandardWsSession::connect + -> logs_subscribe + -> context.slot + signature seulement + -> coordinator d’hydration Worker source-neutral + -> get_transaction_observed + encoding = base64 + maxSupportedTransactionVersion = 1 + -> Common RAW + -> admission centrale existante + -> Store +``` + +`logs` et `err` ne sont jamais copiés dans le signal Worker. Le reconnect, le resubscribe et le backpressure de la subscription restent possédés par le même acteur WebSocket Transport. Retry/reroutage/rate-limit HTTP restent possédés par Transport. + +La voie Yellowstone réutilise le même contrat d’hydration source-neutral et passe elle aussi à `maxSupportedTransactionVersion = 1`; aucune nouvelle queue, aucun second client HTTP et aucun acteur WebSocket Worker ne sont introduits. + +## 30. Identité et bornes pre.003 + +L’identité logique Standard Logs est déterministe et opaque : + +```text +domain KSP live source key +family = standard_logs +network +safe provider identity +safe endpoint identity +commitment +private filter fingerprint +SHA-256 -> [u8; 32] +``` + +Le rôle/pool HTTP d’hydration est volontairement exclu de l’identité live. Les trois variantes `All`, `AllWithVotes`, `Mentions(pubkey)` produisent des empreintes distinctes ; la valeur d’un pubkey `Mentions` n’apparaît pas dans `Debug`. + +Les bornes du coordinator sont maintenant alignées sur les settings effectifs : + +```text +in-flight hydration <= persistence_concurrency +pending source signals <= admission_queue_capacity +``` + +Une activation mono-source Yellowstone ou Standard Logs est autorisée. Une composition `2..32` reste validable mais son démarrage demeure `runtime_resources.multi_source_activation_pending` avant spawn jusqu’à `pre.007`. + +## 31. Preuves déterministes ajoutées pre.003 + +Worker unit tests ajoutés : + +```text +validation protocole/commitment/route HTTP Standard Logs +source_key filter-sensitive et hydration-role-neutral +redaction Mentions/URL/source-key bytes +projection reference-only slot/signature +fixture logs -> une hydration HTTP -> exact matériau Common RAW +requête getTransaction = base64 + maxSupportedTransactionVersion 1 +rejet duplicate Standard Logs identity +``` + +Canaris externes ajoutés/étendus : + +```text +public_api : construction et composition Standard Logs depuis crate root +dependency_boundary : Transport facade only, sans Config/Job/backend/direct network crate +hardening : aucune copie logs/err/payload/url, Debug borné +release_completeness : nouvelle surface et nouveaux canaris obligatoires +``` + +Les preuves de reconnect/resubscribe/backpressure restent celles de `ksp-onchain-transport-lib`, car le Worker délègue ces mécanismes au `SolanaStandardWsSession` existant plutôt que de les réimplémenter. + +## 32. Validations locales post-modification pre.003 + +Le toolchain Cargo/Rustfmt n’est pas disponible dans l’environnement d’assemblage. Les commandes Cargo post-`pre.003` sont donc `NON EXÉCUTÉ LOCAL` et devront être confirmées par l’opérateur. + +Les audits statiques réellement exécutés sont consignés dans `deltas/0.3.13/pre.003.md`. + +## 33. Non-claims pre.003 + +`pre.003` ne prétend pas : + +```text +avoir ajouté blockSubscribe productif +avoir ajouté Helius transactionSubscribe productif +avoir ajouté HTTP live polling +avoir activé plusieurs sources simultanément +avoir implémenté le fanout d’observations Store multi-source +avoir rendu le coordinator partagé entre plusieurs tâches source simultanées +avoir ajouté du gap repair ou du backfill +avoir ajouté une dépendance réseau directe au Worker +avoir exécuté Cargo localement après modification +``` + +## 34. Gate opérateur requis avant pre.004 + +Après application du delta : + +```bash +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.004` ne doit pas commencer si une anomalie de format, règle, compilation, API, dépendance ou test apparaît.