From d17161234a53e2bbbcf763e183fe3fa6611ec900 Mon Sep 17 00:00:00 2001 From: SinuS Von SifriduS Date: Sat, 22 Aug 2026 23:08:47 +0200 Subject: [PATCH] v0.2.7-pre.009 --- Cargo.toml | 4 +- crates/ksp-onchain-transport-lib/README.md | 22 +- crates/ksp-onchain-transport-lib/USAGE.md | 36 ++- crates/ksp-onchain-transport-lib/src/lib.rs | 28 +- .../src/rpc_accounts.rs | 12 +- .../src/rpc_common.rs | 14 +- .../src/ws_accounts.rs | 278 ++++++++++++++++++ .../src/ws_session.rs | 7 +- .../src/ws_transactions.rs | 104 +++++++ .../tests/public_api.rs | 27 +- .../unit_tests/ws_accounts.rs | 202 +++++++++++++ .../unit_tests/ws_transactions.rs | 122 ++++++++ deltas/0.2.7/pre.009.md | 197 +++++++++++++ .../014-V0_2_7_ONCHAIN_WEBSOCKET_PLAN.md | 34 ++- .../010-V0_2_7_ONCHAIN_WEBSOCKET.md | 41 ++- 15 files changed, 1093 insertions(+), 35 deletions(-) create mode 100644 crates/ksp-onchain-transport-lib/src/ws_accounts.rs create mode 100644 crates/ksp-onchain-transport-lib/src/ws_transactions.rs create mode 100644 crates/ksp-onchain-transport-lib/unit_tests/ws_accounts.rs create mode 100644 crates/ksp-onchain-transport-lib/unit_tests/ws_transactions.rs create mode 100644 deltas/0.2.7/pre.009.md diff --git a/Cargo.toml b/Cargo.toml index 0638db8..2e222f6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,12 +1,12 @@ # file: Cargo.toml -# version: 204 +# version: 205 [workspace] resolver = "3" members = ["crates/ksp-app-config-desk", "crates/ksp-app-wallet-desk", "crates/ksp-config-lib", "crates/ksp-core-lib", "crates/ksp-logging-lib", "crates/ksp-onchain-transport-lib", "crates/ksp-wallet-lib"] [workspace.package] -version = "0.2.7-pre.8.fix.1" +version = "0.2.7-pre.9" edition = "2024" license = "MIT" repository = "https://git.sasedev.com/Sasedev/khadhroony-solana-project" diff --git a/crates/ksp-onchain-transport-lib/README.md b/crates/ksp-onchain-transport-lib/README.md index fbb55b8..cdf69af 100644 --- a/crates/ksp-onchain-transport-lib/README.md +++ b/crates/ksp-onchain-transport-lib/README.md @@ -1,5 +1,5 @@ - + # `ksp-onchain-transport-lib` @@ -117,7 +117,7 @@ Ping/Pong/Close sont traités comme control frames : le Pong automatique Tungste Le même actor possède maintenant le registre des subscriptions logiques, sans exposer les IDs numériques distants. Chaque subscription reçoit un `WsSubscriptionId` local stable, et le mapping `remote_subscription_id -> WsSubscriptionId` reste strictement runtime/interne. -La création générique typed reste `pub(crate)` jusqu'aux wrappers standard des tranches `pre.009+`. Le handle public `WsSubscription` expose uniquement : +La création générique typed reste `pub(crate)` et n'est jamais exposée comme API raw provider-extension. À partir de `pre.009`, les wrappers standard publics l'utilisent derrière leurs DTOs et paramètres typés. Le handle public `WsSubscription` expose uniquement : - `id()` et `kind()` ; - `state()` ; @@ -146,6 +146,24 @@ Les autres terminaisons en échec publient également un code KSP sûr sur le ha Les fixtures adversariales prouvent l'isolation d'un consumer lent, la survie d'une subscription saine, le cleanup distant best-effort, la réutilisation de capacité après unsubscribe ou abandon du receiver et la conservation du compteur d'overflow à travers les snapshots. Aucune promesse de livraison lossless n'est ajoutée. +### Wrappers stables lot A `0.2.7-pre.009` + +Les trois premiers wrappers WebSocket standards sont publics sur `WsSession` : + +```text +account_subscribe -> WsSubscription> +program_subscribe -> WsSubscription +logs_subscribe -> WsSubscription> +``` + +`SolanaAccountSubscribeConfig` expose uniquement les options réellement effectives du PubSub audité : `encoding`, `dataSlice` et `commitment`. `minContextSlot` reste volontairement absent car le handler Agave ciblé l'ignore pour `accountSubscribe`; KSP ne transforme donc pas un champ partagé mais inopérant en promesse WebSocket. + +`SolanaProgramSubscribeConfig` réutilise les encodings, slices et commitments account, ajoute les filtres programme et conserve `withContext`. Le décodeur `SolanaProgramNotification` accepte aussi bien le keyed account non contexté que la forme `RpcResponse` contextée afin de préserver les deux formes retenues par l'audit sans perte. `sortResults`, présent sur la surface HTTP `getProgramAccounts`, n'est pas exposé ici car le handler PubSub audité ne le consomme pas. + +`SolanaLogsSubscribeFilter` rend les trois filtres upstream explicites : `All`, `AllWithVotes` et `Mentions(Pubkey)`. La variante `Mentions` encode par construction exactement une adresse. `SolanaLogsNotification` conserve la signature opaque, le `err` nullable et l'ordre des messages `logs`, enveloppés dans `SolanaRpcResponse`. + +L'unsubscribe de ces trois familles passe toujours par `WsSubscription::unsubscribe()`: le caller ne voit ni ne fournit l'ID serveur. Les paramètres initiaux restent conservés par l'actor pour le resubscribe déterministe acquis en `pre.007`, et toutes les règles de backpressure/terminal error acquises en `pre.008` s'appliquent sans branche spéciale aux nouveaux DTOs publics. + ## Résilience L'admission est calculée par couple endpoint/rôle. Le pool applique : diff --git a/crates/ksp-onchain-transport-lib/USAGE.md b/crates/ksp-onchain-transport-lib/USAGE.md index db11f66..37ccd6f 100644 --- a/crates/ksp-onchain-transport-lib/USAGE.md +++ b/crates/ksp-onchain-transport-lib/USAGE.md @@ -1,5 +1,5 @@ - + # Utilisation de `ksp-onchain-transport-lib` @@ -92,12 +92,40 @@ let snapshot = session.snapshot(); Deux appels `WsSession::connect` avec le même endpoint créent volontairement deux connexions physiques distinctes. Il n'existe encore aucun pool de sessions automatique. -Le socket brut et la primitive JSON-RPC générique ne sont pas publics. À partir de `pre.006`, le moteur générique de subscription typed existe dans la crate mais sa création reste `pub(crate)` jusqu'aux wrappers standard publics des tranches `pre.009+` ; il ne constitue donc toujours pas une escape hatch provider-specific. +Le socket brut et la primitive JSON-RPC générique ne sont pas publics. Le moteur générique de subscription typed existe depuis `pre.006` mais sa création reste `pub(crate)` même après l'ouverture des premiers wrappers standards en `pre.009`; il ne constitue donc pas une escape hatch provider-specific. -`WsSubscription` est déjà le handle public commun que ces wrappers retourneront. Il porte un `WsSubscriptionId` local stable, jamais le remote ID numérique du serveur. Les notifications arrivent via un receiver typed borné et `unsubscribe().await` exécute le `*Unsubscribe` correspondant en préservant son résultat booléen. +`WsSubscription` est le handle public commun retourné par les wrappers standards. Il porte un `WsSubscriptionId` local stable, jamais le remote ID numérique du serveur. Les notifications arrivent via un receiver typed borné et `unsubscribe().await` exécute le `*Unsubscribe` correspondant en préservant son résultat booléen. Le snapshot de session expose les subscriptions actuellement enregistrées via `WsSubscriptionSnapshot`, avec `remote_bound: bool` seulement. Le remote ID réel n'est jamais projeté. +### Premiers wrappers standards publics + +Depuis `0.2.7-pre.009`, trois familles stables peuvent être créées directement sur la session : + +```rust +let account = match "11111111111111111111111111111111".parse::() { + Ok(value) => value, + Err(error) => return Err(error.into()), +}; +let account_config = ksp_onchain_transport_lib::SolanaAccountSubscribeConfig::new( + Some(ksp_onchain_transport_lib::SolanaAccountEncoding::Base64), + None, + Some(ksp_onchain_transport_lib::SolanaCommitment::Confirmed), +); +let mut account_subscription = match session.account_subscribe(&account, Some(&account_config)).await { + Ok(value) => value, + Err(error) => return Err(error), +}; +let next = account_subscription.recv().await; +let removed = account_subscription.unsubscribe().await; +``` + +La même session expose `program_subscribe()` avec `SolanaProgramSubscribeConfig` et `logs_subscribe()` avec `SolanaLogsSubscribeFilter` plus `SolanaCommitmentConfig`. Pour `logsSubscribe`, `Mentions(pubkey)` représente exactement une adresse, conformément à la contrainte upstream retenue par l'audit. + +`accountSubscribe` ne propose pas `minContextSlot`: le champ existe dans un config partagé upstream mais est ignoré par le handler PubSub audité. `programSubscribe` conserve en revanche `withContext`; `SolanaProgramNotification` permet au consumer de traiter explicitement une notification contextée ou non contextée. + +Les trois wrappers retournent le même handle `WsSubscription` : reconnect, resubscribe, overflow, cause terminale et unsubscribe restent donc uniformes. Aucun wrapper public n'accepte un nom de méthode JSON-RPC arbitraire ni un remote subscription ID. + ### Reconnect automatique borné Depuis `0.2.7-pre.007`, les settings de session contrôlent réellement le reconnect physique. Une perte de socket publie `Reconnecting { attempt }`, invalide les remote IDs et incrémente `continuity_gap_count`. Avec la policy par défaut `ActiveSubscriptions`, les handles logiques gardent leur `WsSubscriptionId` et passent temporairement en `Resubscribing`; l'actor recrée leurs subscriptions dans l'ordre local avant de republier `Active`. @@ -122,7 +150,7 @@ Le consumer doit traiter `overflow_count` et `continuity_gap_count` comme deux s ```rust let session = ksp_onchain_transport_lib::WsSession::connect(endpoint).await?; -// ... utilisation future des subscriptions typed ... +// ... account_subscribe/program_subscribe/logs_subscribe puis recv()/unsubscribe() ... session.close().await?; ``` diff --git a/crates/ksp-onchain-transport-lib/src/lib.rs b/crates/ksp-onchain-transport-lib/src/lib.rs index 345d9c9..c0e56a9 100644 --- a/crates/ksp-onchain-transport-lib/src/lib.rs +++ b/crates/ksp-onchain-transport-lib/src/lib.rs @@ -1,5 +1,5 @@ // file: crates/ksp-onchain-transport-lib/src/lib.rs -// version: 25 +// version: 26 #![warn(missing_docs)] #![deny(unreachable_pub)] @@ -23,6 +23,8 @@ //! request/frame/message limits and control-frame handling. `0.2.7-pre.006` adds the typed subscription registry with stable local IDs and internal remote-ID //! routing. `0.2.7-pre.007` adds finite reconnect, deterministic resubscribe and continuity-gap tracking. `0.2.7-pre.008` makes per-subscription notification //! backpressure terminal and observable, preserves safe terminal error codes, performs best-effort remote cleanup and proves bounded capacity reuse. +//! `0.2.7-pre.009` opens the first stable typed WebSocket wrappers for account, program-account and transaction-log subscriptions without exposing a raw +//! provider-extension subscription API. mod client; mod constants; @@ -41,10 +43,12 @@ mod rpc_method; mod rpc_tokens; mod rpc_transactions; mod settings; +mod ws_accounts; mod ws_lifecycle; mod ws_session; mod ws_settings; mod ws_subscription; +mod ws_transactions; /// Passive runtime availability reported for one logical HTTP endpoint. pub use self::client::HttpEndpointAvailability; @@ -122,9 +126,9 @@ pub use self::resilience::evaluate_transport_retry; pub use self::rpc_accounts::SolanaAccount; /// Address and lamport balance returned by `getLargestAccounts`. pub use self::rpc_accounts::SolanaAccountBalance; -/// Wire-preserving account data returned by Solana HTTP account methods. +/// Wire-preserving account data returned by Solana account HTTP and WebSocket methods. pub use self::rpc_accounts::SolanaAccountData; -/// Account-data encoding accepted by Solana HTTP account methods. +/// Account-data encoding accepted by Solana account HTTP and WebSocket methods. pub use self::rpc_accounts::SolanaAccountEncoding; /// Shared account configuration used by account-info and token-account list methods. pub use self::rpc_accounts::SolanaAccountInfoConfig; @@ -202,15 +206,15 @@ pub use self::rpc_cluster::SolanaVoteAccountInfo; pub use self::rpc_cluster::SolanaVoteAccountStatus; /// Configuration accepted by `getVoteAccounts`. pub use self::rpc_cluster::SolanaVoteAccountsConfig; -/// Commitment level accepted by typed Solana HTTP RPC adapters. +/// Commitment level accepted by typed Solana HTTP and WebSocket adapters. pub use self::rpc_common::SolanaCommitment; -/// Optional commitment-only configuration shared by typed Solana HTTP RPC methods. +/// Optional commitment-only configuration shared by typed Solana RPC methods. pub use self::rpc_common::SolanaCommitmentConfig; /// Optional commitment and minimum-context configuration shared by typed Solana HTTP RPC methods. pub use self::rpc_common::SolanaContextConfig; -/// Typed Solana RPC context shared by contextual HTTP responses. +/// Typed Solana RPC context shared by contextual HTTP and WebSocket responses. pub use self::rpc_common::SolanaRpcContext; -/// Generic contextual result returned by typed Solana HTTP RPC adapters. +/// Generic contextual result returned by typed Solana HTTP and WebSocket adapters. pub use self::rpc_common::SolanaRpcResponse; /// Inflation-governor values returned by `getInflationGovernor`. pub use self::rpc_economics::SolanaInflationGovernor; @@ -310,6 +314,12 @@ pub use self::settings::HttpRoleLimits; pub use self::settings::HttpRoleName; /// Complete runtime settings consumed by the Solana HTTP transport foundation. pub use self::settings::HttpTransportSettings; +/// Configuration accepted by the standard Solana `accountSubscribe` WebSocket method. +pub use self::ws_accounts::SolanaAccountSubscribeConfig; +/// One `programNotification` payload preserving contextual and non-contextual upstream forms. +pub use self::ws_accounts::SolanaProgramNotification; +/// Configuration accepted by the standard Solana `programSubscribe` WebSocket method. +pub use self::ws_accounts::SolanaProgramSubscribeConfig; /// Stable local identity assigned to one physical WebSocket session. pub use self::ws_lifecycle::WsSessionId; /// Safe runtime snapshot for one physical WebSocket session. @@ -346,6 +356,10 @@ pub use self::ws_settings::WsSessionSettings; pub use self::ws_settings::WsTransportSettings; /// Typed handle for one logical Solana WebSocket subscription. pub use self::ws_subscription::WsSubscription; +/// Typed value carried by a contextual Solana `logsNotification`. +pub use self::ws_transactions::SolanaLogsNotification; +/// Filter accepted by the standard Solana `logsSubscribe` WebSocket method. +pub use self::ws_transactions::SolanaLogsSubscribeFilter; /// Owning tracing target for events emitted by the on-chain transport crate. pub(crate) use self::constants::TRACING_TARGET; diff --git a/crates/ksp-onchain-transport-lib/src/rpc_accounts.rs b/crates/ksp-onchain-transport-lib/src/rpc_accounts.rs index b441f13..5d80ff3 100644 --- a/crates/ksp-onchain-transport-lib/src/rpc_accounts.rs +++ b/crates/ksp-onchain-transport-lib/src/rpc_accounts.rs @@ -1,11 +1,11 @@ // file: crates/ksp-onchain-transport-lib/src/rpc_accounts.rs -// version: 6 +// version: 7 const MAX_MEMCMP_BYTES: usize = 128; const MAX_MULTIPLE_ACCOUNTS: usize = 100; const MAX_PROGRAM_ACCOUNT_FILTERS: usize = 4; -/// Account-data encoding accepted by Solana HTTP account methods. +/// Account-data encoding accepted by Solana account HTTP and WebSocket methods. #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] pub enum SolanaAccountEncoding { /// Legacy binary/base58 request encoding. @@ -71,7 +71,8 @@ impl SolanaDataSliceConfig { return self.length; } - fn to_json_value(self) -> serde_json::Value { + /// Serializes this data-slice configuration to the Solana JSON-RPC wire object. + pub(crate) fn to_json_value(self) -> serde_json::Value { return serde_json::json!({"offset": self.offset, "length": self.length}); } } @@ -277,7 +278,8 @@ pub enum SolanaProgramAccountFilter { } impl SolanaProgramAccountFilter { - fn to_json_value(&self) -> serde_json::Value { + /// Serializes this program-account filter to the Solana JSON-RPC wire representation. + pub(crate) fn to_json_value(&self) -> serde_json::Value { return match self { Self::DataSize(size) => serde_json::json!({"dataSize": size}), Self::Memcmp(filter) => serde_json::json!({"memcmp": filter.to_json_value()}), @@ -385,7 +387,7 @@ impl SolanaParsedAccountData { } } -/// Wire-preserving account data returned by Solana HTTP account methods. +/// Wire-preserving account data returned by Solana account HTTP and WebSocket methods. #[derive(Clone, Debug, PartialEq)] pub enum SolanaAccountData { /// Legacy single-string binary form retained for backwards compatibility. diff --git a/crates/ksp-onchain-transport-lib/src/rpc_common.rs b/crates/ksp-onchain-transport-lib/src/rpc_common.rs index 5a88879..affb13b 100644 --- a/crates/ksp-onchain-transport-lib/src/rpc_common.rs +++ b/crates/ksp-onchain-transport-lib/src/rpc_common.rs @@ -1,7 +1,7 @@ // file: crates/ksp-onchain-transport-lib/src/rpc_common.rs -// version: 5 +// version: 6 -/// Commitment level accepted by typed Solana HTTP RPC adapters. +/// Commitment level accepted by typed Solana HTTP and WebSocket adapters. #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] pub enum SolanaCommitment { /// Query the most recent processed bank. @@ -24,7 +24,7 @@ impl SolanaCommitment { } } -/// Optional commitment-only configuration shared by typed Solana HTTP RPC methods. +/// Optional commitment-only configuration shared by typed Solana RPC methods. #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] pub struct SolanaCommitmentConfig { commitment: std::option::Option, @@ -94,7 +94,7 @@ impl SolanaContextConfig { } } -/// Typed Solana RPC context shared by contextual HTTP responses. +/// Typed Solana RPC context shared by contextual HTTP and WebSocket responses. #[derive(Clone, Debug, Eq, PartialEq)] pub struct SolanaRpcContext { slot: u64, @@ -128,7 +128,7 @@ impl SolanaRpcContext { } } -/// Generic contextual result returned by typed Solana HTTP RPC adapters. +/// Generic contextual result returned by typed Solana HTTP and WebSocket adapters. #[derive(Clone, Debug, PartialEq)] pub struct SolanaRpcResponse { context: crate::SolanaRpcContext, @@ -168,7 +168,7 @@ pub(crate) fn decode_wire_json(method: &str, val return match decoded { std::result::Result::Ok(decoded) => std::result::Result::Ok(decoded), std::result::Result::Err(error) => std::result::Result::Err( - ksp_core_lib::Error::new(crate::ERROR_CODE_INVALID_RESPONSE, "typed Solana HTTP response has an invalid wire shape") + ksp_core_lib::Error::new(crate::ERROR_CODE_INVALID_RESPONSE, "typed Solana RPC response has an invalid wire shape") .with_context("rpc_method", method) .with_source(error), ), @@ -181,7 +181,7 @@ pub(crate) fn parse_wire_pubkey(method: &str, field: &str, value: &str) -> ksp_c return match parsed { std::result::Result::Ok(pubkey) => std::result::Result::Ok(pubkey), std::result::Result::Err(_) => std::result::Result::Err( - ksp_core_lib::Error::new(crate::ERROR_CODE_INVALID_RESPONSE, "typed Solana HTTP response contains an invalid public key") + ksp_core_lib::Error::new(crate::ERROR_CODE_INVALID_RESPONSE, "typed Solana RPC response contains an invalid public key") .with_context("rpc_method", method) .with_context("field", field), ), diff --git a/crates/ksp-onchain-transport-lib/src/ws_accounts.rs b/crates/ksp-onchain-transport-lib/src/ws_accounts.rs new file mode 100644 index 0000000..4d01057 --- /dev/null +++ b/crates/ksp-onchain-transport-lib/src/ws_accounts.rs @@ -0,0 +1,278 @@ +// file: crates/ksp-onchain-transport-lib/src/ws_accounts.rs +// version: 1 + +const MAX_PROGRAM_SUBSCRIBE_FILTERS: usize = 4; +const MAX_PROGRAM_SUBSCRIBE_RAW_MEMCMP_BYTES: usize = 128; + +/// Configuration accepted by the standard Solana `accountSubscribe` WebSocket method. +/// +/// `minContextSlot` is deliberately absent: Agave `v4.2.1` carries that field in the shared account config but the PubSub handler ignores it, so KSP does +/// not expose it as an effective WebSocket option. +#[derive(Clone, Debug, Default, Eq, PartialEq)] +pub struct SolanaAccountSubscribeConfig { + encoding: std::option::Option, + data_slice: std::option::Option, + commitment: std::option::Option, +} + +impl SolanaAccountSubscribeConfig { + /// Creates an explicit `accountSubscribe` configuration. + #[must_use] + pub const fn new( + encoding: std::option::Option, + data_slice: std::option::Option, + commitment: std::option::Option, + ) -> Self { + return Self { encoding, data_slice, commitment }; + } + + /// Returns the optional account-data encoding. + #[must_use] + pub const fn encoding(&self) -> std::option::Option { + return self.encoding; + } + + /// Returns the optional account-data slice. + #[must_use] + pub const fn data_slice(&self) -> std::option::Option { + return self.data_slice; + } + + /// Returns the optional commitment level. + #[must_use] + pub const fn commitment(&self) -> std::option::Option { + return self.commitment; + } + + fn is_empty(&self) -> bool { + return self.encoding.is_none() && self.data_slice.is_none() && self.commitment.is_none(); + } + + fn to_json_value(&self) -> serde_json::Value { + let mut object = serde_json::Map::new(); + if let std::option::Option::Some(encoding) = self.encoding { + object.insert("encoding".to_owned(), serde_json::Value::String(encoding.as_str().to_owned())); + } + if let std::option::Option::Some(data_slice) = self.data_slice { + object.insert("dataSlice".to_owned(), data_slice.to_json_value()); + } + if let std::option::Option::Some(commitment) = self.commitment { + object.insert("commitment".to_owned(), serde_json::Value::String(commitment.as_str().to_owned())); + } + return serde_json::Value::Object(object); + } +} + +/// Configuration accepted by the standard Solana `programSubscribe` WebSocket method. +#[derive(Clone, Debug, Default, Eq, PartialEq)] +pub struct SolanaProgramSubscribeConfig { + account: crate::SolanaAccountSubscribeConfig, + filters: std::vec::Vec, + with_context: std::option::Option, +} + +impl SolanaProgramSubscribeConfig { + /// Creates an explicit `programSubscribe` configuration. + #[must_use] + pub fn new( + account: crate::SolanaAccountSubscribeConfig, + filters: std::vec::Vec, + with_context: std::option::Option, + ) -> Self { + return Self { account, filters, with_context }; + } + + /// Returns the shared WebSocket account configuration. + #[must_use] + pub const fn account(&self) -> &crate::SolanaAccountSubscribeConfig { + return &self.account; + } + + /// Returns the ordered program-account filters. + #[must_use] + pub fn filters(&self) -> &[crate::SolanaProgramAccountFilter] { + return self.filters.as_slice(); + } + + /// Returns the optional `withContext` request; omission uses the upstream default `false`. + #[must_use] + pub const fn with_context(&self) -> std::option::Option { + return self.with_context; + } + + fn is_empty(&self) -> bool { + return self.account.is_empty() && self.filters.is_empty() && self.with_context.is_none(); + } + + fn to_json_value(&self) -> serde_json::Value { + let account = self.account.to_json_value(); + let mut object = match account { + serde_json::Value::Object(object) => object, + _ => serde_json::Map::new(), + }; + if !self.filters.is_empty() { + let filters = self.filters.iter().map(crate::SolanaProgramAccountFilter::to_json_value).collect::>(); + object.insert("filters".to_owned(), serde_json::Value::Array(filters)); + } + if let std::option::Option::Some(with_context) = self.with_context { + object.insert("withContext".to_owned(), serde_json::Value::Bool(with_context)); + } + return serde_json::Value::Object(object); + } +} + +/// One `programNotification` payload, preserving whether the upstream wire result was contextualized. +#[derive(Clone, Debug, PartialEq)] +pub enum SolanaProgramNotification { + /// Program account payload without a surrounding RPC context. + Account(crate::SolanaKeyedAccount), + /// Program account payload wrapped in an RPC context. + Context(crate::SolanaRpcResponse), +} + +impl SolanaProgramNotification { + /// Returns the program account regardless of the upstream context-wrapper form. + #[must_use] + pub const fn account(&self) -> &crate::SolanaKeyedAccount { + return match self { + Self::Account(account) => account, + Self::Context(response) => response.value(), + }; + } + + /// Returns the RPC context when the upstream notification included one. + #[must_use] + pub const fn context(&self) -> std::option::Option<&crate::SolanaRpcContext> { + return match self { + Self::Account(_) => std::option::Option::None, + Self::Context(response) => std::option::Option::Some(response.context()), + }; + } +} + +impl crate::WsSession { + /// Subscribes to changes for one Solana account through standard `accountSubscribe`. + pub async fn account_subscribe( + &self, + account: &ksp_core_lib::Pubkey, + config: std::option::Option<&crate::SolanaAccountSubscribeConfig>, + ) -> ksp_core_lib::Result>> { + let mut params = std::vec![serde_json::Value::String(account.to_string())]; + if let std::option::Option::Some(config) = config + && !config.is_empty() + { + params.push(config.to_json_value()); + } + return self.subscribe_typed(crate::WsSubscriptionKind::Account, params, |value| decode_account_notification("accountSubscribe", value)).await; + } + + /// Subscribes to account changes owned by one Solana program through standard `programSubscribe`. + pub async fn program_subscribe( + &self, + program_id: &ksp_core_lib::Pubkey, + config: std::option::Option<&crate::SolanaProgramSubscribeConfig>, + ) -> ksp_core_lib::Result> { + if let std::option::Option::Some(config) = config { + let validation = validate_program_subscribe_filters(config.filters()); + if let std::result::Result::Err(error) = validation { + return std::result::Result::Err(error); + } + } + let mut params = std::vec![serde_json::Value::String(program_id.to_string())]; + if let std::option::Option::Some(config) = config + && !config.is_empty() + { + params.push(config.to_json_value()); + } + return self.subscribe_typed(crate::WsSubscriptionKind::Program, params, |value| decode_program_notification("programSubscribe", value)).await; + } +} + +#[derive(serde::Deserialize)] +struct WireRpcResponse { + context: serde_json::Value, + value: serde_json::Value, +} + +#[derive(serde::Deserialize)] +#[serde(untagged)] +enum WireProgramNotification { + Context(WireRpcResponse), + Account(serde_json::Value), +} + +fn decode_account_notification(method: &str, value: serde_json::Value) -> ksp_core_lib::Result> { + let decoded = crate::decode_wire_json::(method, value); + let wire = match decoded { + std::result::Result::Ok(wire) => wire, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let context = crate::SolanaRpcContext::decode_wire(method, wire.context); + let context = match context { + std::result::Result::Ok(context) => context, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let account = crate::SolanaAccount::decode_wire(method, wire.value); + let account = match account { + std::result::Result::Ok(account) => account, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + return std::result::Result::Ok(crate::SolanaRpcResponse::new(context, account)); +} + +fn decode_program_notification(method: &str, value: serde_json::Value) -> ksp_core_lib::Result { + let decoded = crate::decode_wire_json::(method, value); + let wire = match decoded { + std::result::Result::Ok(wire) => wire, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + return match wire { + WireProgramNotification::Account(value) => { + let account = crate::SolanaKeyedAccount::decode_wire(method, value); + match account { + std::result::Result::Ok(account) => std::result::Result::Ok(crate::SolanaProgramNotification::Account(account)), + std::result::Result::Err(error) => std::result::Result::Err(error), + } + }, + WireProgramNotification::Context(wire) => { + let context = crate::SolanaRpcContext::decode_wire(method, wire.context); + let context = match context { + std::result::Result::Ok(context) => context, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let account = crate::SolanaKeyedAccount::decode_wire(method, wire.value); + let account = match account { + std::result::Result::Ok(account) => account, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + std::result::Result::Ok(crate::SolanaProgramNotification::Context(crate::SolanaRpcResponse::new(context, account))) + }, + }; +} + +fn validate_program_subscribe_filters(filters: &[crate::SolanaProgramAccountFilter]) -> ksp_core_lib::Result<()> { + if filters.len() > MAX_PROGRAM_SUBSCRIBE_FILTERS { + return std::result::Result::Err( + ksp_core_lib::Error::new(crate::ERROR_CODE_INVALID_RPC_PARAMETERS, "programSubscribe accepts at most 4 filters on the targeted Agave runtime") + .with_context("rpc_method", "programSubscribe") + .with_context("filter_count", filters.len().to_string()), + ); + } + for filter in filters { + if let crate::SolanaProgramAccountFilter::Memcmp(memcmp) = filter + && let crate::SolanaMemcmpBytes::Bytes(bytes) = memcmp.bytes() + && bytes.len() > MAX_PROGRAM_SUBSCRIBE_RAW_MEMCMP_BYTES + { + return std::result::Result::Err( + ksp_core_lib::Error::new(crate::ERROR_CODE_INVALID_RPC_PARAMETERS, "raw programSubscribe memcmp data accepts at most 128 bytes") + .with_context("rpc_method", "programSubscribe") + .with_context("memcmp_byte_count", bytes.len().to_string()), + ); + } + } + return std::result::Result::Ok(()); +} + +#[cfg(test)] +#[path = "../unit_tests/ws_accounts.rs"] +mod tests; diff --git a/crates/ksp-onchain-transport-lib/src/ws_session.rs b/crates/ksp-onchain-transport-lib/src/ws_session.rs index 3b34b58..074ef2c 100644 --- a/crates/ksp-onchain-transport-lib/src/ws_session.rs +++ b/crates/ksp-onchain-transport-lib/src/ws_session.rs @@ -1,5 +1,5 @@ // file: crates/ksp-onchain-transport-lib/src/ws_session.rs -// version: 9 +// version: 10 use futures_util::SinkExt; // rust-rules: trait-import use futures_util::StreamExt; // rust-rules: trait-import @@ -156,9 +156,8 @@ impl WsSession { /// Creates one crate-internal typed standard Solana subscription through the actor-owned registry. /// - /// Public typed wrappers are introduced in later prereleases. Keeping this constructor crate-private prevents a raw provider-extension subscription API - /// from becoming part of the stable KSP surface while still making the generic typed engine testable and reusable by those wrappers. - #[allow(dead_code)] // Staged in pre.006 and consumed by the public typed standard wrappers starting in pre.009. + /// Public typed standard wrappers consume this constructor while it remains crate-private, preventing a raw provider-extension subscription API from + /// becoming part of the stable KSP surface. pub(crate) async fn subscribe_typed( &self, kind: crate::WsSubscriptionKind, diff --git a/crates/ksp-onchain-transport-lib/src/ws_transactions.rs b/crates/ksp-onchain-transport-lib/src/ws_transactions.rs new file mode 100644 index 0000000..4f227c3 --- /dev/null +++ b/crates/ksp-onchain-transport-lib/src/ws_transactions.rs @@ -0,0 +1,104 @@ +// file: crates/ksp-onchain-transport-lib/src/ws_transactions.rs +// version: 1 + +/// Filter accepted by the standard Solana `logsSubscribe` WebSocket method. +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum SolanaLogsSubscribeFilter { + /// Subscribe to all transactions except simple vote transactions. + All, + /// Subscribe to all transactions including simple vote transactions. + AllWithVotes, + /// Subscribe only to transactions mentioning exactly one public key. + Mentions(ksp_core_lib::Pubkey), +} + +impl SolanaLogsSubscribeFilter { + fn to_json_value(&self) -> serde_json::Value { + return match self { + Self::All => serde_json::Value::String("all".to_owned()), + Self::AllWithVotes => serde_json::Value::String("allWithVotes".to_owned()), + Self::Mentions(pubkey) => serde_json::json!({"mentions": [pubkey.to_string()]}), + }; + } +} + +/// Typed value carried by a contextual Solana `logsNotification`. +#[derive(Clone, Debug, PartialEq)] +pub struct SolanaLogsNotification { + signature: std::string::String, + err: std::option::Option, + logs: std::vec::Vec, +} + +impl SolanaLogsNotification { + /// Returns the base58 transaction signature exactly as reported by the RPC node. + #[must_use] + pub fn signature(&self) -> &str { + return self.signature.as_str(); + } + + /// Returns the nullable transaction-error wire value without interpreting Program/runtime error semantics. + #[must_use] + pub const fn err(&self) -> std::option::Option<&serde_json::Value> { + return self.err.as_ref(); + } + + /// Returns the ordered transaction log messages. + #[must_use] + pub fn logs(&self) -> &[std::string::String] { + return self.logs.as_slice(); + } +} + +impl crate::WsSession { + /// Subscribes to Solana transaction logs through standard `logsSubscribe`. + pub async fn logs_subscribe( + &self, + filter: &crate::SolanaLogsSubscribeFilter, + config: std::option::Option<&crate::SolanaCommitmentConfig>, + ) -> ksp_core_lib::Result>> { + let mut params = std::vec![filter.to_json_value()]; + if let std::option::Option::Some(config) = config + && config.commitment().is_some() + { + params.push(config.to_json_value()); + } + return self.subscribe_typed(crate::WsSubscriptionKind::Logs, params, |value| decode_logs_notification("logsSubscribe", value)).await; + } +} + +#[derive(serde::Deserialize)] +struct WireRpcResponse { + context: serde_json::Value, + value: WireLogsNotification, +} + +#[derive(serde::Deserialize)] +struct WireLogsNotification { + signature: std::string::String, + err: serde_json::Value, + logs: std::vec::Vec, +} + +fn decode_logs_notification(method: &str, value: serde_json::Value) -> ksp_core_lib::Result> { + let decoded = crate::decode_wire_json::(method, value); + let wire = match decoded { + std::result::Result::Ok(wire) => wire, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let context = crate::SolanaRpcContext::decode_wire(method, wire.context); + let context = match context { + std::result::Result::Ok(context) => context, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let err = match wire.value.err { + serde_json::Value::Null => std::option::Option::None, + value => std::option::Option::Some(value), + }; + let notification = crate::SolanaLogsNotification { signature: wire.value.signature, err, logs: wire.value.logs }; + return std::result::Result::Ok(crate::SolanaRpcResponse::new(context, notification)); +} + +#[cfg(test)] +#[path = "../unit_tests/ws_transactions.rs"] +mod tests; diff --git a/crates/ksp-onchain-transport-lib/tests/public_api.rs b/crates/ksp-onchain-transport-lib/tests/public_api.rs index d6d65b8..9ddb57e 100644 --- a/crates/ksp-onchain-transport-lib/tests/public_api.rs +++ b/crates/ksp-onchain-transport-lib/tests/public_api.rs @@ -1,5 +1,5 @@ // file: crates/ksp-onchain-transport-lib/tests/public_api.rs -// version: 29 +// version: 30 //! Integration tests for the public `ksp-onchain-transport-lib` consumer contract. @@ -585,3 +585,28 @@ fn public_v0_2_7_pre_008_backpressure_observability_contract_is_available_from_c let _overflow_count = ksp_onchain_transport_lib::WsSessionSnapshot::overflow_count; assert_eq!(ksp_onchain_transport_lib::ERROR_CODE_WS_BACKPRESSURE_OVERFLOW.code(), "ws_backpressure_overflow"); } + +#[test] +fn public_v0_2_7_pre_009_stable_websocket_lot_a_wrappers_and_dtos_are_available_from_crate_root() { + let _account_subscribe = ksp_onchain_transport_lib::WsSession::account_subscribe; + let _program_subscribe = ksp_onchain_transport_lib::WsSession::program_subscribe; + let _logs_subscribe = ksp_onchain_transport_lib::WsSession::logs_subscribe; + let account_config = ksp_onchain_transport_lib::SolanaAccountSubscribeConfig::new( + std::option::Option::Some(ksp_onchain_transport_lib::SolanaAccountEncoding::Base64), + std::option::Option::Some(ksp_onchain_transport_lib::SolanaDataSliceConfig::new(0, 32)), + std::option::Option::Some(ksp_onchain_transport_lib::SolanaCommitment::Confirmed), + ); + assert_eq!(account_config.commitment(), std::option::Option::Some(ksp_onchain_transport_lib::SolanaCommitment::Confirmed)); + let program_config = ksp_onchain_transport_lib::SolanaProgramSubscribeConfig::new( + account_config, + std::vec![ksp_onchain_transport_lib::SolanaProgramAccountFilter::DataSize(80)], + std::option::Option::Some(true), + ); + assert_eq!(program_config.with_context(), std::option::Option::Some(true)); + let mention = "11111111111111111111111111111111".parse::().expect("public logs mention fixture must parse"); + let filter = ksp_onchain_transport_lib::SolanaLogsSubscribeFilter::Mentions(mention); + let filter_name = std::any::type_name_of_val(&filter); + assert!(filter_name.ends_with("SolanaLogsSubscribeFilter")); + let _program_notification = std::any::type_name::(); + let _logs_notification = std::any::type_name::(); +} diff --git a/crates/ksp-onchain-transport-lib/unit_tests/ws_accounts.rs b/crates/ksp-onchain-transport-lib/unit_tests/ws_accounts.rs new file mode 100644 index 0000000..4330cae --- /dev/null +++ b/crates/ksp-onchain-transport-lib/unit_tests/ws_accounts.rs @@ -0,0 +1,202 @@ +// file: crates/ksp-onchain-transport-lib/unit_tests/ws_accounts.rs +// version: 1 + +use futures_util::SinkExt; // rust-rules: trait-import +use futures_util::StreamExt; // rust-rules: trait-import + +fn local_endpoint(url: &str) -> crate::WsEndpointSettings { + return crate::WsEndpointSettings::new( + "local_ws_accounts", + true, + crate::WsProviderName::new("local-fixture"), + crate::WsClusterName::new("local"), + crate::WsProtocolKind::SolanaStandard, + crate::WsEndpointUrl::parse(url).expect("local test WebSocket URL must parse"), + crate::WsSessionSettings::default(), + ); +} + +async fn bind_local_listener() -> (tokio::net::TcpListener, std::string::String) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.expect("local listener must bind"); + let address = listener.local_addr().expect("local listener must expose address"); + return (listener, format!("ws://{address}")); +} + +async fn read_request(websocket: &mut tokio_tungstenite::WebSocketStream) -> serde_json::Value { + let message = websocket.next().await.expect("request message must exist").expect("request message must decode"); + let text = message.to_text().expect("request must be text"); + return serde_json::from_str(text).expect("request must contain JSON"); +} + +async fn send_result(websocket: &mut tokio_tungstenite::WebSocketStream, request: &serde_json::Value, result: serde_json::Value) { + let id = request.get("id").and_then(serde_json::Value::as_u64).expect("request id must be numeric"); + let response = serde_json::json!({"jsonrpc":"2.0","id":id,"result":result}); + websocket.send(tokio_tungstenite::tungstenite::Message::Text(response.to_string().into())).await.expect("local response must send"); +} + +async fn send_notification(websocket: &mut tokio_tungstenite::WebSocketStream, method: &str, remote_id: u64, result: serde_json::Value) { + let notification = serde_json::json!({"jsonrpc":"2.0","method":method,"params":{"result":result,"subscription":remote_id}}); + websocket.send(tokio_tungstenite::tungstenite::Message::Text(notification.to_string().into())).await.expect("local notification must send"); +} + +async fn wait_for_close_frame(websocket: &mut tokio_tungstenite::WebSocketStream) { + loop { + let message = websocket.next().await; + match message { + std::option::Option::Some(std::result::Result::Ok(tokio_tungstenite::tungstenite::Message::Close(_))) => return, + std::option::Option::Some(std::result::Result::Ok(_)) => {}, + std::option::Option::Some(std::result::Result::Err(_)) | std::option::Option::None => return, + } + } +} + +fn account_wire(lamports: u64) -> serde_json::Value { + return serde_json::json!({ + "lamports": lamports, + "data": ["AQID", "base64"], + "owner": "11111111111111111111111111111111", + "executable": false, + "rentEpoch": 7, + "space": 3 + }); +} + +#[test] +fn account_subscribe_config_preserves_effective_websocket_options_without_min_context_slot() { + let config = crate::SolanaAccountSubscribeConfig::new( + std::option::Option::Some(crate::SolanaAccountEncoding::Base64Zstd), + std::option::Option::Some(crate::SolanaDataSliceConfig::new(4, 16)), + std::option::Option::Some(crate::SolanaCommitment::Confirmed), + ); + assert_eq!(config.encoding(), std::option::Option::Some(crate::SolanaAccountEncoding::Base64Zstd)); + assert_eq!(config.data_slice(), std::option::Option::Some(crate::SolanaDataSliceConfig::new(4, 16))); + assert_eq!(config.commitment(), std::option::Option::Some(crate::SolanaCommitment::Confirmed)); + assert_eq!(config.to_json_value(), serde_json::json!({"encoding":"base64+zstd","dataSlice":{"offset":4,"length":16},"commitment":"confirmed"})); + assert!(config.to_json_value().get("minContextSlot").is_none()); +} + +#[test] +fn program_subscribe_config_preserves_filters_with_context_and_deterministic_bounds() { + let config = crate::SolanaProgramSubscribeConfig::new( + crate::SolanaAccountSubscribeConfig::new( + std::option::Option::Some(crate::SolanaAccountEncoding::Base64), + std::option::Option::None, + std::option::Option::Some(crate::SolanaCommitment::Finalized), + ), + std::vec![ + crate::SolanaProgramAccountFilter::DataSize(80), + crate::SolanaProgramAccountFilter::Memcmp(crate::SolanaMemcmpFilter::new(4, crate::SolanaMemcmpBytes::Bytes(std::vec![1, 2, 3]))), + crate::SolanaProgramAccountFilter::TokenAccountState, + ], + std::option::Option::Some(true), + ); + assert_eq!(config.with_context(), std::option::Option::Some(true)); + assert_eq!(config.filters().len(), 3); + assert_eq!( + config.to_json_value(), + serde_json::json!({ + "encoding":"base64", + "commitment":"finalized", + "filters":[{"dataSize":80},{"memcmp":{"offset":4,"bytes":[1,2,3],"encoding":"bytes"}},"tokenAccountState"], + "withContext":true + }) + ); + let too_many = std::vec![ + crate::SolanaProgramAccountFilter::DataSize(1), + crate::SolanaProgramAccountFilter::DataSize(2), + crate::SolanaProgramAccountFilter::DataSize(3), + crate::SolanaProgramAccountFilter::DataSize(4), + crate::SolanaProgramAccountFilter::DataSize(5), + ]; + let error = super::validate_program_subscribe_filters(too_many.as_slice()).expect_err("five programSubscribe filters must reject locally"); + assert_eq!(error.code(), crate::ERROR_CODE_INVALID_RPC_PARAMETERS); + let oversized = + std::vec![crate::SolanaProgramAccountFilter::Memcmp(crate::SolanaMemcmpFilter::new(0, crate::SolanaMemcmpBytes::Bytes(std::vec![0; 129]),))]; + let error = super::validate_program_subscribe_filters(oversized.as_slice()).expect_err("oversized raw memcmp bytes must reject locally"); + assert_eq!(error.code(), crate::ERROR_CODE_INVALID_RPC_PARAMETERS); +} + +#[test] +fn program_notification_decoder_accepts_contextual_and_non_contextual_wire_forms() { + let keyed = serde_json::json!({"pubkey":"11111111111111111111111111111111","account":account_wire(42)}); + let bare = super::decode_program_notification("programSubscribe", keyed.clone()).expect("bare program notification must decode"); + assert!(bare.context().is_none()); + assert_eq!(bare.account().account().lamports(), 42); + let contextual = super::decode_program_notification("programSubscribe", serde_json::json!({"context":{"slot":99,"apiVersion":"4.2.1"},"value":keyed})) + .expect("contextual program notification must decode"); + assert_eq!(contextual.context().expect("context must be retained").slot(), 99); + assert_eq!(contextual.account().account().lamports(), 42); +} + +#[tokio::test(flavor = "current_thread")] +async fn stable_account_and_program_wrappers_use_exact_methods_decode_notifications_and_unsubscribe_by_handle() { + let (listener, url) = bind_local_listener().await; + let server = tokio::spawn(async move { + let (stream, _) = listener.accept().await.expect("local server must accept client"); + let mut websocket = tokio_tungstenite::accept_async(stream).await.expect("local WebSocket handshake must succeed"); + let account_request = read_request(&mut websocket).await; + assert_eq!(account_request["method"], serde_json::json!("accountSubscribe")); + assert_eq!( + account_request["params"], + serde_json::json!(["11111111111111111111111111111111", {"encoding":"base64","dataSlice":{"offset":1,"length":2},"commitment":"confirmed"}]) + ); + send_result(&mut websocket, &account_request, serde_json::json!(51)).await; + send_notification(&mut websocket, "accountNotification", 51, serde_json::json!({"context":{"slot":700},"value":account_wire(123)})).await; + let account_unsubscribe = read_request(&mut websocket).await; + assert_eq!(account_unsubscribe["method"], serde_json::json!("accountUnsubscribe")); + assert_eq!(account_unsubscribe["params"], serde_json::json!([51])); + send_result(&mut websocket, &account_unsubscribe, serde_json::json!(true)).await; + let program_request = read_request(&mut websocket).await; + assert_eq!(program_request["method"], serde_json::json!("programSubscribe")); + assert_eq!( + program_request["params"], + serde_json::json!(["11111111111111111111111111111111", {"encoding":"jsonParsed","filters":[{"dataSize":80}],"withContext":true}]) + ); + send_result(&mut websocket, &program_request, serde_json::json!(73)).await; + send_notification( + &mut websocket, + "programNotification", + 73, + serde_json::json!({ + "context":{"slot":701}, + "value":{"pubkey":"11111111111111111111111111111111","account":account_wire(456)} + }), + ) + .await; + let program_unsubscribe = read_request(&mut websocket).await; + assert_eq!(program_unsubscribe["method"], serde_json::json!("programUnsubscribe")); + assert_eq!(program_unsubscribe["params"], serde_json::json!([73])); + send_result(&mut websocket, &program_unsubscribe, serde_json::json!(true)).await; + wait_for_close_frame(&mut websocket).await; + }); + let session = crate::WsSession::connect(local_endpoint(url.as_str())).await.expect("client handshake must succeed"); + let account_pubkey = "11111111111111111111111111111111".parse::().expect("account pubkey fixture must parse"); + let account_config = crate::SolanaAccountSubscribeConfig::new( + std::option::Option::Some(crate::SolanaAccountEncoding::Base64), + std::option::Option::Some(crate::SolanaDataSliceConfig::new(1, 2)), + std::option::Option::Some(crate::SolanaCommitment::Confirmed), + ); + let mut account = session.account_subscribe(&account_pubkey, std::option::Option::Some(&account_config)).await.expect("accountSubscribe must register"); + assert_eq!(account.kind(), crate::WsSubscriptionKind::Account); + let notification = account.recv().await.expect("account notification must arrive").expect("account notification must decode"); + assert_eq!(notification.context().slot(), 700); + assert_eq!(notification.value().lamports(), 123); + assert!(account.unsubscribe().await.expect("account unsubscribe must complete")); + let program_config = crate::SolanaProgramSubscribeConfig::new( + crate::SolanaAccountSubscribeConfig::new( + std::option::Option::Some(crate::SolanaAccountEncoding::JsonParsed), + std::option::Option::None, + std::option::Option::None, + ), + std::vec![crate::SolanaProgramAccountFilter::DataSize(80)], + std::option::Option::Some(true), + ); + let mut program = session.program_subscribe(&account_pubkey, std::option::Option::Some(&program_config)).await.expect("programSubscribe must register"); + assert_eq!(program.kind(), crate::WsSubscriptionKind::Program); + let notification = program.recv().await.expect("program notification must arrive").expect("program notification must decode"); + assert_eq!(notification.context().expect("program context must be retained").slot(), 701); + assert_eq!(notification.account().account().lamports(), 456); + assert!(program.unsubscribe().await.expect("program unsubscribe must complete")); + session.close().await.expect("session close must complete"); + server.await.expect("local server task must complete"); +} diff --git a/crates/ksp-onchain-transport-lib/unit_tests/ws_transactions.rs b/crates/ksp-onchain-transport-lib/unit_tests/ws_transactions.rs new file mode 100644 index 0000000..1e32ee9 --- /dev/null +++ b/crates/ksp-onchain-transport-lib/unit_tests/ws_transactions.rs @@ -0,0 +1,122 @@ +// file: crates/ksp-onchain-transport-lib/unit_tests/ws_transactions.rs +// version: 1 + +use futures_util::SinkExt; // rust-rules: trait-import +use futures_util::StreamExt; // rust-rules: trait-import + +fn local_endpoint(url: &str) -> crate::WsEndpointSettings { + return crate::WsEndpointSettings::new( + "local_ws_transactions", + true, + crate::WsProviderName::new("local-fixture"), + crate::WsClusterName::new("local"), + crate::WsProtocolKind::SolanaStandard, + crate::WsEndpointUrl::parse(url).expect("local test WebSocket URL must parse"), + crate::WsSessionSettings::default(), + ); +} + +async fn bind_local_listener() -> (tokio::net::TcpListener, std::string::String) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.expect("local listener must bind"); + let address = listener.local_addr().expect("local listener must expose address"); + return (listener, format!("ws://{address}")); +} + +async fn read_request(websocket: &mut tokio_tungstenite::WebSocketStream) -> serde_json::Value { + let message = websocket.next().await.expect("request message must exist").expect("request message must decode"); + let text = message.to_text().expect("request must be text"); + return serde_json::from_str(text).expect("request must contain JSON"); +} + +async fn send_result(websocket: &mut tokio_tungstenite::WebSocketStream, request: &serde_json::Value, result: serde_json::Value) { + let id = request.get("id").and_then(serde_json::Value::as_u64).expect("request id must be numeric"); + let response = serde_json::json!({"jsonrpc":"2.0","id":id,"result":result}); + websocket.send(tokio_tungstenite::tungstenite::Message::Text(response.to_string().into())).await.expect("local response must send"); +} + +async fn wait_for_close_frame(websocket: &mut tokio_tungstenite::WebSocketStream) { + loop { + let message = websocket.next().await; + match message { + std::option::Option::Some(std::result::Result::Ok(tokio_tungstenite::tungstenite::Message::Close(_))) => return, + std::option::Option::Some(std::result::Result::Ok(_)) => {}, + std::option::Option::Some(std::result::Result::Err(_)) | std::option::Option::None => return, + } + } +} + +#[test] +fn logs_subscribe_filters_preserve_all_all_with_votes_and_exactly_one_mention() { + let pubkey = "11111111111111111111111111111111".parse::().expect("mention fixture must parse"); + assert_eq!(crate::SolanaLogsSubscribeFilter::All.to_json_value(), serde_json::json!("all")); + assert_eq!(crate::SolanaLogsSubscribeFilter::AllWithVotes.to_json_value(), serde_json::json!("allWithVotes")); + assert_eq!(crate::SolanaLogsSubscribeFilter::Mentions(pubkey).to_json_value(), serde_json::json!({"mentions":["11111111111111111111111111111111"]})); +} + +#[test] +fn logs_notification_decoder_preserves_context_signature_nullable_error_and_ordered_logs() { + let success = super::decode_logs_notification( + "logsSubscribe", + serde_json::json!({ + "context":{"slot":81,"apiVersion":"4.2.1"}, + "value":{"signature":"fixture-signature","err":null,"logs":["first","second"]} + }), + ) + .expect("successful logs notification must decode"); + assert_eq!(success.context().slot(), 81); + assert_eq!(success.value().signature(), "fixture-signature"); + assert!(success.value().err().is_none()); + assert_eq!(success.value().logs(), &["first".to_owned(), "second".to_owned()]); + let failed = super::decode_logs_notification( + "logsSubscribe", + serde_json::json!({"context":{"slot":82},"value":{"signature":"fixture-signature-2","err":{"InstructionError":[0,"Custom"]},"logs":[]}}), + ) + .expect("failed logs notification must preserve transaction error wire value"); + assert_eq!(failed.context().slot(), 82); + assert!(failed.value().err().is_some()); + let missing_err = + super::decode_logs_notification("logsSubscribe", serde_json::json!({"context":{"slot":83},"value":{"signature":"fixture-signature-3","logs":[]}})); + assert!(missing_err.is_err()); +} + +#[tokio::test(flavor = "current_thread")] +async fn stable_logs_wrapper_uses_exact_filter_config_notification_and_handle_unsubscribe() { + let (listener, url) = bind_local_listener().await; + let server = tokio::spawn(async move { + let (stream, _) = listener.accept().await.expect("local server must accept client"); + let mut websocket = tokio_tungstenite::accept_async(stream).await.expect("local WebSocket handshake must succeed"); + let subscribe = read_request(&mut websocket).await; + assert_eq!(subscribe["method"], serde_json::json!("logsSubscribe")); + assert_eq!(subscribe["params"], serde_json::json!([{"mentions":["11111111111111111111111111111111"]},{"commitment":"finalized"}])); + send_result(&mut websocket, &subscribe, serde_json::json!(88)).await; + let id = subscribe.get("id").and_then(serde_json::Value::as_u64).expect("subscribe request id must exist"); + assert!(id > 0); + let notification = serde_json::json!({ + "jsonrpc":"2.0", + "method":"logsNotification", + "params":{ + "result":{"context":{"slot":900},"value":{"signature":"fixture-signature","err":null,"logs":["Program fixture success"]}}, + "subscription":88 + } + }); + websocket.send(tokio_tungstenite::tungstenite::Message::Text(notification.to_string().into())).await.expect("logs notification must send"); + let unsubscribe = read_request(&mut websocket).await; + assert_eq!(unsubscribe["method"], serde_json::json!("logsUnsubscribe")); + assert_eq!(unsubscribe["params"], serde_json::json!([88])); + send_result(&mut websocket, &unsubscribe, serde_json::json!(true)).await; + wait_for_close_frame(&mut websocket).await; + }); + let session = crate::WsSession::connect(local_endpoint(url.as_str())).await.expect("client handshake must succeed"); + let mention = "11111111111111111111111111111111".parse::().expect("mention fixture must parse"); + let filter = crate::SolanaLogsSubscribeFilter::Mentions(mention); + let config = crate::SolanaCommitmentConfig::new(std::option::Option::Some(crate::SolanaCommitment::Finalized)); + let mut subscription = session.logs_subscribe(&filter, std::option::Option::Some(&config)).await.expect("logsSubscribe must register"); + assert_eq!(subscription.kind(), crate::WsSubscriptionKind::Logs); + let notification = subscription.recv().await.expect("logs notification must arrive").expect("logs notification must decode"); + assert_eq!(notification.context().slot(), 900); + assert_eq!(notification.value().signature(), "fixture-signature"); + assert_eq!(notification.value().logs(), &["Program fixture success".to_owned()]); + assert!(subscription.unsubscribe().await.expect("logs unsubscribe must complete")); + session.close().await.expect("session close must complete"); + server.await.expect("local server task must complete"); +} diff --git a/deltas/0.2.7/pre.009.md b/deltas/0.2.7/pre.009.md new file mode 100644 index 0000000..13e3886 --- /dev/null +++ b/deltas/0.2.7/pre.009.md @@ -0,0 +1,197 @@ + + + +# Delta `0.2.7-pre.009` — wrappers WebSocket stables lot A + +## Base + +Base requise : + +```text +0.2.7-pre.008-fix.001 +workspace.package.version = 0.2.7-pre.8.fix.1 +``` + +Le checkpoint opérateur de cette base est vert : `cargo fmt --all`, audit Python, `cargo check --workspace`, `cargo clippy --workspace --all-targets`, tests Transport et `cargo test --workspace`. + +## Signal de version + +```text +livraison = 0.2.7-pre.009 +workspace.package.version = 0.2.7-pre.9 +commit = v0.2.7-pre.009 +tag = aucun +``` + +## Objectif + +Ouvrir les trois premiers wrappers WebSocket Solana standard publics sans exposer le moteur générique provider-extension : + +```text +accountSubscribe / accountUnsubscribe +programSubscribe / programUnsubscribe +logsSubscribe / logsUnsubscribe +``` + +L'unsubscribe reste porté par `WsSubscription::unsubscribe()` et traduit l'identité locale stable vers l'ID serveur courant détenu par l'actor. + +## `accountSubscribe` + +Nouvelle configuration dédiée : + +```text +SolanaAccountSubscribeConfig + encoding + data_slice + commitment +``` + +Les cinq encodings déjà acquis par Transport sont conservés : `binary`, `base58`, `base64`, `jsonParsed`, `base64+zstd`. + +`minContextSlot` n'est pas exposé sur ce config WebSocket. Le type partagé upstream le possède, mais le handler PubSub Agave `v4.2.1` l'ignore explicitement ; KSP ne transforme donc pas ce champ en promesse effective. + +Retour typed : + +```text +WsSubscription> +``` + +Le DTO account et le contexte RPC sont réutilisés sans couche de décodage Program/SPL supplémentaire. + +## `programSubscribe` + +Nouvelle configuration dédiée : + +```text +SolanaProgramSubscribeConfig + account: SolanaAccountSubscribeConfig + filters + with_context +``` + +Les filtres `dataSize`, `memcmp` et `tokenAccountState` sont conservés. Les bornes déterministes retenues par l'audit HTTP sont également appliquées à la surface WS : maximum quatre filtres et maximum 128 octets pour `SolanaMemcmpBytes::Bytes`. + +`withContext` conserve les états omitted/false/true et son défaut upstream `false`. `sortResults` n'est pas exposé : cette option appartient à la surface HTTP `getProgramAccounts` et n'est pas consommée par le handler PubSub audité. + +Le résultat est volontairement une union : + +```text +SolanaProgramNotification::Account(SolanaKeyedAccount) +SolanaProgramNotification::Context(SolanaRpcResponse) +``` + +Le décodeur accepte donc la forme non contextée documentée et la forme contextée observée sans figer une hypothèse plus stricte que l'upstream. + +## `logsSubscribe` + +Nouveau filtre public : + +```text +SolanaLogsSubscribeFilter::All +SolanaLogsSubscribeFilter::AllWithVotes +SolanaLogsSubscribeFilter::Mentions(Pubkey) +``` + +La forme `Mentions(Pubkey)` encode par construction exactement une adresse, conformément à la contrainte upstream actuelle. + +Le commitment réutilise `SolanaCommitmentConfig`. La notification typed est : + +```text +SolanaRpcResponse +``` + +`SolanaLogsNotification` conserve : + +```text +signature : String opaque +err : null ou valeur JSON TransactionError +logs : Vec ordonné +``` + +Aucune interprétation locale des erreurs transactionnelles ou des messages de log n'est ajoutée à Transport. + +## Moteur et lifecycle + +Les trois wrappers utilisent exclusivement `WsSession::subscribe_typed`, qui reste `pub(crate)`. Les acquisitions précédentes restent communes : + +```text +IDs locaux stables +remote IDs internes +ACK/register atomique +reconnect fini +resubscribe déterministe +continuity_gap_count +backpressure par subscription +overflow_count +terminal_error_code +cleanup distant best-effort +``` + +Les paramètres typed sérialisés sont conservés par l'actor et rejoués à l'identique après reconnect avec `ActiveSubscriptions`. + +## Tests déterministes ajoutés + +Sept tests unitaires supplémentaires couvrent : + +```text +account config exact + absence minContextSlot +program config filters/withContext + bornes déterministes +program notification contextée et non contextée +account + program end-to-end sur serveur WS local + unsubscribe exact +logs filters all/allWithVotes/mentions +logs notification context/signature/err/logs + err requis +logs end-to-end sur serveur WS local + unsubscribe exact +``` + +Un canari d'API publique supplémentaire vérifie l'adresse des trois méthodes et des nouveaux DTOs depuis la racine de crate. + +Comptages attendus après compilation : + +```text +Transport unit tests = 294 +Transport public API tests = 33 +release completeness = 22 +``` + +## Sécurité / observabilité + +Aucun wrapper ne journalise les paramètres, pubkeys, logs de transaction ou payloads de notification. Les URL et remote subscription IDs restent absents des DTOs et snapshots publics. + +`SolanaLogsNotification` rend le payload disponible au consumer par API typed, mais il n'est jamais utilisé comme metadata de tracing interne. + +## Fichiers ajoutés ou modifiés + +```text +Cargo.toml +crates/ksp-onchain-transport-lib/README.md +crates/ksp-onchain-transport-lib/USAGE.md +crates/ksp-onchain-transport-lib/src/lib.rs +crates/ksp-onchain-transport-lib/src/rpc_accounts.rs +crates/ksp-onchain-transport-lib/src/rpc_common.rs +crates/ksp-onchain-transport-lib/src/ws_accounts.rs +crates/ksp-onchain-transport-lib/src/ws_session.rs +crates/ksp-onchain-transport-lib/src/ws_transactions.rs +crates/ksp-onchain-transport-lib/tests/public_api.rs +crates/ksp-onchain-transport-lib/unit_tests/ws_accounts.rs +crates/ksp-onchain-transport-lib/unit_tests/ws_transactions.rs +docs/plans/014-V0_2_7_ONCHAIN_WEBSOCKET_PLAN.md +docs/validation/010-V0_2_7_ONCHAIN_WEBSOCKET.md +deltas/0.2.7/pre.009.md +``` + +`ROADMAP.md` et `CHANGELOG.md` restent inchangés pendant cette tranche. + +## Validation opérateur requise + +```bash +cargo fmt --all +python3 scripts/audit_rust_workspace_rules.py +cargo check --workspace +cargo clippy --workspace --all-targets +cargo test -p ksp-onchain-transport-lib +cargo test --workspace +``` + +## Tranche suivante + +Si ce checkpoint est vert, `0.2.7-pre.010` ouvre le lot stable B : `signatureSubscribe`, `slotSubscribe` et `rootSubscribe`, avec terminaison one-shot de signature et compliance `KSP-TRANSPORT-007` associée. diff --git a/docs/plans/014-V0_2_7_ONCHAIN_WEBSOCKET_PLAN.md b/docs/plans/014-V0_2_7_ONCHAIN_WEBSOCKET_PLAN.md index 99596b7..25fcb71 100644 --- a/docs/plans/014-V0_2_7_ONCHAIN_WEBSOCKET_PLAN.md +++ b/docs/plans/014-V0_2_7_ONCHAIN_WEBSOCKET_PLAN.md @@ -1,5 +1,5 @@ - + # Plan `0.2.7` — WebSocket Solana standard @@ -780,6 +780,36 @@ Définir des DTOs WS dédiés pour : enveloppes notification, lifecycle, subscri Pour unstable/évolutif, préférer des enums avec fallback `Unknown { type_name, raw }` borné plutôt qu'un rejet session-wide d'un variant upstream nouveau. +### 16.1 Checkpoint wrappers stables lot A `pre.009` + +Le premier lot public conserve exactement les options retenues par l'audit : + +```text +accountSubscribe + pubkey + encoding = binary | base58 | base64 | jsonParsed | base64+zstd + dataSlice + commitment + minContextSlot non exposé : champ upstream partagé mais ignoré par le handler PubSub v4.2.1 + +programSubscribe + program pubkey + mêmes encoding/dataSlice/commitment account + filters = dataSize | memcmp | tokenAccountState + withContext omitted/false/true + sortResults non exposé : option HTTP non consommée par le handler PubSub audité + notification = keyed account bare OU RpcResponse + +logsSubscribe + filter = all | allWithVotes | mentions exactement une pubkey + commitment + notification = RpcResponse<{ signature, err nullable, logs ordonnés }> +``` + +Les wrappers publics sont `WsSession::account_subscribe`, `WsSession::program_subscribe` et `WsSession::logs_subscribe`. Ils retournent tous `WsSubscription` et réutilisent donc sans duplication le registry local, le remapping remote/local, le reconnect/resubscribe, le backpressure et l'unsubscribe par handle. Le constructeur générique `subscribe_typed` reste `pub(crate)` et aucune méthode provider-extension arbitraire n'entre dans l'API publique. + +La forme `Mentions(Pubkey)` rend la cardinalité `mentions == 1` vraie par construction. Les contraintes déterministes déjà retenues pour les filtres programme sont conservées côté WS : maximum quatre filtres et maximum 128 octets pour la variante raw `Bytes` de `memcmp`; les formes encodées restent laissées au runtime upstream comme sur la surface HTTP existante. + ## 17. API générique provider-specific Le moteur interne doit encoder une spec générique `subscribe_method + unsubscribe_method + params + decoder`, afin qu'une release provider-specific puisse réutiliser la session. @@ -887,7 +917,7 @@ pre.005 DONE — limites frame/message/request + control frames + cancellation/ pre.006 DONE — registry subscriptions + IDs locaux + generic subscribe/unsubscribe engine + channels typed bounded pre.007 DONE — reconnect borné + resubscribe déterministe + continuity gap + races unsubscribe/reconnect pre.008 DONE — backpressure per-sub + overflow/limits + causes terminales sûres + leak/lifecycle adversarial tests -pre.009 wrappers stable lot A : account + program + logs, DTOs/options/KSP-TRANSPORT-007 +pre.009 DONE — wrappers stable lot A : account + program + logs, DTOs/options/KSP-TRANSPORT-007 pre.010 wrappers stable lot B : signature + slot + root, terminaison signature/KSP-TRANSPORT-007 pre.011 unstable : block + slotsUpdates + vote, warnings + fallbacks wire/KSP-TRANSPORT-007 pre.012 compliance 18/18 + canaries public API + composition Config + régressions HTTP diff --git a/docs/validation/010-V0_2_7_ONCHAIN_WEBSOCKET.md b/docs/validation/010-V0_2_7_ONCHAIN_WEBSOCKET.md index 45c5635..a80de40 100644 --- a/docs/validation/010-V0_2_7_ONCHAIN_WEBSOCKET.md +++ b/docs/validation/010-V0_2_7_ONCHAIN_WEBSOCKET.md @@ -1,5 +1,5 @@ - + # Validation `0.2.7` — WebSocket Solana standard @@ -481,6 +481,45 @@ resubscribe RPC application error -> rpc_application_error terminal sur le handl `overflow_count` et `continuity_gap_count` sont deux signaux distincts : le premier mesure les saturations locales de consumers, le second les ruptures de continuité physique. Aucun des deux ne déclenche de replay ou de backfill automatique. Les causes terminales n'exposent qu'un `ErrorCode` KSP, jamais le payload distant, le remote subscription ID ou l'URL. +## 9.7 Checkpoint wrappers stables lot A `pre.009` + +Surface publique ajoutée : + +```text +WsSession::account_subscribe +SolanaAccountSubscribeConfig +WsSubscription> + +WsSession::program_subscribe +SolanaProgramSubscribeConfig +SolanaProgramNotification = bare keyed account | contextual keyed account + +WsSession::logs_subscribe +SolanaLogsSubscribeFilter = All | AllWithVotes | Mentions(Pubkey) +SolanaLogsNotification = signature + err nullable + logs ordonnés +``` + +Gates déterministes ajoutés : + +```text +accountSubscribe -> params exacts pubkey + encoding/dataSlice/commitment +accountSubscribe -> aucun minContextSlot promis ou sérialisable par le config WS +accountNotification -> SolanaRpcContext + SolanaAccount partagé +account handle unsubscribe -> accountUnsubscribe avec remote ID interne +programSubscribe -> filters + withContext préservés, sortResults absent +program filters -> >4 et raw memcmp >128 rejetés avant émission subscribe +programNotification -> forme bare acceptée +programNotification -> forme contextualisée acceptée +program handle unsubscribe -> programUnsubscribe exact +logs filter -> all/allWithVotes/mentions exactement une pubkey +logsNotification -> context + signature + err null/object + ordre logs préservés +logsNotification sans champ err requis -> invalid_response typed, session non concernée +logs handle unsubscribe -> logsUnsubscribe exact +API publique -> aucun subscribe(method, raw params) exposé +``` + +Les trois wrappers passent par le même moteur actor/registry acquis en `pre.006`–`pre.008`; les remote IDs ne deviennent donc pas publics et les paramètres typés initiaux restent les specs rejouées lors d'un resubscribe `ActiveSubscriptions`. Le lot B (`signature`, `slot`, `root`) reste explicitement différé à `pre.010`. + ## 10. Validation du gate `pre.001` Exécuté dans le sandbox :