diff --git a/CHANGELOG.md b/CHANGELOG.md index 5077c8b..9de3566 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,8 +1,16 @@ - + # CHANGELOG +## 0.1.0-pre.025 + +- Migration de la couche WebSocket standard de `kb-onchain-transport`. +- Portage des requêtes et notifications typées des neuf familles d’abonnements Solana. +- Portage du client WebSocket one-shot, du pool orienté rôles et des sessions persistantes avec reconnexion et restauration des abonnements. +- Conservation des 27 tests WebSocket de bot2 adaptés au nouveau nom de crate. +- `Cargo.lock` n’est pas livré dans les deltas ; seules les modifications de `Cargo.toml` sont incluses. + ## 0.1.0-pre.024 - Première tranche fonctionnelle de `kb-onchain-transport`. diff --git a/Cargo.toml b/Cargo.toml index 286cf37..b79e81f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,5 @@ # file: Cargo.toml -# version: 2 +# version: 3 [workspace] resolver = "3" @@ -27,7 +27,7 @@ publish = false [workspace.dependencies] argon2 = { version = "^0.5", features = ["std", "zeroize"] } async-trait = { version = "^0.1", features = [] } -base64 = { version = "^0.22", features = [] } +base64 = { version = "^0.23", features = [] } bytemuck = { version = "^1.25", features = ["derive"] } borsh_0_10 = { package = "borsh", version = "^0.10" } borsh = { version = "^1.7", features = ["ascii", "bson", "bytes", "default", "derive", "de_strict_order", "borsh-derive", "indexmap", "std", "rc"] } diff --git a/ROADMAP.md b/ROADMAP.md index 750ef3b..cba4a01 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -1,5 +1,5 @@ - + # ROADMAP — khadhroony-bot3 @@ -15,12 +15,12 @@ - [x] Porter les modèles de `kb_model` vers `kb-lib`. - [x] Porter les décodeurs Solana core, SPL et Metaplex déjà implémentés dans bot2. - [x] Porter leurs matérialisateurs déjà implémentés dans bot2. -- [ ] Porter leurs exécuteurs. +- [x] Porter leurs exécuteurs. - [x] Fusionner `kb_store_core` et `kb_store_pg` dans `kb-store`. - [ ] Migrer `kb-onchain-transport` depuis l’ancienne `kb_rpc`. - [x] Renommer structurellement la crate. - [x] Porter les contrats JSON-RPC, rôles d’endpoints, validation, clients/pools HTTP et méthodes HTTP standard. - - [ ] Porter WebSocket, sessions et pools d’abonnements. + - [x] Porter WebSocket, sessions et pools d’abonnements. - [ ] Porter l’acquisition canonique `getTransaction` et `getSignaturesForAddress`. - [ ] Porter simulation, envoi et confirmation réseau complets. - [ ] Adapter `kb-pipeline` aux nouveaux chemins publics. diff --git a/kb-onchain-transport/Cargo.toml b/kb-onchain-transport/Cargo.toml index b8cdee0..f437b40 100644 --- a/kb-onchain-transport/Cargo.toml +++ b/kb-onchain-transport/Cargo.toml @@ -9,6 +9,7 @@ license.workspace = true publish.workspace = true [dependencies] +futures-util.workspace = true base64.workspace = true bs58.workspace = true kb-config = { path = "../kb-config" } @@ -18,6 +19,7 @@ reqwest.workspace = true serde.workspace = true serde_json.workspace = true tokio.workspace = true +tokio-tungstenite.workspace = true tracing.workspace = true [lints] diff --git a/kb-onchain-transport/src/endpoint_role.rs b/kb-onchain-transport/src/endpoint_role.rs index d78d7cc..8b8cd14 100644 --- a/kb-onchain-transport/src/endpoint_role.rs +++ b/kb-onchain-transport/src/endpoint_role.rs @@ -1,5 +1,5 @@ // file: kb-onchain-transport/src/endpoint_role.rs -// version: 3 +// version: 4 //! Endpoint role helpers shared by HTTP and WebSocket pools. @@ -50,7 +50,7 @@ pub fn request_kind_from_method(method: &str) -> std::string::String { if trimmed == "logsSubscribe" { return "logs_subscribe_mentions".to_string(); } - return crate::endpoint_role::camel_or_pascal_to_snake(trimmed); + return camel_or_pascal_to_snake(trimmed); } /// Returns true when one endpoint role can handle the requested role and kind. diff --git a/kb-onchain-transport/src/execution_rpc.rs b/kb-onchain-transport/src/execution_rpc.rs index d4ffcc1..3677910 100644 --- a/kb-onchain-transport/src/execution_rpc.rs +++ b/kb-onchain-transport/src/execution_rpc.rs @@ -1,5 +1,5 @@ // file: kb-onchain-transport/src/execution_rpc.rs -// version: 7 +// version: 8 //! Typed Solana JSON-RPC adapters used by execution orchestration. @@ -182,10 +182,8 @@ impl crate::GetFeeForMessageConfig { &self, encoded_message: &str, ) -> kb_core::Result> { - let validation_result = crate::execution_rpc::validate_base64_payload( - encoded_message, - "getFeeForMessage message", - ); + let validation_result = + validate_base64_payload(encoded_message, "getFeeForMessage message"); if let std::result::Result::Err(error) = validation_result { return std::result::Result::Err(error); } @@ -334,10 +332,8 @@ impl crate::SimulateTransactionConfig { &self, encoded_transaction: &str, ) -> kb_core::Result> { - let validation_result = crate::execution_rpc::validate_base64_payload( - encoded_transaction, - "simulateTransaction transaction", - ); + let validation_result = + validate_base64_payload(encoded_transaction, "simulateTransaction transaction"); if let std::result::Result::Err(error) = validation_result { return std::result::Result::Err(error); } @@ -805,10 +801,8 @@ impl crate::SendTransactionConfig { &self, encoded_transaction: &str, ) -> kb_core::Result> { - let validation_result = crate::execution_rpc::validate_base64_payload( - encoded_transaction, - "sendTransaction signed transaction", - ); + let validation_result = + validate_base64_payload(encoded_transaction, "sendTransaction signed transaction"); if let std::result::Result::Err(error) = validation_result { return std::result::Result::Err(error); } @@ -1290,7 +1284,7 @@ pub(crate) fn adapt_get_account_info_result_with_data_limit( )); }, }; - let data = match crate::execution_rpc::decode_account_data(value, space, max_data_bytes) { + let data = match decode_account_data(value, space, max_data_bytes) { std::result::Result::Ok(data) => data, std::result::Result::Err(error) => return std::result::Result::Err(error), }; @@ -1418,11 +1412,10 @@ pub fn adapt_get_minimum_balance_for_rent_exemption_result( pub fn adapt_request_airdrop_result( source: &serde_json::Value, ) -> kb_core::Result { - let signature = - match crate::execution_rpc::adapt_signature_result(source, "requestAirdrop result") { - std::result::Result::Ok(signature) => signature, - std::result::Result::Err(error) => return std::result::Result::Err(error), - }; + let signature = match adapt_signature_result(source, "requestAirdrop result") { + std::result::Result::Ok(signature) => signature, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; return std::result::Result::Ok(crate::AirdropResult { signature }); } @@ -1438,11 +1431,10 @@ pub fn adapt_send_transaction_result( if let std::result::Result::Err(error) = expected_validation { return std::result::Result::Err(error); } - let signature = - match crate::execution_rpc::adapt_signature_result(source, "sendTransaction result") { - std::result::Result::Ok(signature) => signature, - std::result::Result::Err(error) => return std::result::Result::Err(error), - }; + let signature = match adapt_signature_result(source, "sendTransaction result") { + std::result::Result::Ok(signature) => signature, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; if signature != *expected_signature { return std::result::Result::Err(kb_core::Error::new( "execution_send_signature_mismatch", @@ -1970,7 +1962,7 @@ impl crate::HttpEndpointPool { if let std::option::Option::Some(status) = status { last_slot = std::option::Option::Some(status.slot); if let std::option::Option::Some(error) = &status.error { - return std::result::Result::Ok(crate::execution_rpc::confirmation_result( + return std::result::Result::Ok(confirmation_result( cluster, signature, kb_lib::ExApiExecutionConfirmationStatus::Failed, @@ -1984,10 +1976,8 @@ impl crate::HttpEndpointPool { std::option::Option::Some(commitment) => commitment, std::option::Option::None => crate::RpcCommitmentLevel::Processed, }; - if crate::execution_rpc::commitment_reached(observed_commitment, config.commitment) - { - let execution_status = - crate::execution_rpc::execution_confirmation_status(observed_commitment); + if commitment_reached(observed_commitment, config.commitment) { + let execution_status = execution_confirmation_status(observed_commitment); tracing::info!( target: crate::TRACING_TARGET, action = "confirm_transaction", @@ -1997,7 +1987,7 @@ impl crate::HttpEndpointPool { confirmation_status = observed_commitment.as_str(), "transaction reached requested commitment" ); - return std::result::Result::Ok(crate::execution_rpc::confirmation_result( + return std::result::Result::Ok(confirmation_result( cluster, signature, execution_status, @@ -2019,7 +2009,7 @@ impl crate::HttpEndpointPool { }; last_observed_block_height = std::option::Option::Some(height.block_height); if height.block_height > last_valid_block_height { - return std::result::Result::Ok(crate::execution_rpc::confirmation_result( + return std::result::Result::Ok(confirmation_result( cluster, signature, kb_lib::ExApiExecutionConfirmationStatus::Expired, @@ -2037,7 +2027,7 @@ impl crate::HttpEndpointPool { tokio::time::sleep(std::time::Duration::from_millis(config.poll_interval_ms)).await; } } - return std::result::Result::Ok(crate::execution_rpc::confirmation_result( + return std::result::Result::Ok(confirmation_result( cluster, signature, kb_lib::ExApiExecutionConfirmationStatus::TimedOut, @@ -2150,8 +2140,7 @@ fn commitment_reached( observed: crate::RpcCommitmentLevel, required: crate::RpcCommitmentLevel, ) -> bool { - return crate::execution_rpc::commitment_rank(observed) - >= crate::execution_rpc::commitment_rank(required); + return commitment_rank(observed) >= commitment_rank(required); } fn execution_confirmation_status( @@ -2688,11 +2677,11 @@ mod tests { let config = crate::ConfirmTransactionConfig::confirmed(100, 500, 20) .unwrap_or_else(|error| panic!("unexpected error: {error}")); assert_eq!(config.last_valid_block_height, std::option::Option::Some(100)); - assert!(crate::execution_rpc::commitment_reached( + assert!(super::commitment_reached( crate::RpcCommitmentLevel::Finalized, crate::RpcCommitmentLevel::Confirmed, )); - assert!(!crate::execution_rpc::commitment_reached( + assert!(!super::commitment_reached( crate::RpcCommitmentLevel::Processed, crate::RpcCommitmentLevel::Confirmed, )); diff --git a/kb-onchain-transport/src/lib.rs b/kb-onchain-transport/src/lib.rs index e1e836d..2211327 100644 --- a/kb-onchain-transport/src/lib.rs +++ b/kb-onchain-transport/src/lib.rs @@ -1,5 +1,5 @@ // file: kb-onchain-transport/src/lib.rs -// version: 4 +// version: 6 #![forbid(unsafe_code)] #![deny(unreachable_pub)] @@ -22,14 +22,16 @@ mod standard_http_economics; mod standard_http_tokens; mod standard_http_transactions; mod standard_methods; +mod standard_ws; mod validation; +mod ws_client; +mod ws_pool; +mod ws_session; /// RPC endpoint configuration. pub use self::client::RpcEndpoint; /// Minimal Solana RPC client abstraction. pub use self::client::SolanaRpcClient; -/// Canonical tracing target for this crate. -pub(crate) use self::constants::TRACING_TARGET; /// Endpoint role snapshot shared by HTTP and WebSocket pools. pub use self::endpoint_role::EndpointRoleSnapshot; /// Converts a JSON-RPC method name into a stable request kind. @@ -340,12 +342,106 @@ pub use self::standard_methods::StandardWsSubscriptionSpec; pub use self::standard_methods::standard_http_method; /// Finds one standard WebSocket subscription from its subscribe or unsubscribe method. pub use self::standard_methods::standard_ws_subscription; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::AccountSubscribeRequest; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::BlockSubscribeRequest; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::LogsSubscribeRequest; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::ProgramSubscribeRequest; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::RootSubscribeRequest; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::SignatureSubscribeRequest; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::SlotSubscribeRequest; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::SlotsUpdatesSubscribeRequest; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::StandardWsCapabilities; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::StandardWsNotification; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::StandardWsRequest; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::VoteSubscribeRequest; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::WsAccountNotification; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::WsAccountSubscribeConfig; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::WsBlockFilter; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::WsBlockNotification; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::WsBlockSubscribeConfig; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::WsBlockUpdate; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::WsLogsFilter; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::WsLogsNotification; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::WsLogsSubscribeConfig; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::WsLogsValue; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::WsProgramNotification; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::WsProgramSubscribeConfig; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::WsRootNotification; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::WsSignatureNotification; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::WsSignatureSubscribeConfig; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::WsSignatureValue; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::WsSlotInfo; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::WsSlotNotification; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::WsSlotTransactionStats; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::WsSlotUpdate; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::WsSlotsUpdatesNotification; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::WsVoteNotification; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::WsVoteValue; +/// Typed contracts for all standard Solana WebSocket subscriptions. +pub use self::standard_ws::adapt_standard_ws_notification; /// Validates one base58 Solana blockhash or genesis hash. pub use self::validation::validate_solana_hash_text; /// Validates one base58 Solana public key. pub use self::validation::validate_solana_pubkey_text; /// Validates one base58 Solana transaction signature. pub use self::validation::validate_transaction_signature_text; +/// Standard Solana WebSocket client bound to one endpoint. +pub use self::ws_client::WsClient; +/// Snapshot of one WebSocket pool endpoint. +pub use self::ws_client::WsPoolClientSnapshot; +/// WebSocket endpoint pool with role-based routing. +pub use self::ws_pool::WsEndpointPool; +/// Persistent WebSocket session contracts. +pub use self::ws_session::WsReconnectPolicy; +/// Persistent WebSocket session contracts. +pub use self::ws_session::WsSession; +/// Persistent WebSocket session contracts. +pub use self::ws_session::WsSessionEvent; +/// Persistent WebSocket session contracts. +pub use self::ws_session::WsSessionSnapshot; +/// Persistent WebSocket session contracts. +pub use self::ws_session::WsSessionState; +/// Persistent WebSocket session contracts. +pub use self::ws_session::WsSubscriptionAck; +/// Persistent WebSocket session contracts. +pub use self::ws_session::WsSubscriptionSnapshot; +/// Acknowledgement returned after a WebSocket unsubscribe request. +pub use self::ws_session::WsUnsubscribeAck; /// Internal DEVNET_GENESIS_HASH contract. pub(crate) use self::constants::DEVNET_GENESIS_HASH; @@ -381,6 +477,8 @@ pub(crate) use self::constants::MAX_SIMULATION_ACCOUNT_COUNT; pub(crate) use self::constants::MAX_SLOT_LEADER_COUNT; /// Internal TESTNET_GENESIS_HASH contract. pub(crate) use self::constants::TESTNET_GENESIS_HASH; +/// Canonical tracing target for this crate. +pub(crate) use self::constants::TRACING_TARGET; /// Internal role_matches contract. pub(crate) use self::endpoint_role::role_matches; /// Internal serialize_parameter contract. diff --git a/kb-onchain-transport/src/standard_http_tokens.rs b/kb-onchain-transport/src/standard_http_tokens.rs index ddebaae..68dd708 100644 --- a/kb-onchain-transport/src/standard_http_tokens.rs +++ b/kb-onchain-transport/src/standard_http_tokens.rs @@ -1,5 +1,5 @@ // file: kb-onchain-transport/src/standard_http_tokens.rs -// version: 2 +// version: 3 //! Configurable standard SPL Token-oriented Solana HTTP JSON-RPC requests. @@ -75,7 +75,7 @@ impl crate::StandardHttpRequest for crate::GetTokenAccountBalanceRequest { const METHOD: &'static str = "getTokenAccountBalance"; fn params(&self) -> kb_core::Result> { - return crate::standard_http_tokens::pubkey_with_optional_commitment_params( + return pubkey_with_optional_commitment_params( Self::METHOD, &self.address, "getTokenAccountBalance address", @@ -101,7 +101,7 @@ impl crate::StandardHttpRequest for crate::GetTokenAccountsByDelegateRequest { const METHOD: &'static str = "getTokenAccountsByDelegate"; fn params(&self) -> kb_core::Result> { - return crate::standard_http_tokens::token_accounts_query_params( + return token_accounts_query_params( Self::METHOD, &self.delegate, "getTokenAccountsByDelegate delegate", @@ -128,7 +128,7 @@ impl crate::StandardHttpRequest for crate::GetTokenAccountsByOwnerRequest { const METHOD: &'static str = "getTokenAccountsByOwner"; fn params(&self) -> kb_core::Result> { - return crate::standard_http_tokens::token_accounts_query_params( + return token_accounts_query_params( Self::METHOD, &self.owner, "getTokenAccountsByOwner owner", @@ -153,7 +153,7 @@ impl crate::StandardHttpRequest for crate::GetTokenLargestAccountsRequest { const METHOD: &'static str = "getTokenLargestAccounts"; fn params(&self) -> kb_core::Result> { - return crate::standard_http_tokens::pubkey_with_optional_commitment_params( + return pubkey_with_optional_commitment_params( Self::METHOD, &self.mint, "getTokenLargestAccounts mint", @@ -177,7 +177,7 @@ impl crate::StandardHttpRequest for crate::GetTokenSupplyRequest { const METHOD: &'static str = "getTokenSupply"; fn params(&self) -> kb_core::Result> { - return crate::standard_http_tokens::pubkey_with_optional_commitment_params( + return pubkey_with_optional_commitment_params( Self::METHOD, &self.mint, "getTokenSupply mint", diff --git a/kb-onchain-transport/src/standard_ws.rs b/kb-onchain-transport/src/standard_ws.rs new file mode 100644 index 0000000..6ceb83d --- /dev/null +++ b/kb-onchain-transport/src/standard_ws.rs @@ -0,0 +1,1251 @@ +// file: kb-onchain-transport/src/standard_ws.rs +// version: 4 + +//! Typed contracts for every standard Solana WebSocket subscription. + +/// Effective endpoint capabilities for unstable standard subscriptions. +/// +/// Configuration enables an optimistic first attempt. A persistent session may disable one +/// capability after the node explicitly reports that the method is absent or not enabled. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, serde::Deserialize, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct StandardWsCapabilities { + /// Whether `blockSubscribe` may be attempted on this endpoint. + pub block_subscribe: bool, + /// Whether `slotsUpdatesSubscribe` may be attempted on this endpoint. + pub slots_updates_subscribe: bool, + /// Whether `voteSubscribe` may be attempted on this endpoint. + pub vote_subscribe: bool, +} + +impl crate::StandardWsCapabilities { + /// Builds initial capabilities from request kinds explicitly advertised by an endpoint. + pub fn from_endpoint(endpoint: &kb_config::WsEndpointConfig) -> Self { + return Self { + block_subscribe: endpoint_declares_method(endpoint, "blockSubscribe"), + slots_updates_subscribe: endpoint_declares_method(endpoint, "slotsUpdatesSubscribe"), + vote_subscribe: endpoint_declares_method(endpoint, "voteSubscribe"), + }; + } + + fn allows(&self, method: &str) -> bool { + return match method { + "blockSubscribe" => self.block_subscribe, + "slotsUpdatesSubscribe" => self.slots_updates_subscribe, + "voteSubscribe" => self.vote_subscribe, + _ => true, + }; + } + + pub(crate) fn disable_method(&mut self, method: &str) -> bool { + let capability = match method { + "blockSubscribe" => &mut self.block_subscribe, + "slotsUpdatesSubscribe" => &mut self.slots_updates_subscribe, + "voteSubscribe" => &mut self.vote_subscribe, + _ => return false, + }; + if !*capability { + return false; + } + *capability = false; + return true; + } + + pub(crate) fn is_unstable_method(method: &str) -> bool { + return matches!(method, "blockSubscribe" | "slotsUpdatesSubscribe" | "voteSubscribe"); + } +} + +/// Account options accepted by `accountSubscribe`. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, serde::Deserialize, serde::Serialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct WsAccountSubscribeConfig { + /// Optional commitment level. + #[serde(skip_serializing_if = "std::option::Option::is_none")] + pub commitment: std::option::Option, + /// Optional account-data encoding. + #[serde(skip_serializing_if = "std::option::Option::is_none")] + pub encoding: std::option::Option, + /// Optional account-data byte slice. + #[serde(skip_serializing_if = "std::option::Option::is_none")] + pub data_slice: std::option::Option, +} + +impl crate::WsAccountSubscribeConfig { + fn validate(&self) -> kb_core::Result<()> { + return crate::RpcAccountInfoConfig { + encoding: self.encoding, + data_slice: self.data_slice, + commitment: self.commitment, + min_context_slot: std::option::Option::None, + } + .validate(); + } +} + +/// Filter accepted by `blockSubscribe`. +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum WsBlockFilter { + /// Subscribe to every block. + All, + /// Subscribe to blocks mentioning one account or Program ID. + MentionsAccountOrProgram(std::string::String), +} + +impl serde::Serialize for crate::WsBlockFilter { + fn serialize( + &self, + serializer: Serializer, + ) -> std::result::Result + where + Serializer: serde::Serializer, + { + return match self { + Self::All => serializer.serialize_str("all"), + Self::MentionsAccountOrProgram(pubkey) => { + let value = serde_json::json!({ "mentionsAccountOrProgram": pubkey }); + serde::Serialize::serialize(&value, serializer) + }, + }; + } +} + +impl<'de> serde::Deserialize<'de> for crate::WsBlockFilter { + fn deserialize( + deserializer: Deserializer, + ) -> std::result::Result + where + Deserializer: serde::Deserializer<'de>, + { + let value = match ::deserialize(deserializer) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + if value == serde_json::Value::String("all".to_string()) { + return std::result::Result::Ok(Self::All); + } + let pubkey = value + .as_object() + .filter(|object| return object.len() == 1) + .and_then(|object| return object.get("mentionsAccountOrProgram")) + .and_then(serde_json::Value::as_str); + return match pubkey { + std::option::Option::Some(pubkey) => { + std::result::Result::Ok(Self::MentionsAccountOrProgram(pubkey.to_string())) + }, + std::option::Option::None => { + std::result::Result::Err(::custom( + "block filter must be 'all' or mentionsAccountOrProgram", + )) + }, + }; + } +} + +/// Options accepted by `blockSubscribe`. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, serde::Deserialize, serde::Serialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct WsBlockSubscribeConfig { + /// Optional confirmed or finalized commitment. + #[serde(skip_serializing_if = "std::option::Option::is_none")] + pub commitment: std::option::Option, + /// Optional transaction encoding. + #[serde(skip_serializing_if = "std::option::Option::is_none")] + pub encoding: std::option::Option, + /// Optional transaction detail level. + #[serde(skip_serializing_if = "std::option::Option::is_none")] + pub transaction_details: std::option::Option, + /// Whether rewards are included. + #[serde(skip_serializing_if = "std::option::Option::is_none")] + pub show_rewards: std::option::Option, + /// Highest transaction version the caller can decode. + #[serde(skip_serializing_if = "std::option::Option::is_none")] + pub max_supported_transaction_version: std::option::Option, +} + +impl crate::WsBlockSubscribeConfig { + fn validate(&self) -> kb_core::Result<()> { + if self.commitment == std::option::Option::Some(crate::RpcCommitmentLevel::Processed) { + return std::result::Result::Err(kb_core::Error::config( + "blockSubscribe does not accept processed commitment", + )); + } + return std::result::Result::Ok(()); + } +} + +/// Filter accepted by `logsSubscribe`. +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum WsLogsFilter { + /// Subscribe to non-simple-vote transaction logs. + All, + /// Subscribe to every transaction log, including simple votes. + AllWithVotes, + /// Subscribe to transactions mentioning one account or Program ID. + Mentions(std::string::String), +} + +impl serde::Serialize for crate::WsLogsFilter { + fn serialize( + &self, + serializer: Serializer, + ) -> std::result::Result + where + Serializer: serde::Serializer, + { + return match self { + Self::All => serializer.serialize_str("all"), + Self::AllWithVotes => serializer.serialize_str("allWithVotes"), + Self::Mentions(pubkey) => { + let value = serde_json::json!({ "mentions": [pubkey] }); + serde::Serialize::serialize(&value, serializer) + }, + }; + } +} + +impl<'de> serde::Deserialize<'de> for crate::WsLogsFilter { + fn deserialize( + deserializer: Deserializer, + ) -> std::result::Result + where + Deserializer: serde::Deserializer<'de>, + { + let value = match ::deserialize(deserializer) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + if value == serde_json::Value::String("all".to_string()) { + return std::result::Result::Ok(Self::All); + } + if value == serde_json::Value::String("allWithVotes".to_string()) { + return std::result::Result::Ok(Self::AllWithVotes); + } + let mentions = value + .as_object() + .filter(|object| return object.len() == 1) + .and_then(|object| return object.get("mentions")) + .and_then(serde_json::Value::as_array); + let pubkey = mentions + .and_then(|values| { + if values.len() == 1 { + return values.first(); + } + return std::option::Option::None; + }) + .and_then(serde_json::Value::as_str); + return match pubkey { + std::option::Option::Some(pubkey) => { + std::result::Result::Ok(Self::Mentions(pubkey.to_string())) + }, + std::option::Option::None => { + std::result::Result::Err(::custom( + "logs mentions filter must contain exactly one public key", + )) + }, + }; + } +} + +/// Options accepted by `logsSubscribe`. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, serde::Deserialize, serde::Serialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct WsLogsSubscribeConfig { + /// Optional commitment level. + #[serde(skip_serializing_if = "std::option::Option::is_none")] + pub commitment: std::option::Option, +} + +/// Options accepted by `programSubscribe`. +#[derive(Clone, Debug, Default, Eq, PartialEq, serde::Deserialize, serde::Serialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct WsProgramSubscribeConfig { + /// Optional account filters, evaluated in order. + #[serde(skip_serializing_if = "std::option::Option::is_none")] + pub filters: std::option::Option>, + /// Optional account-data encoding. + #[serde(skip_serializing_if = "std::option::Option::is_none")] + pub encoding: std::option::Option, + /// Optional account-data byte slice. + #[serde(skip_serializing_if = "std::option::Option::is_none")] + pub data_slice: std::option::Option, + /// Optional commitment level. + #[serde(skip_serializing_if = "std::option::Option::is_none")] + pub commitment: std::option::Option, + /// Whether the result includes the standard context wrapper. + #[serde(skip_serializing_if = "std::option::Option::is_none")] + pub with_context: std::option::Option, +} + +impl crate::WsProgramSubscribeConfig { + fn validate(&self) -> kb_core::Result<()> { + return crate::RpcProgramAccountsConfig { + filters: self.filters.clone(), + encoding: self.encoding, + data_slice: self.data_slice, + commitment: self.commitment, + min_context_slot: std::option::Option::None, + with_context: self.with_context, + sort_results: std::option::Option::None, + } + .validate(); + } +} + +/// Options accepted by `signatureSubscribe`. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, serde::Deserialize, serde::Serialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct WsSignatureSubscribeConfig { + /// Optional commitment level. + #[serde(skip_serializing_if = "std::option::Option::is_none")] + pub commitment: std::option::Option, + /// Whether an early `receivedSignature` notification is requested. + #[serde(skip_serializing_if = "std::option::Option::is_none")] + pub enable_received_notification: std::option::Option, +} + +/// Typed account subscription request. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct AccountSubscribeRequest { + /// Account public key. + pub pubkey: std::string::String, + /// Optional commitment, encoding and slice options. + pub config: std::option::Option, +} + +/// Typed block subscription request. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct BlockSubscribeRequest { + /// Block selection filter. + pub filter: crate::WsBlockFilter, + /// Optional commitment, encoding, details, rewards and version options. + pub config: std::option::Option, +} + +/// Typed logs subscription request. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct LogsSubscribeRequest { + /// Log selection filter. + pub filter: crate::WsLogsFilter, + /// Optional commitment. + pub config: std::option::Option, +} + +/// Typed program-account subscription request. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct ProgramSubscribeRequest { + /// Program public key. + pub program_id: std::string::String, + /// Optional filters, commitment, encoding, slice and context switch. + pub config: std::option::Option, +} + +/// Typed root subscription request. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct RootSubscribeRequest; + +/// Typed signature subscription request. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct SignatureSubscribeRequest { + /// Transaction signature. + pub signature: std::string::String, + /// Optional commitment and received-notification switch. + pub config: std::option::Option, +} + +/// Typed slot subscription request. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct SlotSubscribeRequest; + +/// Typed unstable slot-updates subscription request. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct SlotsUpdatesSubscribeRequest; + +/// Typed unstable vote subscription request. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct VoteSubscribeRequest; + +/// One of the nine standard Solana WebSocket subscription requests. +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum StandardWsRequest { + /// `accountSubscribe` request. + Account(crate::AccountSubscribeRequest), + /// `blockSubscribe` request. + Block(crate::BlockSubscribeRequest), + /// `logsSubscribe` request. + Logs(crate::LogsSubscribeRequest), + /// `programSubscribe` request. + Program(crate::ProgramSubscribeRequest), + /// `rootSubscribe` request. + Root(crate::RootSubscribeRequest), + /// `signatureSubscribe` request. + Signature(crate::SignatureSubscribeRequest), + /// `slotSubscribe` request. + Slot(crate::SlotSubscribeRequest), + /// `slotsUpdatesSubscribe` request. + SlotsUpdates(crate::SlotsUpdatesSubscribeRequest), + /// `voteSubscribe` request. + Vote(crate::VoteSubscribeRequest), +} + +impl crate::StandardWsRequest { + /// Returns the canonical subscription specification. + pub fn specification(&self) -> &'static crate::StandardWsSubscriptionSpec { + return match self { + Self::Account(_) => &crate::STANDARD_WS_SUBSCRIPTIONS[0], + Self::Block(_) => &crate::STANDARD_WS_SUBSCRIPTIONS[1], + Self::Logs(_) => &crate::STANDARD_WS_SUBSCRIPTIONS[2], + Self::Program(_) => &crate::STANDARD_WS_SUBSCRIPTIONS[3], + Self::Root(_) => &crate::STANDARD_WS_SUBSCRIPTIONS[4], + Self::Signature(_) => &crate::STANDARD_WS_SUBSCRIPTIONS[5], + Self::Slot(_) => &crate::STANDARD_WS_SUBSCRIPTIONS[6], + Self::SlotsUpdates(_) => &crate::STANDARD_WS_SUBSCRIPTIONS[7], + Self::Vote(_) => &crate::STANDARD_WS_SUBSCRIPTIONS[8], + }; + } + + /// Returns the exact subscribe method. + pub const fn subscribe_method(&self) -> &'static str { + return match self { + Self::Account(_) => "accountSubscribe", + Self::Block(_) => "blockSubscribe", + Self::Logs(_) => "logsSubscribe", + Self::Program(_) => "programSubscribe", + Self::Root(_) => "rootSubscribe", + Self::Signature(_) => "signatureSubscribe", + Self::Slot(_) => "slotSubscribe", + Self::SlotsUpdates(_) => "slotsUpdatesSubscribe", + Self::Vote(_) => "voteSubscribe", + }; + } + + /// Validates capability gates and builds exact positional JSON-RPC parameters. + pub fn params( + &self, + capabilities: crate::StandardWsCapabilities, + ) -> kb_core::Result> { + if !capabilities.allows(self.subscribe_method()) { + return std::result::Result::Err(kb_core::Error::config(format!( + "unstable WebSocket method '{}' is not enabled by the endpoint capability set", + self.subscribe_method() + ))); + } + return match self { + Self::Account(request) => account_params(request), + Self::Block(request) => block_params(request), + Self::Logs(request) => logs_params(request), + Self::Program(request) => program_params(request), + Self::Root(_) | Self::Slot(_) | Self::SlotsUpdates(_) | Self::Vote(_) => { + std::result::Result::Ok(std::vec::Vec::new()) + }, + Self::Signature(request) => signature_params(request), + }; + } +} + +/// Typed account notification. +#[derive(Clone, Debug, PartialEq, serde::Deserialize, serde::Serialize)] +pub struct WsAccountNotification { + /// Remote subscription identifier. + pub subscription: u64, + /// Contextual account value. + pub result: crate::RpcResponse, +} + +/// One block update emitted by `blockSubscribe`. +#[derive(Clone, Debug, PartialEq, serde::Deserialize, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct WsBlockUpdate { + /// Block slot. + pub slot: u64, + /// Block when the node produced one for the slot. + #[serde(default)] + pub block: std::option::Option, + /// Node-provided block error when present. + #[serde(default)] + pub err: std::option::Option, +} + +/// Typed block notification. +#[derive(Clone, Debug, PartialEq, serde::Deserialize, serde::Serialize)] +pub struct WsBlockNotification { + /// Remote subscription identifier. + pub subscription: u64, + /// Contextual block update. + pub result: crate::RpcResponse, +} + +/// One log notification value. +#[derive(Clone, Debug, PartialEq, serde::Deserialize, serde::Serialize)] +pub struct WsLogsValue { + /// Transaction signature. + pub signature: std::string::String, + /// Runtime transaction error, or `None` on success. + pub err: std::option::Option, + /// Runtime log messages. + pub logs: std::vec::Vec, +} + +/// Typed logs notification. +#[derive(Clone, Debug, PartialEq, serde::Deserialize, serde::Serialize)] +pub struct WsLogsNotification { + /// Remote subscription identifier. + pub subscription: u64, + /// Contextual log result. + pub result: crate::RpcResponse, +} + +/// Typed program-account notification. +#[derive(Clone, Debug, PartialEq, serde::Deserialize, serde::Serialize)] +pub struct WsProgramNotification { + /// Remote subscription identifier. + pub subscription: u64, + /// Contextual keyed account result. + pub result: crate::RpcOptionalContext, +} + +/// Typed root notification. +#[derive(Clone, Copy, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)] +pub struct WsRootNotification { + /// Remote subscription identifier. + pub subscription: u64, + /// New root slot. + pub result: u64, +} + +/// Signature notification result. +#[derive(Clone, Debug, PartialEq, serde::Deserialize, serde::Serialize)] +#[serde(untagged)] +pub enum WsSignatureValue { + /// Final signature status. + Status { + /// Runtime transaction error, or `None` on success. + err: std::option::Option, + }, + /// Early node-receipt marker. + Received(std::string::String), +} + +/// Typed signature notification. +#[derive(Clone, Debug, PartialEq, serde::Deserialize, serde::Serialize)] +pub struct WsSignatureNotification { + /// Remote subscription identifier. + pub subscription: u64, + /// Contextual receipt or final status. + pub result: crate::RpcResponse, +} + +/// One slot progression notification. +#[derive(Clone, Copy, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)] +pub struct WsSlotInfo { + /// Current slot. + pub slot: u64, + /// Parent slot. + pub parent: u64, + /// Current root slot. + pub root: u64, +} + +/// Typed slot notification. +#[derive(Clone, Copy, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)] +pub struct WsSlotNotification { + /// Remote subscription identifier. + pub subscription: u64, + /// Slot progression value. + pub result: crate::WsSlotInfo, +} + +/// Transaction counters included by a frozen-slot update. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, serde::Deserialize, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct WsSlotTransactionStats { + /// Total transaction entries processed by the bank. + pub num_transaction_entries: u64, + /// Successful transactions. + pub num_successful_transactions: u64, + /// Failed transactions. + pub num_failed_transactions: u64, + /// Maximum transactions per entry. + pub max_transactions_per_entry: u64, +} + +/// One unstable slot lifecycle update. +#[derive(Clone, Debug, PartialEq, serde::Deserialize, serde::Serialize)] +#[serde(tag = "type", rename_all = "camelCase")] +pub enum WsSlotUpdate { + /// First shred for a slot was received. + FirstShredReceived { + /// Slot. + slot: u64, + /// Unix timestamp in milliseconds. + timestamp: i64, + }, + /// Every shred for a slot was received. + Completed { + /// Slot. + slot: u64, + /// Unix timestamp in milliseconds. + timestamp: i64, + }, + /// A bank was created for a slot. + CreatedBank { + /// Slot. + slot: u64, + /// Parent slot. + parent: u64, + /// Unix timestamp in milliseconds. + timestamp: i64, + }, + /// A bank was frozen. + Frozen { + /// Slot. + slot: u64, + /// Unix timestamp in milliseconds. + timestamp: i64, + /// Bank transaction counters. + stats: crate::WsSlotTransactionStats, + }, + /// A slot was marked dead. + Dead { + /// Slot. + slot: u64, + /// Unix timestamp in milliseconds. + timestamp: i64, + /// Node-provided failure description. + err: std::string::String, + }, + /// A slot reached optimistic confirmation. + OptimisticConfirmation { + /// Slot. + slot: u64, + /// Unix timestamp in milliseconds. + timestamp: i64, + }, + /// A slot became root. + Root { + /// Slot. + slot: u64, + /// Unix timestamp in milliseconds. + timestamp: i64, + }, +} + +/// Typed unstable slot-updates notification. +#[derive(Clone, Debug, PartialEq, serde::Deserialize, serde::Serialize)] +pub struct WsSlotsUpdatesNotification { + /// Remote subscription identifier. + pub subscription: u64, + /// Slot lifecycle update. + pub result: crate::WsSlotUpdate, +} + +/// One vote notification value. +#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct WsVoteValue { + /// Vote account public key. + pub vote_pubkey: std::string::String, + /// Voted slots. + pub slots: std::vec::Vec, + /// Vote hash. + pub hash: std::string::String, + /// Optional vote timestamp. + #[serde(default)] + pub timestamp: std::option::Option, + /// Vote transaction signature. + pub signature: std::string::String, +} + +/// Typed unstable vote notification. +#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)] +pub struct WsVoteNotification { + /// Remote subscription identifier. + pub subscription: u64, + /// Vote value. + pub result: crate::WsVoteValue, +} + +/// Typed notification emitted by one of the nine standard subscriptions. +#[derive(Clone, Debug, PartialEq, serde::Deserialize, serde::Serialize)] +#[serde(tag = "kind", content = "notification", rename_all = "camelCase")] +pub enum StandardWsNotification { + /// `accountNotification`. + Account(crate::WsAccountNotification), + /// `blockNotification`. + Block(crate::WsBlockNotification), + /// `logsNotification`. + Logs(crate::WsLogsNotification), + /// `programNotification`. + Program(crate::WsProgramNotification), + /// `rootNotification`. + Root(crate::WsRootNotification), + /// `signatureNotification`. + Signature(crate::WsSignatureNotification), + /// `slotNotification`. + Slot(crate::WsSlotNotification), + /// `slotsUpdatesNotification`. + SlotsUpdates(crate::WsSlotsUpdatesNotification), + /// `voteNotification`. + Vote(crate::WsVoteNotification), +} + +impl crate::StandardWsNotification { + /// Returns the remote subscription identifier. + pub const fn subscription_id(&self) -> u64 { + return match self { + Self::Account(value) => value.subscription, + Self::Block(value) => value.subscription, + Self::Logs(value) => value.subscription, + Self::Program(value) => value.subscription, + Self::Root(value) => value.subscription, + Self::Signature(value) => value.subscription, + Self::Slot(value) => value.subscription, + Self::SlotsUpdates(value) => value.subscription, + Self::Vote(value) => value.subscription, + }; + } +} + +/// Parses and validates one standard WebSocket notification. +pub fn adapt_standard_ws_notification( + notification: &crate::JsonRpcNotification, +) -> kb_core::Result { + let subscription = notification.params.subscription; + let result = ¬ification.params.result; + return match notification.method.as_str() { + "accountNotification" => { + adapt_notification_result(subscription, result, crate::WsAccountNotification::new) + .map(crate::StandardWsNotification::Account) + }, + "blockNotification" => { + adapt_notification_result(subscription, result, crate::WsBlockNotification::new) + .map(crate::StandardWsNotification::Block) + }, + "logsNotification" => { + adapt_notification_result(subscription, result, crate::WsLogsNotification::new) + .map(crate::StandardWsNotification::Logs) + }, + "programNotification" => { + adapt_notification_result(subscription, result, crate::WsProgramNotification::new) + .map(crate::StandardWsNotification::Program) + }, + "rootNotification" => { + adapt_notification_result(subscription, result, crate::WsRootNotification::new) + .map(crate::StandardWsNotification::Root) + }, + "signatureNotification" => { + let adapted = adapt_notification_result( + subscription, + result, + crate::WsSignatureNotification::new, + ); + return match adapted { + std::result::Result::Ok(notification) => { + if let crate::WsSignatureValue::Received(marker) = ¬ification.result.value { + if marker != "receivedSignature" { + return std::result::Result::Err(kb_core::Error::json(format!( + "unknown signature notification marker '{marker}'" + ))); + } + } + std::result::Result::Ok(crate::StandardWsNotification::Signature(notification)) + }, + std::result::Result::Err(error) => std::result::Result::Err(error), + }; + }, + "slotNotification" => { + adapt_notification_result(subscription, result, crate::WsSlotNotification::new) + .map(crate::StandardWsNotification::Slot) + }, + "slotsUpdatesNotification" => { + adapt_notification_result(subscription, result, crate::WsSlotsUpdatesNotification::new) + .map(crate::StandardWsNotification::SlotsUpdates) + }, + "voteNotification" => { + adapt_notification_result(subscription, result, crate::WsVoteNotification::new) + .map(crate::StandardWsNotification::Vote) + }, + method => std::result::Result::Err(kb_core::Error::json(format!( + "unsupported standard WebSocket notification method '{method}'" + ))), + }; +} + +impl crate::WsAccountNotification { + fn new(subscription: u64, result: crate::RpcResponse) -> Self { + return Self { subscription, result }; + } +} + +impl crate::WsBlockNotification { + fn new(subscription: u64, result: crate::RpcResponse) -> Self { + return Self { subscription, result }; + } +} + +impl crate::WsLogsNotification { + fn new(subscription: u64, result: crate::RpcResponse) -> Self { + return Self { subscription, result }; + } +} + +impl crate::WsProgramNotification { + fn new(subscription: u64, result: crate::RpcOptionalContext) -> Self { + return Self { subscription, result }; + } +} + +impl crate::WsRootNotification { + fn new(subscription: u64, result: u64) -> Self { + return Self { subscription, result }; + } +} + +impl crate::WsSignatureNotification { + fn new(subscription: u64, result: crate::RpcResponse) -> Self { + return Self { subscription, result }; + } +} + +impl crate::WsSlotNotification { + fn new(subscription: u64, result: crate::WsSlotInfo) -> Self { + return Self { subscription, result }; + } +} + +impl crate::WsSlotsUpdatesNotification { + fn new(subscription: u64, result: crate::WsSlotUpdate) -> Self { + return Self { subscription, result }; + } +} + +impl crate::WsVoteNotification { + fn new(subscription: u64, result: crate::WsVoteValue) -> Self { + return Self { subscription, result }; + } +} + +fn endpoint_declares_method(endpoint: &kb_config::WsEndpointConfig, method: &str) -> bool { + let request_kind = crate::request_kind_from_method(method); + for role in &endpoint.roles { + if !role.enabled { + continue; + } + for configured_kind in &role.request_kinds { + if configured_kind == "*" || configured_kind == &request_kind { + return true; + } + } + } + return false; +} + +fn account_params( + request: &crate::AccountSubscribeRequest, +) -> kb_core::Result> { + let validation_result = crate::validate_solana_pubkey_text(&request.pubkey, "account pubkey"); + if let std::result::Result::Err(error) = validation_result { + return std::result::Result::Err(error); + } + let mut params = std::vec![serde_json::Value::String(request.pubkey.clone())]; + if let std::option::Option::Some(config) = request.config { + let config_result = config.validate(); + if let std::result::Result::Err(error) = config_result { + return std::result::Result::Err(error); + } + let value = match crate::serialize_parameter("accountSubscribe", &config) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + params.push(value); + } + return std::result::Result::Ok(params); +} + +fn block_params( + request: &crate::BlockSubscribeRequest, +) -> kb_core::Result> { + if let crate::WsBlockFilter::MentionsAccountOrProgram(pubkey) = &request.filter { + let validation_result = + crate::validate_solana_pubkey_text(pubkey, "block mentions account or program"); + if let std::result::Result::Err(error) = validation_result { + return std::result::Result::Err(error); + } + } + let filter = match crate::serialize_parameter("blockSubscribe", &request.filter) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let mut params = std::vec![filter]; + if let std::option::Option::Some(config) = request.config { + let config_result = config.validate(); + if let std::result::Result::Err(error) = config_result { + return std::result::Result::Err(error); + } + let value = match crate::serialize_parameter("blockSubscribe", &config) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + params.push(value); + } + return std::result::Result::Ok(params); +} + +fn logs_params( + request: &crate::LogsSubscribeRequest, +) -> kb_core::Result> { + if let crate::WsLogsFilter::Mentions(pubkey) = &request.filter { + let validation_result = crate::validate_solana_pubkey_text(pubkey, "logs mention"); + if let std::result::Result::Err(error) = validation_result { + return std::result::Result::Err(error); + } + } + let filter = match crate::serialize_parameter("logsSubscribe", &request.filter) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let mut params = std::vec![filter]; + if let std::option::Option::Some(config) = request.config { + let value = match crate::serialize_parameter("logsSubscribe", &config) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + params.push(value); + } + return std::result::Result::Ok(params); +} + +fn program_params( + request: &crate::ProgramSubscribeRequest, +) -> kb_core::Result> { + let validation_result = crate::validate_solana_pubkey_text(&request.program_id, "program id"); + if let std::result::Result::Err(error) = validation_result { + return std::result::Result::Err(error); + } + let mut params = std::vec![serde_json::Value::String(request.program_id.clone())]; + if let std::option::Option::Some(config) = &request.config { + let config_result = config.validate(); + if let std::result::Result::Err(error) = config_result { + return std::result::Result::Err(error); + } + let value = match crate::serialize_parameter("programSubscribe", config) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + params.push(value); + } + return std::result::Result::Ok(params); +} + +fn signature_params( + request: &crate::SignatureSubscribeRequest, +) -> kb_core::Result> { + let validation_result = + crate::validate_transaction_signature_text(&request.signature, "transaction signature"); + if let std::result::Result::Err(error) = validation_result { + return std::result::Result::Err(error); + } + let mut params = std::vec![serde_json::Value::String(request.signature.clone())]; + if let std::option::Option::Some(config) = request.config { + let value = match crate::serialize_parameter("signatureSubscribe", &config) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + params.push(value); + } + return std::result::Result::Ok(params); +} + +fn adapt_notification_result( + subscription: u64, + value: &serde_json::Value, + constructor: Constructor, +) -> kb_core::Result +where + ResultType: serde::de::DeserializeOwned, + Constructor: FnOnce(u64, ResultType) -> NotificationType, +{ + let result = match serde_json::from_value::(value.clone()) { + std::result::Result::Ok(result) => result, + std::result::Result::Err(error) => { + return std::result::Result::Err(kb_core::Error::json(format!( + "cannot decode standard WebSocket notification result: {error}" + ))); + }, + }; + return std::result::Result::Ok(constructor(subscription, result)); +} + +#[cfg(test)] +mod tests { + const PUBKEY: &str = "11111111111111111111111111111111"; + const SIGNATURE: &str = "1111111111111111111111111111111111111111111111111111111111111111"; + + #[test] + fn every_standard_request_maps_to_the_compiled_registry() { + let requests = [ + crate::StandardWsRequest::Account(crate::AccountSubscribeRequest { + pubkey: PUBKEY.to_string(), + config: std::option::Option::None, + }), + crate::StandardWsRequest::Block(crate::BlockSubscribeRequest { + filter: crate::WsBlockFilter::All, + config: std::option::Option::None, + }), + crate::StandardWsRequest::Logs(crate::LogsSubscribeRequest { + filter: crate::WsLogsFilter::All, + config: std::option::Option::None, + }), + crate::StandardWsRequest::Program(crate::ProgramSubscribeRequest { + program_id: PUBKEY.to_string(), + config: std::option::Option::None, + }), + crate::StandardWsRequest::Root(crate::RootSubscribeRequest), + crate::StandardWsRequest::Signature(crate::SignatureSubscribeRequest { + signature: SIGNATURE.to_string(), + config: std::option::Option::None, + }), + crate::StandardWsRequest::Slot(crate::SlotSubscribeRequest), + crate::StandardWsRequest::SlotsUpdates(crate::SlotsUpdatesSubscribeRequest), + crate::StandardWsRequest::Vote(crate::VoteSubscribeRequest), + ]; + let mut methods = std::collections::BTreeSet::new(); + for request in requests { + assert!(methods.insert(request.subscribe_method())); + assert_eq!(request.specification().subscribe_method, request.subscribe_method()); + } + assert_eq!(methods.len(), crate::STANDARD_WS_SUBSCRIPTIONS.len()); + } + + #[test] + fn stable_request_options_are_independently_serialized() { + let request = crate::StandardWsRequest::Account(crate::AccountSubscribeRequest { + pubkey: PUBKEY.to_string(), + config: std::option::Option::Some(crate::WsAccountSubscribeConfig { + commitment: std::option::Option::Some(crate::RpcCommitmentLevel::Confirmed), + encoding: std::option::Option::Some(crate::RpcAccountEncoding::Base64Zstd), + data_slice: std::option::Option::Some(crate::RpcDataSlice { + offset: 8, + length: 16, + }), + }), + }); + let params = match request.params(crate::StandardWsCapabilities::default()) { + std::result::Result::Ok(params) => params, + std::result::Result::Err(error) => panic!("params failed: {error}"), + }; + assert_eq!(params[0], serde_json::Value::String(PUBKEY.to_string())); + assert_eq!(params[1]["commitment"], "confirmed"); + assert_eq!(params[1]["encoding"], "base64+zstd"); + assert_eq!(params[1]["dataSlice"]["offset"], 8); + } + + #[test] + fn custom_filter_objects_reject_unknown_fields() { + let block = serde_json::json!({ + "mentionsAccountOrProgram": PUBKEY, + "unexpected": true + }); + assert!(serde_json::from_value::(block).is_err()); + let logs = serde_json::json!({ + "mentions": [PUBKEY], + "unexpected": true + }); + assert!(serde_json::from_value::(logs).is_err()); + } + + #[test] + fn logs_mentions_keeps_exact_single_key_shape() { + let request = crate::StandardWsRequest::Logs(crate::LogsSubscribeRequest { + filter: crate::WsLogsFilter::Mentions(PUBKEY.to_string()), + config: std::option::Option::Some(crate::WsLogsSubscribeConfig { + commitment: std::option::Option::Some(crate::RpcCommitmentLevel::Finalized), + }), + }); + let params = match request.params(crate::StandardWsCapabilities::default()) { + std::result::Result::Ok(params) => params, + std::result::Result::Err(error) => panic!("params failed: {error}"), + }; + assert_eq!(params[0]["mentions"][0], PUBKEY); + assert_eq!(params[1]["commitment"], "finalized"); + } + + #[test] + fn unstable_requests_require_explicit_capabilities() { + let requests = [ + crate::StandardWsRequest::Block(crate::BlockSubscribeRequest { + filter: crate::WsBlockFilter::All, + config: std::option::Option::None, + }), + crate::StandardWsRequest::SlotsUpdates(crate::SlotsUpdatesSubscribeRequest), + crate::StandardWsRequest::Vote(crate::VoteSubscribeRequest), + ]; + for request in &requests { + assert!(request.params(crate::StandardWsCapabilities::default()).is_err()); + } + let capabilities = crate::StandardWsCapabilities { + block_subscribe: true, + slots_updates_subscribe: true, + vote_subscribe: true, + }; + for request in &requests { + assert!(request.params(capabilities).is_ok()); + } + } + + #[test] + fn unstable_capabilities_can_be_disabled_after_runtime_rejection() { + let mut capabilities = crate::StandardWsCapabilities { + block_subscribe: true, + slots_updates_subscribe: true, + vote_subscribe: true, + }; + assert!(capabilities.disable_method("blockSubscribe")); + assert!(!capabilities.block_subscribe); + assert!(!capabilities.disable_method("blockSubscribe")); + assert!(capabilities.disable_method("slotsUpdatesSubscribe")); + assert!(capabilities.disable_method("voteSubscribe")); + assert!(!capabilities.disable_method("slotSubscribe")); + assert!(crate::StandardWsCapabilities::is_unstable_method("blockSubscribe")); + assert!(!crate::StandardWsCapabilities::is_unstable_method("slotSubscribe")); + } + + #[test] + fn block_processed_commitment_is_rejected() { + let request = crate::StandardWsRequest::Block(crate::BlockSubscribeRequest { + filter: crate::WsBlockFilter::All, + config: std::option::Option::Some(crate::WsBlockSubscribeConfig { + commitment: std::option::Option::Some(crate::RpcCommitmentLevel::Processed), + ..crate::WsBlockSubscribeConfig::default() + }), + }); + let capabilities = crate::StandardWsCapabilities { + block_subscribe: true, + slots_updates_subscribe: false, + vote_subscribe: false, + }; + assert!(request.params(capabilities).is_err()); + } + + #[test] + fn notification_adapter_decodes_each_standard_shape() { + let account = serde_json::json!({ + "context": { "slot": 1 }, + "value": { + "lamports": 1, + "owner": PUBKEY, + "executable": false, + "rentEpoch": 2, + "space": 0, + "data": ["", "base64"] + } + }); + let keyed_account = serde_json::json!({ + "context": { "slot": 1 }, + "value": { + "pubkey": PUBKEY, + "account": { + "lamports": 1, + "owner": PUBKEY, + "executable": false, + "rentEpoch": 2, + "space": 0, + "data": ["", "base64"] + } + } + }); + let cases = [ + ("accountNotification", account, "account"), + ( + "blockNotification", + serde_json::json!({ + "context": { "slot": 4 }, + "value": { + "slot": 4, + "block": { + "blockhash": PUBKEY, + "previousBlockhash": PUBKEY, + "parentSlot": 3 + }, + "err": null + } + }), + "block", + ), + ( + "logsNotification", + serde_json::json!({ + "context": { "slot": 5 }, + "value": { "signature": SIGNATURE, "err": null, "logs": ["Program log"] } + }), + "logs", + ), + ("programNotification", keyed_account, "program"), + ("rootNotification", serde_json::Value::from(42_u64), "root"), + ( + "signatureNotification", + serde_json::json!({ + "context": { "slot": 6 }, + "value": { "err": null } + }), + "signature", + ), + ( + "slotNotification", + serde_json::json!({ "slot": 3, "parent": 2, "root": 1 }), + "slot", + ), + ( + "slotsUpdatesNotification", + serde_json::json!({ "type": "completed", "slot": 3, "timestamp": 4 }), + "slotsUpdates", + ), + ( + "voteNotification", + serde_json::json!({ + "votePubkey": PUBKEY, + "slots": [1, 2], + "hash": PUBKEY, + "timestamp": null, + "signature": SIGNATURE + }), + "vote", + ), + ]; + for (method, result, kind) in cases { + let notification = crate::JsonRpcNotification { + jsonrpc: "2.0".to_string(), + method: method.to_string(), + params: crate::JsonRpcNotificationParams { result, subscription: 7 }, + }; + let adapted = match crate::adapt_standard_ws_notification(¬ification) { + std::result::Result::Ok(adapted) => adapted, + std::result::Result::Err(error) => panic!("notification failed: {error}"), + }; + let serialized = match serde_json::to_value(adapted) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("serialization failed: {error}"), + }; + assert_eq!(serialized["kind"], kind); + } + } + + #[test] + fn signature_received_marker_is_exact() { + let valid = crate::JsonRpcNotification { + jsonrpc: "2.0".to_string(), + method: "signatureNotification".to_string(), + params: crate::JsonRpcNotificationParams { + result: serde_json::json!({ + "context": { "slot": 1 }, + "value": "receivedSignature" + }), + subscription: 9, + }, + }; + assert!(crate::adapt_standard_ws_notification(&valid).is_ok()); + let mut invalid = valid; + invalid.params.result["value"] = serde_json::Value::String("received".to_string()); + assert!(crate::adapt_standard_ws_notification(&invalid).is_err()); + } +} diff --git a/kb-onchain-transport/src/validation.rs b/kb-onchain-transport/src/validation.rs index e76d697..b317c8c 100644 --- a/kb-onchain-transport/src/validation.rs +++ b/kb-onchain-transport/src/validation.rs @@ -1,21 +1,21 @@ // file: kb-onchain-transport/src/validation.rs -// version: 3 +// version: 4 //! Shared validation helpers for Solana RPC addresses, hashes and signatures. /// Validates one base58 Solana transaction signature. pub fn validate_transaction_signature_text(value: &str, field_name: &str) -> kb_core::Result<()> { - return crate::validation::validate_base58_length(value, field_name, 64); + return validate_base58_length(value, field_name, 64); } /// Validates one base58 Solana public key or account address. pub fn validate_solana_pubkey_text(value: &str, field_name: &str) -> kb_core::Result<()> { - return crate::validation::validate_base58_length(value, field_name, 32); + return validate_base58_length(value, field_name, 32); } /// Validates one base58 Solana blockhash or genesis hash. pub fn validate_solana_hash_text(value: &str, field_name: &str) -> kb_core::Result<()> { - return crate::validation::validate_base58_length(value, field_name, 32); + return validate_base58_length(value, field_name, 32); } fn validate_base58_length( diff --git a/kb-onchain-transport/src/ws_client.rs b/kb-onchain-transport/src/ws_client.rs new file mode 100644 index 0000000..ff3dcf4 --- /dev/null +++ b/kb-onchain-transport/src/ws_client.rs @@ -0,0 +1,386 @@ +// file: kb-onchain-transport/src/ws_client.rs +// version: 8 + +//! Standard Solana WebSocket client helpers. + +use futures_util::SinkExt; // rust-rules: trait-import +use futures_util::StreamExt; // rust-rules: trait-import + +/// Snapshot of one pooled WebSocket endpoint. +#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct WsPoolClientSnapshot { + /// Logical endpoint name. + pub endpoint_name: std::string::String, + /// Provider name. + pub provider: std::string::String, + /// Endpoint URL. + pub endpoint_url: std::string::String, + /// Supported roles. + pub roles: std::vec::Vec, + /// Status string. + pub status: std::string::String, +} + +/// Standard Solana WebSocket client bound to one configured endpoint. +#[derive(Clone, Debug)] +pub struct WsClient { + endpoint: kb_config::WsEndpointConfig, + next_request_id: std::sync::Arc, +} + +impl crate::WsClient { + /// Creates a new WebSocket client bound to one endpoint. + pub fn new(endpoint: kb_config::WsEndpointConfig) -> kb_core::Result { + if !endpoint.enabled { + tracing::error!(target: crate::TRACING_TARGET, action = "create_ws_client", endpoint_name = %endpoint.name, provider = %endpoint.provider, error_code = "ws_endpoint_disabled", "cannot create WebSocket client for disabled endpoint"); + return std::result::Result::Err(kb_core::Error::config(format!( + "ws endpoint '{}' is disabled", + endpoint.name + ))); + } + tracing::debug!(target: crate::TRACING_TARGET, action = "create_ws_client", endpoint_name = %endpoint.name, provider = %endpoint.provider, role_count = endpoint.roles.len(), "WebSocket client created"); + return std::result::Result::Ok(Self { + endpoint, + next_request_id: std::sync::Arc::new(std::sync::atomic::AtomicU64::new(1)), + }); + } + + /// Returns the endpoint name. + pub fn endpoint_name(&self) -> &str { + return self.endpoint.name.as_str(); + } + + /// Returns the provider name. + pub fn provider(&self) -> &str { + return self.endpoint.provider.as_str(); + } + + /// Returns the endpoint URL. + pub fn endpoint_url(&self) -> &str { + return self.endpoint.url.as_str(); + } + + /// Returns the endpoint configuration. + pub fn endpoint_config(&self) -> &kb_config::WsEndpointConfig { + return &self.endpoint; + } + + /// Returns true when this endpoint supports the required role and request kind. + pub fn can_handle(&self, required_role: &str, request_kind: &str) -> bool { + if !self.endpoint.enabled { + return false; + } + for role in &self.endpoint.roles { + if crate::role_matches(role, required_role, request_kind) { + return true; + } + } + return false; + } + + /// Returns a serializable endpoint snapshot. + pub fn snapshot(&self) -> crate::WsPoolClientSnapshot { + let mut roles = std::vec::Vec::new(); + for role in &self.endpoint.roles { + roles.push(crate::EndpointRoleSnapshot::from_config(role)); + } + return crate::WsPoolClientSnapshot { + endpoint_name: self.endpoint.name.clone(), + provider: self.endpoint.provider.clone(), + endpoint_url: self.endpoint.url.clone(), + roles, + status: "idle".to_string(), + }; + } + + /// Builds a JSON-RPC request with a generated id. + pub fn build_json_rpc_request( + &self, + method: std::string::String, + params: std::vec::Vec, + ) -> crate::JsonRpcRequest { + let request_id = self.next_request_id.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + return crate::JsonRpcRequest::new_with_u64_id(request_id, method, params); + } + + /// Builds a subscribe request for one explicitly registered standard subscription. + pub fn build_standard_subscribe_request( + &self, + subscription: &crate::StandardWsSubscriptionSpec, + params: std::vec::Vec, + ) -> crate::JsonRpcRequest { + return self.build_json_rpc_request(subscription.subscribe_method.to_string(), params); + } + + /// Builds an unsubscribe request for one explicitly registered standard subscription. + pub fn build_standard_unsubscribe_request( + &self, + subscription: &crate::StandardWsSubscriptionSpec, + subscription_id: u64, + ) -> crate::JsonRpcRequest { + return self.build_json_rpc_request( + subscription.unsubscribe_method.to_string(), + std::vec![serde_json::Value::from(subscription_id)], + ); + } + + /// Connects, sends one JSON-RPC request, waits for one response and closes. + pub async fn execute_json_rpc_once( + &self, + method: std::string::String, + params: std::vec::Vec, + ) -> kb_core::Result { + let parameter_count = params.len(); + let request = self.build_json_rpc_request(method.clone(), params); + tracing::debug!(target: crate::TRACING_TARGET, action = "execute_ws_json_rpc_once", endpoint_name = %self.endpoint.name, provider = %self.endpoint.provider, request_id = ?request.id, method = %method, parameter_count, "start one-shot WebSocket JSON-RPC request"); + let request_text = match request.to_json_string() { + std::result::Result::Ok(text) => text, + std::result::Result::Err(error) => { + tracing::error!(target: crate::TRACING_TARGET, action = "serialize_ws_json_rpc_request", endpoint_name = %self.endpoint.name, provider = %self.endpoint.provider, request_id = ?request.id, method = %method, error = %error, "WebSocket JSON-RPC request serialization failed"); + return std::result::Result::Err(error); + }, + }; + let connect_timeout = std::time::Duration::from_millis(self.endpoint.connect_timeout_ms); + let connect_future = tokio_tungstenite::connect_async(self.endpoint.url.as_str()); + let connect_timeout_result = tokio::time::timeout(connect_timeout, connect_future).await; + let connect_result = match connect_timeout_result { + std::result::Result::Ok(result) => result, + std::result::Result::Err(_) => { + tracing::error!(target: crate::TRACING_TARGET, action = "connect_ws_endpoint", endpoint_name = %self.endpoint.name, provider = %self.endpoint.provider, request_id = ?request.id, method = %method, timeout_ms = self.endpoint.connect_timeout_ms, error_code = "ws_connect_timeout", "WebSocket endpoint connection timed out"); + return std::result::Result::Err(kb_core::Error::ws(format!( + "websocket connect timed out for endpoint '{}'", + self.endpoint.name + ))); + }, + }; + let (mut stream, _response) = match connect_result { + std::result::Result::Ok(pair) => pair, + std::result::Result::Err(error) => { + tracing::error!(target: crate::TRACING_TARGET, action = "connect_ws_endpoint", endpoint_name = %self.endpoint.name, provider = %self.endpoint.provider, request_id = ?request.id, method = %method, error = %error, "WebSocket endpoint connection failed"); + return std::result::Result::Err(kb_core::Error::ws(format!( + "cannot connect websocket endpoint '{}': {error}", + self.endpoint.name + ))); + }, + }; + let send_result = stream + .send(tokio_tungstenite::tungstenite::Message::Text(request_text.into())) + .await; + if let std::result::Result::Err(error) = send_result { + tracing::error!(target: crate::TRACING_TARGET, action = "send_ws_json_rpc_request", endpoint_name = %self.endpoint.name, provider = %self.endpoint.provider, request_id = ?request.id, method = %method, error = %error, "WebSocket JSON-RPC request send failed"); + return std::result::Result::Err(kb_core::Error::ws(format!( + "cannot send websocket request '{}' to endpoint '{}': {error}", + method, self.endpoint.name + ))); + } + let response_timeout = std::time::Duration::from_millis(self.endpoint.request_timeout_ms); + let next_timeout_result = tokio::time::timeout(response_timeout, stream.next()).await; + let next_result = match next_timeout_result { + std::result::Result::Ok(result) => result, + std::result::Result::Err(_) => { + tracing::error!(target: crate::TRACING_TARGET, action = "read_ws_json_rpc_response", endpoint_name = %self.endpoint.name, provider = %self.endpoint.provider, request_id = ?request.id, method = %method, timeout_ms = self.endpoint.request_timeout_ms, error_code = "ws_response_timeout", "WebSocket JSON-RPC response timed out"); + return std::result::Result::Err(kb_core::Error::ws(format!( + "websocket response timed out for endpoint '{}'", + self.endpoint.name + ))); + }, + }; + let message = match next_result { + std::option::Option::Some(result) => match result { + std::result::Result::Ok(message) => message, + std::result::Result::Err(error) => { + tracing::error!(target: crate::TRACING_TARGET, action = "read_ws_json_rpc_response", endpoint_name = %self.endpoint.name, provider = %self.endpoint.provider, request_id = ?request.id, method = %method, error = %error, "WebSocket JSON-RPC response read failed"); + return std::result::Result::Err(kb_core::Error::ws(format!( + "websocket read failed for endpoint '{}': {error}", + self.endpoint.name + ))); + }, + }, + std::option::Option::None => { + tracing::error!(target: crate::TRACING_TARGET, action = "read_ws_json_rpc_response", endpoint_name = %self.endpoint.name, provider = %self.endpoint.provider, request_id = ?request.id, method = %method, error_code = "ws_closed_before_response", "WebSocket endpoint closed before response"); + return std::result::Result::Err(kb_core::Error::ws(format!( + "websocket endpoint '{}' closed before response", + self.endpoint.name + ))); + }, + }; + let close_result = stream + .send(tokio_tungstenite::tungstenite::Message::Close(std::option::Option::None)) + .await; + if let std::result::Result::Err(error) = close_result { + tracing::debug!(target: crate::TRACING_TARGET, action = "close_ws_endpoint", endpoint_name = %self.endpoint.name, provider = %self.endpoint.provider, request_id = ?request.id, method = %method, error = %error, "WebSocket close send failed"); + } + return match message { + tokio_tungstenite::tungstenite::Message::Text(text) => { + let parse_result = crate::parse_json_rpc_text(text.as_str()); + match parse_result { + std::result::Result::Ok(response) => { + if let crate::JsonRpcResponse::Error(error_response) = &response { + tracing::error!(target: crate::TRACING_TARGET, action = "execute_ws_json_rpc_once", endpoint_name = %self.endpoint.name, provider = %self.endpoint.provider, request_id = ?request.id, method = %method, response_kind = response.kind_name(), rpc_error_code = error_response.error.code, rpc_error_message = %error_response.error.message, "WebSocket JSON-RPC endpoint returned an RPC error"); + } else { + tracing::debug!(target: crate::TRACING_TARGET, action = "execute_ws_json_rpc_once", endpoint_name = %self.endpoint.name, provider = %self.endpoint.provider, request_id = ?request.id, method = %method, response_kind = response.kind_name(), "one-shot WebSocket JSON-RPC request completed"); + } + std::result::Result::Ok(response) + }, + std::result::Result::Err(error) => { + tracing::error!(target: crate::TRACING_TARGET, action = "parse_ws_json_rpc_response", endpoint_name = %self.endpoint.name, provider = %self.endpoint.provider, request_id = ?request.id, method = %method, response_byte_length = text.len(), error = %error, "WebSocket JSON-RPC response parsing failed"); + std::result::Result::Err(error) + }, + } + }, + tokio_tungstenite::tungstenite::Message::Binary(binary) => { + tracing::error!(target: crate::TRACING_TARGET, action = "read_ws_json_rpc_response", endpoint_name = %self.endpoint.name, provider = %self.endpoint.provider, request_id = ?request.id, method = %method, message_kind = "binary", response_byte_length = binary.len(), "WebSocket endpoint returned binary data before JSON response"); + std::result::Result::Err(kb_core::Error::ws(format!( + "websocket endpoint '{}' returned binary message with {} bytes", + self.endpoint.name, + binary.len() + ))) + }, + tokio_tungstenite::tungstenite::Message::Ping(_) => { + tracing::error!(target: crate::TRACING_TARGET, action = "read_ws_json_rpc_response", endpoint_name = %self.endpoint.name, provider = %self.endpoint.provider, request_id = ?request.id, method = %method, message_kind = "ping", "WebSocket endpoint returned ping before JSON response"); + std::result::Result::Err(kb_core::Error::ws(format!( + "websocket endpoint '{}' returned ping before json response", + self.endpoint.name + ))) + }, + tokio_tungstenite::tungstenite::Message::Pong(_) => { + tracing::error!(target: crate::TRACING_TARGET, action = "read_ws_json_rpc_response", endpoint_name = %self.endpoint.name, provider = %self.endpoint.provider, request_id = ?request.id, method = %method, message_kind = "pong", "WebSocket endpoint returned pong before JSON response"); + std::result::Result::Err(kb_core::Error::ws(format!( + "websocket endpoint '{}' returned pong before json response", + self.endpoint.name + ))) + }, + tokio_tungstenite::tungstenite::Message::Close(_) => { + tracing::error!(target: crate::TRACING_TARGET, action = "read_ws_json_rpc_response", endpoint_name = %self.endpoint.name, provider = %self.endpoint.provider, request_id = ?request.id, method = %method, message_kind = "close", "WebSocket endpoint closed before JSON response"); + std::result::Result::Err(kb_core::Error::ws(format!( + "websocket endpoint '{}' closed before json response", + self.endpoint.name + ))) + }, + tokio_tungstenite::tungstenite::Message::Frame(_) => { + tracing::error!(target: crate::TRACING_TARGET, action = "read_ws_json_rpc_response", endpoint_name = %self.endpoint.name, provider = %self.endpoint.provider, request_id = ?request.id, method = %method, message_kind = "frame", "WebSocket endpoint returned raw frame before JSON response"); + std::result::Result::Err(kb_core::Error::ws(format!( + "websocket endpoint '{}' returned raw frame before json response", + self.endpoint.name + ))) + }, + }; + } +} + +#[cfg(test)] +mod tests { + fn role_config( + role: &str, + request_kinds: std::vec::Vec, + ) -> kb_config::EndpointRoleConfig { + return kb_config::EndpointRoleConfig { + role: role.to_string(), + enabled: true, + request_kinds, + priority: 1, + requests_per_second: 10, + burst_capacity: 10, + max_concurrent_requests: 4, + max_subscriptions: 16, + pause_after_rate_limit_ms: 1500, + }; + } + + fn endpoint(enabled: bool) -> kb_config::WsEndpointConfig { + return kb_config::WsEndpointConfig { + name: "ws_a".to_string(), + enabled, + provider: "test".to_string(), + cluster: "devnet".to_string(), + url: "wss://example.invalid".to_string(), + connect_timeout_ms: 100, + request_timeout_ms: 100, + unsubscribe_timeout_ms: 100, + write_channel_capacity: 8, + event_channel_capacity: 16, + auto_reconnect: false, + roles: std::vec![ + role_config("slot_notifications", std::vec!["slot_subscribe".to_string()]), + role_config("program_subscribe", std::vec!["program_subscribe".to_string()]), + role_config("ws_any", std::vec!["*".to_string()]), + ], + }; + } + + #[test] + fn new_rejects_disabled_endpoint() { + let result = crate::WsClient::new(endpoint(false)); + assert!(result.is_err()); + } + + #[test] + fn can_handle_matches_exact_role_and_kind() { + let client = match crate::WsClient::new(endpoint(true)) { + std::result::Result::Ok(client) => client, + std::result::Result::Err(error) => panic!("client creation failed: {error}"), + }; + assert!(client.can_handle("slot_notifications", "slot_subscribe")); + assert!(!client.can_handle("slot_notifications", "root_subscribe")); + } + + #[test] + fn can_handle_matches_wildcard_kind() { + let client = match crate::WsClient::new(endpoint(true)) { + std::result::Result::Ok(client) => client, + std::result::Result::Err(error) => panic!("client creation failed: {error}"), + }; + assert!(client.can_handle("ws_any", "logs_subscribe_mentions")); + } + + #[test] + fn snapshot_preserves_endpoint_metadata() { + let client = match crate::WsClient::new(endpoint(true)) { + std::result::Result::Ok(client) => client, + std::result::Result::Err(error) => panic!("client creation failed: {error}"), + }; + let snapshot = client.snapshot(); + assert_eq!(snapshot.endpoint_name, "ws_a"); + assert_eq!(snapshot.provider, "test"); + assert_eq!(snapshot.endpoint_url, "wss://example.invalid"); + assert_eq!(snapshot.roles.len(), 3); + } + + #[test] + fn standard_subscription_builders_preserve_exact_methods_and_subscription_id() { + let client = match crate::WsClient::new(endpoint(true)) { + std::result::Result::Ok(client) => client, + std::result::Result::Err(error) => panic!("client creation failed: {error}"), + }; + let subscription = match crate::standard_ws_subscription("slotSubscribe") { + std::option::Option::Some(value) => value, + std::option::Option::None => panic!("slot subscription missing"), + }; + let subscribe = client.build_standard_subscribe_request( + subscription, + std::vec![serde_json::json!({"commitment": "confirmed"})], + ); + let unsubscribe = client.build_standard_unsubscribe_request(subscription, 42); + assert_eq!(subscribe.method, "slotSubscribe"); + assert_eq!(subscribe.params.len(), 1); + assert_eq!(unsubscribe.method, "slotUnsubscribe"); + assert_eq!(unsubscribe.params, std::vec![serde_json::Value::from(42)]); + assert_eq!(subscribe.id, serde_json::Value::from(1)); + assert_eq!(unsubscribe.id, serde_json::Value::from(2)); + } + + #[test] + fn build_json_rpc_request_increments_ids() { + let client = match crate::WsClient::new(endpoint(true)) { + std::result::Result::Ok(client) => client, + std::result::Result::Err(error) => panic!("client creation failed: {error}"), + }; + let first = + client.build_json_rpc_request("slotSubscribe".to_string(), std::vec::Vec::new()); + let second = + client.build_json_rpc_request("rootSubscribe".to_string(), std::vec::Vec::new()); + assert_eq!(first.id, serde_json::Value::from(1)); + assert_eq!(second.id, serde_json::Value::from(2)); + } +} diff --git a/kb-onchain-transport/src/ws_pool.rs b/kb-onchain-transport/src/ws_pool.rs new file mode 100644 index 0000000..f512ffe --- /dev/null +++ b/kb-onchain-transport/src/ws_pool.rs @@ -0,0 +1,266 @@ +// file: kb-onchain-transport/src/ws_pool.rs +// version: 6 + +//! WebSocket endpoint pool and role-based routing. + +/// Pool of standard Solana WebSocket endpoints. +#[derive(Clone, Debug)] +pub struct WsEndpointPool { + clients: std::vec::Vec, + next_index: std::sync::Arc, +} + +impl crate::WsEndpointPool { + /// Builds a pool from the active profile WebSocket endpoint list. + pub fn from_profile(profile: &kb_config::ProfileConfig) -> kb_core::Result { + let mut clients = std::vec::Vec::new(); + for endpoint in &profile.solana.ws_endpoints { + if !endpoint.enabled { + continue; + } + let client = match crate::WsClient::new(endpoint.clone()) { + std::result::Result::Ok(client) => client, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + clients.push(client); + } + return crate::WsEndpointPool::new(clients); + } + + /// Creates a pool from already constructed clients. + pub fn new(clients: std::vec::Vec) -> kb_core::Result { + if clients.is_empty() { + tracing::error!(target: crate::TRACING_TARGET, action = "create_ws_pool", error_code = "ws_pool_empty", "WebSocket endpoint pool has no enabled endpoint"); + return std::result::Result::Err(kb_core::Error::config( + "ws endpoint pool requires at least one enabled endpoint".to_string(), + )); + } + tracing::debug!(target: crate::TRACING_TARGET, action = "create_ws_pool", endpoint_count = clients.len(), "WebSocket endpoint pool created"); + return std::result::Result::Ok(Self { + clients, + next_index: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)), + }); + } + + /// Returns a serializable snapshot of every endpoint in the pool. + pub fn snapshot(&self) -> std::vec::Vec { + let mut snapshots = std::vec::Vec::new(); + for client in &self.clients { + snapshots.push(client.snapshot()); + } + return snapshots; + } + + /// Selects one endpoint for the requested role and method. + pub fn select_client_for_role_and_method( + &self, + required_role: &str, + method: &str, + ) -> kb_core::Result { + let request_kind = crate::request_kind_from_method(method); + return self.select_client_for_role_and_kind(required_role, &request_kind); + } + + /// Selects one endpoint for the requested role and request kind. + pub fn select_client_for_role_and_kind( + &self, + required_role: &str, + request_kind: &str, + ) -> kb_core::Result { + if self.clients.is_empty() { + tracing::error!(target: crate::TRACING_TARGET, action = "select_ws_endpoint", required_role, request_kind, error_code = "ws_pool_empty", "WebSocket endpoint pool has no clients"); + return std::result::Result::Err(kb_core::Error::not_connected( + "ws endpoint pool has no clients".to_string(), + )); + } + let start_index = self.next_index.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + let client_count = self.clients.len(); + let mut offset = 0_usize; + while offset < client_count { + let index = (start_index + offset) % client_count; + let client = self.clients[index].clone(); + if client.can_handle(required_role, request_kind) { + tracing::debug!(target: crate::TRACING_TARGET, action = "select_ws_endpoint", required_role, request_kind, endpoint_name = %client.endpoint_name(), provider = %client.provider(), "selected WebSocket endpoint"); + return std::result::Result::Ok(client); + } + offset += 1; + } + tracing::error!(target: crate::TRACING_TARGET, action = "select_ws_endpoint", required_role, request_kind, endpoint_count = self.clients.len(), error_code = "ws_endpoint_not_found", "no WebSocket endpoint supports requested role and kind"); + return std::result::Result::Err(kb_core::Error::config(format!( + "no ws endpoint supports role '{}' and request kind '{}'", + required_role, request_kind + ))); + } + + /// Selects one endpoint for an explicitly registered standard WebSocket subscription. + pub fn select_client_for_standard_subscription( + &self, + required_role: &str, + subscription: &crate::StandardWsSubscriptionSpec, + ) -> kb_core::Result { + return self + .select_client_for_role_and_method(required_role, subscription.subscribe_method); + } + + /// Executes one short WebSocket JSON-RPC request through the selected endpoint. + pub async fn execute_json_rpc_once_for_role( + &self, + required_role: &str, + method: std::string::String, + params: std::vec::Vec, + ) -> kb_core::Result { + let client = match self.select_client_for_role_and_method(required_role, &method) { + std::result::Result::Ok(client) => client, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + return client.execute_json_rpc_once(method, params).await; + } +} + +#[cfg(test)] +mod tests { + fn role_config( + role: &str, + request_kinds: std::vec::Vec, + ) -> kb_config::EndpointRoleConfig { + return kb_config::EndpointRoleConfig { + role: role.to_string(), + enabled: true, + request_kinds, + priority: 1, + requests_per_second: 10, + burst_capacity: 10, + max_concurrent_requests: 4, + max_subscriptions: 16, + pause_after_rate_limit_ms: 1500, + }; + } + + fn endpoint( + name: &str, + role: &str, + request_kinds: std::vec::Vec, + ) -> kb_config::WsEndpointConfig { + return kb_config::WsEndpointConfig { + name: name.to_string(), + enabled: true, + provider: "test".to_string(), + cluster: "devnet".to_string(), + url: format!("wss://{name}.invalid"), + connect_timeout_ms: 100, + request_timeout_ms: 100, + unsubscribe_timeout_ms: 100, + write_channel_capacity: 8, + event_channel_capacity: 16, + auto_reconnect: false, + roles: std::vec![role_config(role, request_kinds)], + }; + } + + fn client(endpoint: kb_config::WsEndpointConfig) -> crate::WsClient { + match crate::WsClient::new(endpoint) { + std::result::Result::Ok(client) => return client, + std::result::Result::Err(error) => panic!("client creation failed: {error}"), + } + } + + #[test] + fn new_rejects_empty_pool() { + let result = crate::WsEndpointPool::new(std::vec::Vec::new()); + assert!(result.is_err()); + } + + #[test] + fn snapshot_lists_every_client() { + let pool = match crate::WsEndpointPool::new(std::vec![ + client(endpoint("a", "slot_notifications", std::vec!["slot_subscribe".to_string()])), + client(endpoint("b", "slot_notifications", std::vec!["slot_subscribe".to_string()])), + ]) { + std::result::Result::Ok(pool) => pool, + std::result::Result::Err(error) => panic!("pool creation failed: {error}"), + }; + let snapshot = pool.snapshot(); + assert_eq!(snapshot.len(), 2); + assert_eq!(snapshot[0].endpoint_name, "a"); + assert_eq!(snapshot[1].endpoint_name, "b"); + } + + #[test] + fn select_client_round_robins_matching_clients() { + let pool = match crate::WsEndpointPool::new(std::vec![ + client(endpoint("a", "slot_notifications", std::vec!["slot_subscribe".to_string()])), + client(endpoint("b", "slot_notifications", std::vec!["slot_subscribe".to_string()])), + ]) { + std::result::Result::Ok(pool) => pool, + std::result::Result::Err(error) => panic!("pool creation failed: {error}"), + }; + let first = + match pool.select_client_for_role_and_method("slot_notifications", "slotSubscribe") { + std::result::Result::Ok(client) => client, + std::result::Result::Err(error) => panic!("selection failed: {error}"), + }; + let second = + match pool.select_client_for_role_and_method("slot_notifications", "slotSubscribe") { + std::result::Result::Ok(client) => client, + std::result::Result::Err(error) => panic!("selection failed: {error}"), + }; + assert_eq!(first.endpoint_name(), "a"); + assert_eq!(second.endpoint_name(), "b"); + } + + #[test] + fn select_client_skips_unsupported_clients() { + let pool = match crate::WsEndpointPool::new(std::vec![ + client(endpoint("a", "program_subscribe", std::vec!["program_subscribe".to_string()])), + client(endpoint("b", "slot_notifications", std::vec!["slot_subscribe".to_string()])), + ]) { + std::result::Result::Ok(pool) => pool, + std::result::Result::Err(error) => panic!("pool creation failed: {error}"), + }; + let selected = + match pool.select_client_for_role_and_method("slot_notifications", "slotSubscribe") { + std::result::Result::Ok(client) => client, + std::result::Result::Err(error) => panic!("selection failed: {error}"), + }; + assert_eq!(selected.endpoint_name(), "b"); + } + + #[test] + fn standard_subscription_selection_uses_the_subscribe_method() { + let pool = match crate::WsEndpointPool::new(std::vec![client(endpoint( + "a", + "slot_notifications", + std::vec!["slot_subscribe".to_string()], + ))]) { + std::result::Result::Ok(pool) => pool, + std::result::Result::Err(error) => panic!("pool creation failed: {error}"), + }; + let subscription = match crate::standard_ws_subscription("slotSubscribe") { + std::option::Option::Some(value) => value, + std::option::Option::None => panic!("slot subscription missing"), + }; + let selected = match pool + .select_client_for_standard_subscription("slot_notifications", subscription) + { + std::result::Result::Ok(client) => client, + std::result::Result::Err(error) => panic!("selection failed: {error}"), + }; + assert_eq!(selected.endpoint_name(), "a"); + } + + #[test] + fn select_client_returns_error_for_missing_role() { + let pool = match crate::WsEndpointPool::new(std::vec![client(endpoint( + "a", + "slot_notifications", + std::vec!["slot_subscribe".to_string()] + )),]) + { + std::result::Result::Ok(pool) => pool, + std::result::Result::Err(error) => panic!("pool creation failed: {error}"), + }; + let selected = + pool.select_client_for_role_and_kind("program_subscribe", "program_subscribe"); + assert!(selected.is_err()); + } +} diff --git a/kb-onchain-transport/src/ws_session.rs b/kb-onchain-transport/src/ws_session.rs new file mode 100644 index 0000000..a2d13bb --- /dev/null +++ b/kb-onchain-transport/src/ws_session.rs @@ -0,0 +1,1672 @@ +// file: kb-onchain-transport/src/ws_session.rs +// version: 5 + +//! Persistent multiplexed WebSocket session with bounded reconnect and resubscription. + +use futures_util::SinkExt; // rust-rules: trait-import +use futures_util::StreamExt; // rust-rules: trait-import + +/// Bounded reconnect policy for a persistent WebSocket session. +#[derive(Clone, Copy, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct WsReconnectPolicy { + /// Whether reconnect and resubscription are enabled. + pub enabled: bool, + /// Maximum connection attempts after one transport failure. + pub max_attempts: u32, + /// Delay before the first reconnect attempt. + pub initial_delay_ms: u64, + /// Maximum exponential-backoff delay. + pub max_delay_ms: u64, +} + +impl crate::WsReconnectPolicy { + /// Returns a policy that never reconnects. + pub const fn disabled() -> Self { + return Self { + enabled: false, + max_attempts: 0, + initial_delay_ms: 0, + max_delay_ms: 0, + }; + } + + /// Creates a bounded reconnect policy. + pub fn bounded( + max_attempts: u32, + initial_delay_ms: u64, + max_delay_ms: u64, + ) -> kb_core::Result { + if max_attempts == 0 { + return std::result::Result::Err(kb_core::Error::config( + "WebSocket reconnect max attempts must be greater than zero", + )); + } + if initial_delay_ms == 0 { + return std::result::Result::Err(kb_core::Error::config( + "WebSocket reconnect initial delay must be greater than zero", + )); + } + if max_delay_ms < initial_delay_ms { + return std::result::Result::Err(kb_core::Error::config( + "WebSocket reconnect maximum delay must not be smaller than the initial delay", + )); + } + return std::result::Result::Ok(Self { + enabled: true, + max_attempts, + initial_delay_ms, + max_delay_ms, + }); + } + + fn delay_for_attempt(&self, attempt: u32) -> std::time::Duration { + let exponent = attempt.saturating_sub(1).min(31); + let multiplier = match 1_u64.checked_shl(exponent) { + std::option::Option::Some(multiplier) => multiplier, + std::option::Option::None => u64::MAX, + }; + let delay = self.initial_delay_ms.saturating_mul(multiplier).min(self.max_delay_ms); + return std::time::Duration::from_millis(delay); + } +} + +/// Observable lifecycle state of a persistent WebSocket session. +#[derive(Clone, Copy, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)] +#[serde(rename_all = "snake_case")] +pub enum WsSessionState { + /// Socket is connected and accepts commands. + Connected, + /// Socket failed and a bounded reconnect is running. + Reconnecting, + /// Session is permanently disconnected. + Disconnected, +} + +/// One active standard subscription in a persistent session. +#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct WsSubscriptionSnapshot { + /// Stable local identifier retained across reconnects. + pub local_subscription_id: u64, + /// Current server identifier, absent while resubscription is pending. + pub remote_subscription_id: std::option::Option, + /// Exact subscribe method. + pub subscribe_method: std::string::String, + /// Exact unsubscribe method. + pub unsubscribe_method: std::string::String, + /// Exact notification method. + pub notification_method: std::string::String, +} + +/// Current snapshot of one persistent WebSocket session. +#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct WsSessionSnapshot { + /// Endpoint name. + pub endpoint_name: std::string::String, + /// Provider name. + pub provider: std::string::String, + /// Endpoint URL. + pub endpoint_url: std::string::String, + /// Current lifecycle state. + pub state: crate::WsSessionState, + /// Number of successful reconnects since session creation. + pub reconnect_count: u32, + /// Effective standard WebSocket capabilities after runtime probing. + pub capabilities: crate::StandardWsCapabilities, + /// Active subscriptions ordered by stable local identifier. + pub subscriptions: std::vec::Vec, +} + +/// Successful subscription acknowledgement. +#[derive(Clone, Debug, PartialEq)] +pub struct WsSubscriptionAck { + /// JSON-RPC response received from the node. + pub response: crate::JsonRpcResponse, + /// Registered subscription snapshot. + pub subscription: crate::WsSubscriptionSnapshot, +} + +/// Successful explicit unsubscribe acknowledgement. +#[derive(Clone, Debug, PartialEq)] +pub struct WsUnsubscribeAck { + /// JSON-RPC response received from the node. + pub response: crate::JsonRpcResponse, + /// Removed subscription snapshot. + pub subscription: crate::WsSubscriptionSnapshot, +} + +/// Event emitted by a persistent WebSocket session. +#[derive(Clone, Debug, PartialEq)] +pub enum WsSessionEvent { + /// Initial socket connection is ready. + Connected, + /// Reconnect attempt started. + Reconnecting { + /// One-based attempt number. + attempt: u32, + /// Configured attempt limit. + maximum_attempts: u32, + }, + /// Socket reconnected and active subscriptions are being restored. + Reconnected { + /// Successful reconnect count. + reconnect_count: u32, + }, + /// New subscription was acknowledged. + SubscriptionAdded(crate::WsSubscriptionSnapshot), + /// Existing subscription received a new remote identifier after reconnect. + SubscriptionRemapped { + /// Stable local identifier. + local_subscription_id: u64, + /// Previous remote identifier. + previous_remote_subscription_id: std::option::Option, + /// New remote identifier. + remote_subscription_id: u64, + }, + /// Subscription was removed explicitly or completed by a one-shot server contract. + SubscriptionRemoved(crate::WsSubscriptionSnapshot), + /// Typed standard notification. + Notification { + /// Subscription that received the notification. + subscription: crate::WsSubscriptionSnapshot, + /// Typed notification payload. + notification: std::boxed::Box, + /// Original JSON-RPC notification envelope. + raw: crate::JsonRpcNotification, + }, + /// Protocol or transport diagnostic. + Diagnostic { + /// Stable error family or diagnostic code. + code: std::string::String, + /// Human-readable diagnostic. + message: std::string::String, + }, + /// Session ended and will not reconnect. + Disconnected, +} + +/// Persistent multiplexed standard Solana WebSocket session. +pub struct WsSession { + endpoint: kb_config::WsEndpointConfig, + capabilities: std::sync::Arc>, + control_sender: tokio::sync::mpsc::Sender, + event_sender: tokio::sync::broadcast::Sender, + snapshot: std::sync::Arc>, +} + +impl std::fmt::Debug for crate::WsSession { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + return formatter + .debug_struct("WsSession") + .field("endpoint_name", &self.endpoint.name) + .field("provider", &self.endpoint.provider) + .field("endpoint_url", &self.endpoint.url) + .finish_non_exhaustive(); + } +} + +impl crate::WsSession { + /// Connects one persistent session and starts its multiplexing task. + /// + /// Enabled unstable methods are probed by their first real subscription attempt. When the + /// node explicitly rejects one as absent or disabled, that capability is disabled for the + /// remaining lifetime of this session. + pub async fn connect( + client: crate::WsClient, + capabilities: crate::StandardWsCapabilities, + reconnect_policy: crate::WsReconnectPolicy, + ) -> kb_core::Result { + let stream = match connect_stream(&client).await { + std::result::Result::Ok(stream) => stream, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let endpoint = client.endpoint_config().clone(); + let channel_capacity = match usize::try_from(endpoint.write_channel_capacity) { + std::result::Result::Ok(capacity) if capacity > 0 => capacity, + _ => 1, + }; + let event_capacity = match usize::try_from(endpoint.event_channel_capacity) { + std::result::Result::Ok(capacity) if capacity > 0 => capacity, + _ => 1, + }; + let (control_sender, control_receiver) = tokio::sync::mpsc::channel(channel_capacity); + let (event_sender, _event_receiver) = tokio::sync::broadcast::channel(event_capacity); + let initial_capabilities = capabilities; + let capabilities = std::sync::Arc::new(tokio::sync::RwLock::new(initial_capabilities)); + let snapshot = std::sync::Arc::new(tokio::sync::RwLock::new(crate::WsSessionSnapshot { + endpoint_name: endpoint.name.clone(), + provider: endpoint.provider.clone(), + endpoint_url: endpoint.url.clone(), + state: crate::WsSessionState::Connected, + reconnect_count: 0, + capabilities: initial_capabilities, + subscriptions: std::vec::Vec::new(), + })); + let runtime = WsSessionRuntime { + client, + reconnect_policy, + capabilities: capabilities.clone(), + control_receiver, + event_sender: event_sender.clone(), + snapshot: snapshot.clone(), + active: std::collections::BTreeMap::new(), + remote_to_local: std::collections::BTreeMap::new(), + pending: std::collections::BTreeMap::new(), + next_local_subscription_id: 1, + reconnect_count: 0, + }; + tokio::spawn(async move { + runtime.run(stream).await; + }); + let _send_result = event_sender.send(crate::WsSessionEvent::Connected); + tracing::info!(target: crate::TRACING_TARGET, action = "connect_ws_session", endpoint_name = %endpoint.name, provider = %endpoint.provider, "persistent WebSocket session connected"); + return std::result::Result::Ok(Self { + endpoint, + capabilities, + control_sender, + event_sender, + snapshot, + }); + } + + /// Returns a receiver for session events. + pub fn subscribe_events(&self) -> tokio::sync::broadcast::Receiver { + return self.event_sender.subscribe(); + } + + /// Returns the current session snapshot. + pub async fn snapshot(&self) -> crate::WsSessionSnapshot { + return self.snapshot.read().await.clone(); + } + + /// Subscribes one typed standard request on the existing socket. + /// + /// For an enabled unstable method, the first request also verifies actual node support. + pub async fn subscribe( + &self, + request: crate::StandardWsRequest, + ) -> kb_core::Result { + let capabilities = *self.capabilities.read().await; + let params = match request.params(capabilities) { + std::result::Result::Ok(params) => params, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let (response_sender, response_receiver) = tokio::sync::oneshot::channel(); + let command = WsSessionCommand::Subscribe { request, params, response_sender }; + let send_result = self.control_sender.send(command).await; + if send_result.is_err() { + return std::result::Result::Err(kb_core::Error::not_connected(format!( + "WebSocket session '{}' is not accepting subscriptions", + self.endpoint.name + ))); + } + return match response_receiver.await { + std::result::Result::Ok(result) => result, + std::result::Result::Err(error) => { + std::result::Result::Err(kb_core::Error::not_connected(format!( + "WebSocket subscription response channel closed for endpoint '{}': {error}", + self.endpoint.name + ))) + }, + }; + } + + /// Explicitly unsubscribes one current remote subscription identifier. + pub async fn unsubscribe( + &self, + remote_subscription_id: u64, + ) -> kb_core::Result { + let (response_sender, response_receiver) = tokio::sync::oneshot::channel(); + let command = WsSessionCommand::Unsubscribe { remote_subscription_id, response_sender }; + let send_result = self.control_sender.send(command).await; + if send_result.is_err() { + return std::result::Result::Err(kb_core::Error::not_connected(format!( + "WebSocket session '{}' is not accepting unsubscribe commands", + self.endpoint.name + ))); + } + return match response_receiver.await { + std::result::Result::Ok(result) => result, + std::result::Result::Err(error) => { + std::result::Result::Err(kb_core::Error::not_connected(format!( + "WebSocket unsubscribe response channel closed for endpoint '{}': {error}", + self.endpoint.name + ))) + }, + }; + } + + /// Unsubscribes active subscriptions, closes the socket and stops the session task. + pub async fn disconnect(&self) -> kb_core::Result<()> { + let (response_sender, response_receiver) = tokio::sync::oneshot::channel(); + let send_result = + self.control_sender.send(WsSessionCommand::Disconnect { response_sender }).await; + if send_result.is_err() { + return std::result::Result::Ok(()); + } + return match response_receiver.await { + std::result::Result::Ok(result) => result, + std::result::Result::Err(_) => std::result::Result::Ok(()), + }; + } +} + +type WsStream = + tokio_tungstenite::WebSocketStream>; + +enum WsSessionCommand { + Subscribe { + request: crate::StandardWsRequest, + params: std::vec::Vec, + response_sender: tokio::sync::oneshot::Sender>, + }, + Unsubscribe { + remote_subscription_id: u64, + response_sender: tokio::sync::oneshot::Sender>, + }, + Disconnect { + response_sender: tokio::sync::oneshot::Sender>, + }, +} + +struct ActiveWsSubscription { + request: crate::StandardWsRequest, + params: std::vec::Vec, + snapshot: crate::WsSubscriptionSnapshot, +} + +enum PendingWsRequest { + Subscribe { + local_subscription_id: u64, + request: crate::StandardWsRequest, + params: std::vec::Vec, + deadline: tokio::time::Instant, + response_sender: tokio::sync::oneshot::Sender>, + }, + Unsubscribe { + local_subscription_id: u64, + remote_subscription_id: u64, + deadline: tokio::time::Instant, + response_sender: tokio::sync::oneshot::Sender>, + }, + Resubscribe { + local_subscription_id: u64, + previous_remote_subscription_id: std::option::Option, + deadline: tokio::time::Instant, + }, +} + +struct WsSessionRuntime { + client: crate::WsClient, + reconnect_policy: crate::WsReconnectPolicy, + capabilities: std::sync::Arc>, + control_receiver: tokio::sync::mpsc::Receiver, + event_sender: tokio::sync::broadcast::Sender, + snapshot: std::sync::Arc>, + active: std::collections::BTreeMap, + remote_to_local: std::collections::BTreeMap, + pending: std::collections::BTreeMap, + next_local_subscription_id: u64, + reconnect_count: u32, +} + +impl WsSessionRuntime { + async fn run(mut self, mut stream: WsStream) { + let mut timeout_tick = tokio::time::interval(std::time::Duration::from_millis(50)); + timeout_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + tokio::select! { + command = self.control_receiver.recv() => { + let should_continue = match command { + std::option::Option::Some(command) => self.handle_command(&mut stream, command).await, + std::option::Option::None => false, + }; + if !should_continue { + break; + } + }, + message = stream.next() => { + let transport_ok = self.handle_stream_message(&mut stream, message).await; + if !transport_ok { + let reconnect = self.reconnect().await; + match reconnect { + std::option::Option::Some(reconnected_stream) => { + stream = reconnected_stream; + }, + std::option::Option::None => break, + } + } + }, + _ = timeout_tick.tick() => { + self.expire_pending().await; + }, + } + } + self.finish_disconnected().await; + } + + async fn handle_command(&mut self, stream: &mut WsStream, command: WsSessionCommand) -> bool { + return match command { + WsSessionCommand::Subscribe { request, params, response_sender } => { + self.send_subscribe(stream, request, params, response_sender).await; + true + }, + WsSessionCommand::Unsubscribe { remote_subscription_id, response_sender } => { + self.send_unsubscribe(stream, remote_subscription_id, response_sender).await; + true + }, + WsSessionCommand::Disconnect { response_sender } => { + let result = self.close_stream(stream).await; + let _send_result = response_sender.send(result); + false + }, + }; + } + + async fn send_subscribe( + &mut self, + stream: &mut WsStream, + request: crate::StandardWsRequest, + params: std::vec::Vec, + response_sender: tokio::sync::oneshot::Sender>, + ) { + let local_subscription_id = self.next_local_subscription_id; + self.next_local_subscription_id = self.next_local_subscription_id.saturating_add(1); + let json_request = self + .client + .build_json_rpc_request(request.subscribe_method().to_string(), params.clone()); + let request_id = match json_request.id.as_u64() { + std::option::Option::Some(request_id) => request_id, + std::option::Option::None => { + let _send_result = response_sender.send(std::result::Result::Err( + kb_core::Error::invalid_state("numeric WebSocket request id is required"), + )); + return; + }, + }; + let text = match json_request.to_json_string() { + std::result::Result::Ok(text) => text, + std::result::Result::Err(error) => { + let _send_result = response_sender.send(std::result::Result::Err(error)); + return; + }, + }; + let send_result = + stream.send(tokio_tungstenite::tungstenite::Message::Text(text.into())).await; + if let std::result::Result::Err(error) = send_result { + let _send_result = response_sender.send(std::result::Result::Err(kb_core::Error::ws( + format!("cannot send WebSocket subscribe request: {error}"), + ))); + return; + } + self.pending.insert( + request_id, + PendingWsRequest::Subscribe { + local_subscription_id, + request, + params, + deadline: tokio::time::Instant::now() + + std::time::Duration::from_millis( + self.client.endpoint_config().request_timeout_ms, + ), + response_sender, + }, + ); + } + + async fn send_unsubscribe( + &mut self, + stream: &mut WsStream, + remote_subscription_id: u64, + response_sender: tokio::sync::oneshot::Sender>, + ) { + let local_subscription_id = match self.remote_to_local.get(&remote_subscription_id) { + std::option::Option::Some(local_subscription_id) => *local_subscription_id, + std::option::Option::None => { + let _send_result = response_sender.send(std::result::Result::Err( + kb_core::Error::invalid_state(format!( + "unknown remote WebSocket subscription id {remote_subscription_id}" + )), + )); + return; + }, + }; + let subscription = match self.active.get(&local_subscription_id) { + std::option::Option::Some(subscription) => subscription, + std::option::Option::None => { + let _send_result = + response_sender.send(std::result::Result::Err(kb_core::Error::invalid_state( + format!("missing local WebSocket subscription id {local_subscription_id}"), + ))); + return; + }, + }; + let json_request = self.client.build_json_rpc_request( + subscription.snapshot.unsubscribe_method.clone(), + std::vec![serde_json::Value::from(remote_subscription_id)], + ); + let request_id = match json_request.id.as_u64() { + std::option::Option::Some(request_id) => request_id, + std::option::Option::None => { + let _send_result = response_sender.send(std::result::Result::Err( + kb_core::Error::invalid_state("numeric WebSocket request id is required"), + )); + return; + }, + }; + let text = match json_request.to_json_string() { + std::result::Result::Ok(text) => text, + std::result::Result::Err(error) => { + let _send_result = response_sender.send(std::result::Result::Err(error)); + return; + }, + }; + let send_result = + stream.send(tokio_tungstenite::tungstenite::Message::Text(text.into())).await; + if let std::result::Result::Err(error) = send_result { + let _send_result = response_sender.send(std::result::Result::Err(kb_core::Error::ws( + format!("cannot send WebSocket unsubscribe request: {error}"), + ))); + return; + } + self.pending.insert( + request_id, + PendingWsRequest::Unsubscribe { + local_subscription_id, + remote_subscription_id, + deadline: tokio::time::Instant::now() + + std::time::Duration::from_millis( + self.client.endpoint_config().unsubscribe_timeout_ms, + ), + response_sender, + }, + ); + } + + async fn handle_stream_message( + &mut self, + stream: &mut WsStream, + message: std::option::Option< + std::result::Result< + tokio_tungstenite::tungstenite::Message, + tokio_tungstenite::tungstenite::Error, + >, + >, + ) -> bool { + let message = match message { + std::option::Option::Some(std::result::Result::Ok(message)) => message, + std::option::Option::Some(std::result::Result::Err(error)) => { + self.emit_diagnostic("ws_read", format!("WebSocket read failed: {error}")); + return false; + }, + std::option::Option::None => { + self.emit_diagnostic("ws_closed", "WebSocket stream ended".to_string()); + return false; + }, + }; + return match message { + tokio_tungstenite::tungstenite::Message::Text(text) => { + self.handle_text(text.as_str()).await; + true + }, + tokio_tungstenite::tungstenite::Message::Ping(payload) => { + let pong_result = + stream.send(tokio_tungstenite::tungstenite::Message::Pong(payload)).await; + if let std::result::Result::Err(error) = pong_result { + self.emit_diagnostic("ws_pong", format!("WebSocket pong failed: {error}")); + return false; + } + true + }, + tokio_tungstenite::tungstenite::Message::Close(_) => false, + tokio_tungstenite::tungstenite::Message::Binary(_) + | tokio_tungstenite::tungstenite::Message::Pong(_) + | tokio_tungstenite::tungstenite::Message::Frame(_) => true, + }; + } + + async fn handle_text(&mut self, text: &str) { + let response = match crate::parse_json_rpc_text(text) { + std::result::Result::Ok(response) => response, + std::result::Result::Err(error) => { + self.emit_error(error); + return; + }, + }; + match response { + crate::JsonRpcResponse::Success(success) => { + self.handle_success(success).await; + }, + crate::JsonRpcResponse::Error(error_response) => { + self.handle_error_response(error_response).await; + }, + crate::JsonRpcResponse::Notification(notification) => { + self.handle_notification(notification).await; + }, + } + } + + async fn handle_success(&mut self, success: crate::JsonRpcSuccessResponse) { + let request_id = match success.id.as_u64() { + std::option::Option::Some(request_id) => request_id, + std::option::Option::None => { + self.emit_diagnostic( + "ws_response_id", + "WebSocket success response has no numeric request id".to_string(), + ); + return; + }, + }; + let pending = match self.pending.remove(&request_id) { + std::option::Option::Some(pending) => pending, + std::option::Option::None => { + self.emit_diagnostic( + "ws_unmatched_response", + format!("unmatched WebSocket response id {request_id}"), + ); + return; + }, + }; + let response = crate::JsonRpcResponse::Success(success.clone()); + match pending { + PendingWsRequest::Subscribe { + local_subscription_id, + request, + params, + deadline: _, + response_sender, + } => { + let remote_subscription_id = match success.result.as_u64() { + std::option::Option::Some(value) => value, + std::option::Option::None => { + let _send_result = response_sender.send(std::result::Result::Err( + kb_core::Error::json("subscribe result must be a numeric id"), + )); + return; + }, + }; + let specification = request.specification(); + let snapshot = crate::WsSubscriptionSnapshot { + local_subscription_id, + remote_subscription_id: std::option::Option::Some(remote_subscription_id), + subscribe_method: specification.subscribe_method.to_string(), + unsubscribe_method: specification.unsubscribe_method.to_string(), + notification_method: specification.notification_method.to_string(), + }; + self.remote_to_local.insert(remote_subscription_id, local_subscription_id); + self.active.insert( + local_subscription_id, + ActiveWsSubscription { + request, + params, + snapshot: snapshot.clone(), + }, + ); + self.refresh_snapshot().await; + let _event_result = self + .event_sender + .send(crate::WsSessionEvent::SubscriptionAdded(snapshot.clone())); + let _send_result = + response_sender.send(std::result::Result::Ok(crate::WsSubscriptionAck { + response, + subscription: snapshot, + })); + }, + PendingWsRequest::Unsubscribe { + local_subscription_id, + remote_subscription_id, + deadline: _, + response_sender, + } => { + if success.result.as_bool() != std::option::Option::Some(true) { + let _send_result = response_sender.send(std::result::Result::Err( + kb_core::Error::json("unsubscribe result must be true"), + )); + return; + } + self.remote_to_local.remove(&remote_subscription_id); + let removed = self.active.remove(&local_subscription_id); + let snapshot = match removed { + std::option::Option::Some(subscription) => subscription.snapshot, + std::option::Option::None => { + let _send_result = response_sender.send(std::result::Result::Err( + kb_core::Error::invalid_state( + "acknowledged subscription disappeared before removal", + ), + )); + return; + }, + }; + self.refresh_snapshot().await; + let _event_result = self + .event_sender + .send(crate::WsSessionEvent::SubscriptionRemoved(snapshot.clone())); + let _send_result = + response_sender.send(std::result::Result::Ok(WsUnsubscribeAck { + response, + subscription: snapshot, + })); + }, + PendingWsRequest::Resubscribe { + local_subscription_id, + previous_remote_subscription_id, + deadline: _, + } => { + let remote_subscription_id = match success.result.as_u64() { + std::option::Option::Some(value) => value, + std::option::Option::None => { + self.emit_diagnostic( + "ws_resubscribe_result", + "resubscribe result must be a numeric id".to_string(), + ); + return; + }, + }; + let remapped = match self.active.get_mut(&local_subscription_id) { + std::option::Option::Some(active) => { + active.snapshot.remote_subscription_id = + std::option::Option::Some(remote_subscription_id); + true + }, + std::option::Option::None => false, + }; + if remapped { + self.remote_to_local.insert(remote_subscription_id, local_subscription_id); + let _event_result = + self.event_sender.send(crate::WsSessionEvent::SubscriptionRemapped { + local_subscription_id, + previous_remote_subscription_id, + remote_subscription_id, + }); + self.refresh_snapshot().await; + } + }, + } + } + + async fn handle_error_response(&mut self, error_response: crate::JsonRpcErrorResponse) { + let request_id = error_response.id.as_u64(); + let error = kb_core::Error::ws(format!( + "WebSocket JSON-RPC error {}: {}", + error_response.error.code, error_response.error.message + )); + if let std::option::Option::Some(request_id) = request_id { + if let std::option::Option::Some(pending) = self.pending.remove(&request_id) { + match pending { + PendingWsRequest::Subscribe { request, response_sender, .. } => { + self.disable_unsupported_capability(&request, &error_response).await; + let _send_result = response_sender.send(std::result::Result::Err(error)); + }, + PendingWsRequest::Unsubscribe { response_sender, .. } => { + let _send_result = response_sender.send(std::result::Result::Err(error)); + }, + PendingWsRequest::Resubscribe { local_subscription_id, .. } => { + let request = self + .active + .get(&local_subscription_id) + .map(|active| return active.request.clone()); + let disabled = match request { + std::option::Option::Some(request) => { + self.disable_unsupported_capability(&request, &error_response).await + }, + std::option::Option::None => false, + }; + if disabled { + let removed = self.active.remove(&local_subscription_id); + if let std::option::Option::Some(removed) = removed { + self.refresh_snapshot().await; + let _event_result = self.event_sender.send( + crate::WsSessionEvent::SubscriptionRemoved(removed.snapshot), + ); + } + } else { + self.emit_error(error); + } + }, + } + return; + } + } + self.emit_error(error); + } + + async fn disable_unsupported_capability( + &mut self, + request: &crate::StandardWsRequest, + error_response: &crate::JsonRpcErrorResponse, + ) -> bool { + let method = request.subscribe_method(); + if !crate::StandardWsCapabilities::is_unstable_method(method) + || !is_method_unavailable_error(error_response) + { + return false; + } + let disabled = { + let mut capabilities = self.capabilities.write().await; + capabilities.disable_method(method) + }; + if disabled { + self.refresh_snapshot().await; + self.emit_diagnostic( + "ws_capability_disabled", + format!( + "endpoint '{}' rejected unstable method '{method}'; capability disabled for this session: JSON-RPC {} {}", + self.client.endpoint_name(), + error_response.error.code, + error_response.error.message + ), + ); + } + return true; + } + + async fn handle_notification(&mut self, notification: crate::JsonRpcNotification) { + let remote_subscription_id = notification.params.subscription; + let local_subscription_id = match self.remote_to_local.get(&remote_subscription_id) { + std::option::Option::Some(value) => *value, + std::option::Option::None => { + self.emit_diagnostic( + "ws_unknown_subscription", + format!( + "notification references unknown remote subscription id {remote_subscription_id}" + ), + ); + return; + }, + }; + let subscription = match self.active.get(&local_subscription_id) { + std::option::Option::Some(active) => active.snapshot.clone(), + std::option::Option::None => return, + }; + if notification.method != subscription.notification_method { + self.emit_diagnostic( + "ws_notification_method", + format!( + "notification method '{}' does not match expected '{}'", + notification.method, subscription.notification_method + ), + ); + return; + } + let typed = match crate::adapt_standard_ws_notification(¬ification) { + std::result::Result::Ok(typed) => typed, + std::result::Result::Err(error) => { + self.emit_error(error); + return; + }, + }; + let terminal_signature = matches!( + &typed, + crate::StandardWsNotification::Signature(crate::WsSignatureNotification { + result: crate::RpcResponse { + value: crate::WsSignatureValue::Status { .. }, + .. + }, + .. + }) + ); + let _event_result = self.event_sender.send(crate::WsSessionEvent::Notification { + subscription: subscription.clone(), + notification: std::boxed::Box::new(typed), + raw: notification, + }); + if terminal_signature { + self.remote_to_local.remove(&remote_subscription_id); + self.active.remove(&local_subscription_id); + self.refresh_snapshot().await; + let _event_result = + self.event_sender.send(crate::WsSessionEvent::SubscriptionRemoved(subscription)); + } + } + + async fn expire_pending(&mut self) { + let now = tokio::time::Instant::now(); + let mut expired = std::vec::Vec::new(); + for (request_id, pending) in &self.pending { + let deadline = match pending { + PendingWsRequest::Subscribe { deadline, .. } + | PendingWsRequest::Unsubscribe { deadline, .. } + | PendingWsRequest::Resubscribe { deadline, .. } => *deadline, + }; + if deadline <= now { + expired.push(*request_id); + } + } + return for request_id in expired { + let pending = self.pending.remove(&request_id); + if let std::option::Option::Some(pending) = pending { + let error = + kb_core::Error::ws(format!("WebSocket request id {request_id} timed out")); + match pending { + PendingWsRequest::Subscribe { response_sender, .. } => { + let _send_result = response_sender.send(std::result::Result::Err(error)); + }, + PendingWsRequest::Unsubscribe { response_sender, .. } => { + let _send_result = response_sender.send(std::result::Result::Err(error)); + }, + PendingWsRequest::Resubscribe { .. } => { + self.emit_error(error); + }, + } + } + }; + } + + async fn reconnect(&mut self) -> std::option::Option { + self.fail_pending(kb_core::Error::ws( + "WebSocket transport failed before request completion", + )); + if !self.reconnect_policy.enabled { + return std::option::Option::None; + } + let mut previous_remote_ids = std::collections::BTreeMap::new(); + self.remote_to_local.clear(); + for (local_subscription_id, active) in &mut self.active { + previous_remote_ids + .insert(*local_subscription_id, active.snapshot.remote_subscription_id); + active.snapshot.remote_subscription_id = std::option::Option::None; + } + self.set_state(crate::WsSessionState::Reconnecting).await; + let mut attempt = 1_u32; + while attempt <= self.reconnect_policy.max_attempts { + let _event_result = self.event_sender.send(crate::WsSessionEvent::Reconnecting { + attempt, + maximum_attempts: self.reconnect_policy.max_attempts, + }); + tokio::time::sleep(self.reconnect_policy.delay_for_attempt(attempt)).await; + let stream_result = connect_stream(&self.client).await; + match stream_result { + std::result::Result::Ok(mut stream) => { + self.reconnect_count = self.reconnect_count.saturating_add(1); + self.set_state(crate::WsSessionState::Connected).await; + let _event_result = + self.event_sender.send(crate::WsSessionEvent::Reconnected { + reconnect_count: self.reconnect_count, + }); + let restore_result = + self.restore_subscriptions(&mut stream, &previous_remote_ids).await; + if let std::result::Result::Err(error) = restore_result { + self.emit_error(error); + attempt = attempt.saturating_add(1); + continue; + } + return std::option::Option::Some(stream); + }, + std::result::Result::Err(error) => self.emit_error(error), + } + attempt = attempt.saturating_add(1); + } + return std::option::Option::None; + } + + async fn restore_subscriptions( + &mut self, + stream: &mut WsStream, + previous_remote_ids: &std::collections::BTreeMap>, + ) -> kb_core::Result<()> { + let local_ids = self.active.keys().copied().collect::>(); + for local_subscription_id in local_ids { + let (subscribe_method, params, previous_remote_subscription_id) = + match self.active.get(&local_subscription_id) { + std::option::Option::Some(active) => ( + active.request.subscribe_method().to_string(), + active.params.clone(), + previous_remote_ids.get(&local_subscription_id).copied().flatten(), + ), + std::option::Option::None => continue, + }; + let json_request = self.client.build_json_rpc_request(subscribe_method, params); + let request_id = match json_request.id.as_u64() { + std::option::Option::Some(value) => value, + std::option::Option::None => { + return std::result::Result::Err(kb_core::Error::invalid_state( + "numeric WebSocket request id is required", + )); + }, + }; + let text = match json_request.to_json_string() { + std::result::Result::Ok(text) => text, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let send_result = + stream.send(tokio_tungstenite::tungstenite::Message::Text(text.into())).await; + if let std::result::Result::Err(error) = send_result { + return std::result::Result::Err(kb_core::Error::ws(format!( + "cannot restore WebSocket subscription: {error}" + ))); + } + self.pending.insert( + request_id, + PendingWsRequest::Resubscribe { + local_subscription_id, + previous_remote_subscription_id, + deadline: tokio::time::Instant::now() + + std::time::Duration::from_millis( + self.client.endpoint_config().request_timeout_ms, + ), + }, + ); + } + return std::result::Result::Ok(()); + } + + fn fail_pending(&mut self, error: kb_core::Error) { + let pending = std::mem::take(&mut self.pending); + for (_request_id, request) in pending { + match request { + PendingWsRequest::Subscribe { response_sender, .. } => { + let _send_result = + response_sender.send(std::result::Result::Err(error.clone())); + }, + PendingWsRequest::Unsubscribe { response_sender, .. } => { + let _send_result = + response_sender.send(std::result::Result::Err(error.clone())); + }, + PendingWsRequest::Resubscribe { .. } => {}, + } + } + } + + async fn close_stream(&mut self, stream: &mut WsStream) -> kb_core::Result<()> { + let subscriptions = self + .active + .values() + .filter_map(|active| { + return active.snapshot.remote_subscription_id.map(|remote_id| { + return (active.snapshot.unsubscribe_method.clone(), remote_id); + }); + }) + .collect::>(); + for (method, remote_id) in subscriptions { + let request = self + .client + .build_json_rpc_request(method, std::vec![serde_json::Value::from(remote_id)]); + let text = match request.to_json_string() { + std::result::Result::Ok(text) => text, + std::result::Result::Err(error) => { + self.emit_error(error); + continue; + }, + }; + let send_result = + stream.send(tokio_tungstenite::tungstenite::Message::Text(text.into())).await; + if let std::result::Result::Err(error) = send_result { + self.emit_diagnostic( + "ws_disconnect_unsubscribe", + format!("cannot send disconnect unsubscribe: {error}"), + ); + break; + } + } + let close_result = stream + .send(tokio_tungstenite::tungstenite::Message::Close(std::option::Option::None)) + .await; + return match close_result { + std::result::Result::Ok(()) => std::result::Result::Ok(()), + std::result::Result::Err(error) => std::result::Result::Err(kb_core::Error::ws( + format!("cannot close WebSocket session: {error}"), + )), + }; + } + + async fn refresh_snapshot(&self) { + let capabilities = *self.capabilities.read().await; + let mut snapshot = self.snapshot.write().await; + snapshot.reconnect_count = self.reconnect_count; + snapshot.capabilities = capabilities; + snapshot.subscriptions = + self.active.values().map(|active| return active.snapshot.clone()).collect(); + } + + async fn set_state(&self, state: crate::WsSessionState) { + let capabilities = *self.capabilities.read().await; + let mut snapshot = self.snapshot.write().await; + snapshot.state = state; + snapshot.reconnect_count = self.reconnect_count; + snapshot.capabilities = capabilities; + snapshot.subscriptions = + self.active.values().map(|active| return active.snapshot.clone()).collect(); + } + + async fn finish_disconnected(&mut self) { + self.fail_pending(kb_core::Error::not_connected( + "persistent WebSocket session disconnected", + )); + self.set_state(crate::WsSessionState::Disconnected).await; + let _event_result = self.event_sender.send(crate::WsSessionEvent::Disconnected); + tracing::info!(target: crate::TRACING_TARGET, action = "disconnect_ws_session", endpoint_name = %self.client.endpoint_name(), provider = %self.client.provider(), "persistent WebSocket session disconnected"); + } + + fn emit_error(&self, error: kb_core::Error) { + self.emit_diagnostic(error.code(), error.message().to_string()); + } + + fn emit_diagnostic(&self, code: &str, message: std::string::String) { + tracing::warn!(target: crate::TRACING_TARGET, action = "ws_session_diagnostic", endpoint_name = %self.client.endpoint_name(), provider = %self.client.provider(), diagnostic_code = code, diagnostic = %message, "persistent WebSocket session diagnostic"); + let _event_result = self + .event_sender + .send(crate::WsSessionEvent::Diagnostic { code: code.to_string(), message }); + } +} + +fn is_method_unavailable_error(error_response: &crate::JsonRpcErrorResponse) -> bool { + if error_response.error.code == -32601 { + return true; + } + let message = error_response.error.message.to_ascii_lowercase(); + return message.contains("method not found") + || message.contains("method is not available") + || message.contains("method not available") + || message.contains("method is not enabled") + || message.contains("method not enabled") + || message.contains("unsupported method") + || message.contains("unstable method"); +} + +async fn connect_stream(client: &crate::WsClient) -> kb_core::Result { + let connect_future = tokio_tungstenite::connect_async(client.endpoint_url()); + let timeout = std::time::Duration::from_millis(client.endpoint_config().connect_timeout_ms); + let timeout_result = tokio::time::timeout(timeout, connect_future).await; + let connect_result = match timeout_result { + std::result::Result::Ok(connect_result) => connect_result, + std::result::Result::Err(_) => { + return std::result::Result::Err(kb_core::Error::ws(format!( + "WebSocket connect timed out for endpoint '{}'", + client.endpoint_name() + ))); + }, + }; + return match connect_result { + std::result::Result::Ok((stream, _response)) => std::result::Result::Ok(stream), + std::result::Result::Err(error) => std::result::Result::Err(kb_core::Error::ws(format!( + "cannot connect WebSocket endpoint '{}': {error}", + client.endpoint_name() + ))), + }; +} + +#[cfg(test)] +mod tests { + use futures_util::SinkExt; // rust-rules: trait-import + use futures_util::StreamExt; // rust-rules: trait-import + + fn endpoint(url: std::string::String) -> kb_config::WsEndpointConfig { + return kb_config::WsEndpointConfig { + name: "local-test".to_string(), + enabled: true, + provider: "offline".to_string(), + cluster: "localnet".to_string(), + url, + connect_timeout_ms: 1_000, + request_timeout_ms: 1_000, + unsubscribe_timeout_ms: 1_000, + write_channel_capacity: 16, + event_channel_capacity: 32, + auto_reconnect: true, + roles: std::vec![kb_config::EndpointRoleConfig { + role: "slot_notifications".to_string(), + enabled: true, + request_kinds: std::vec!["slot_subscribe".to_string()], + priority: 1, + requests_per_second: 100, + burst_capacity: 100, + max_concurrent_requests: 8, + max_subscriptions: 32, + pause_after_rate_limit_ms: 10, + }], + }; + } + + async fn next_text( + stream: &mut tokio_tungstenite::WebSocketStream, + ) -> serde_json::Value { + let message = match stream.next().await { + std::option::Option::Some(std::result::Result::Ok(message)) => message, + std::option::Option::Some(std::result::Result::Err(error)) => { + panic!("server read failed: {error}") + }, + std::option::Option::None => panic!("server stream ended"), + }; + let text = match message { + tokio_tungstenite::tungstenite::Message::Text(text) => text, + other => panic!("expected text, got {other:?}"), + }; + return match serde_json::from_str(text.as_str()) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("server json failed: {error}"), + }; + } + + async fn send_json( + stream: &mut tokio_tungstenite::WebSocketStream, + value: serde_json::Value, + ) { + let text = match serde_json::to_string(&value) { + std::result::Result::Ok(text) => text, + std::result::Result::Err(error) => panic!("server serialization failed: {error}"), + }; + if let std::result::Result::Err(error) = + stream.send(tokio_tungstenite::tungstenite::Message::Text(text.into())).await + { + panic!("server send failed: {error}"); + } + } + + #[test] + fn reconnect_policy_is_bounded_and_validated() { + assert!(crate::WsReconnectPolicy::bounded(0, 100, 1000).is_err()); + assert!(crate::WsReconnectPolicy::bounded(3, 0, 1000).is_err()); + assert!(crate::WsReconnectPolicy::bounded(3, 1000, 100).is_err()); + let policy = match crate::WsReconnectPolicy::bounded(3, 100, 250) { + std::result::Result::Ok(policy) => policy, + std::result::Result::Err(error) => panic!("policy failed: {error}"), + }; + assert_eq!(policy.delay_for_attempt(1), std::time::Duration::from_millis(100)); + assert_eq!(policy.delay_for_attempt(2), std::time::Duration::from_millis(200)); + assert_eq!(policy.delay_for_attempt(3), std::time::Duration::from_millis(250)); + } + + #[test] + fn disabled_policy_never_reconnects() { + assert_eq!( + crate::WsReconnectPolicy::disabled(), + crate::WsReconnectPolicy { + enabled: false, + max_attempts: 0, + initial_delay_ms: 0, + max_delay_ms: 0, + } + ); + } + + #[tokio::test] + async fn persistent_session_multiplexes_notification_and_explicit_unsubscribe() { + let listener = match tokio::net::TcpListener::bind("127.0.0.1:0").await { + std::result::Result::Ok(listener) => listener, + std::result::Result::Err(error) => panic!("listener failed: {error}"), + }; + let address = match listener.local_addr() { + std::result::Result::Ok(address) => address, + std::result::Result::Err(error) => panic!("local address failed: {error}"), + }; + let server = tokio::spawn(async move { + let (socket, _peer) = match listener.accept().await { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("accept failed: {error}"), + }; + let mut stream = match tokio_tungstenite::accept_async(socket).await { + std::result::Result::Ok(stream) => stream, + std::result::Result::Err(error) => panic!("handshake failed: {error}"), + }; + let subscribe = next_text(&mut stream).await; + assert_eq!(subscribe["method"], "slotSubscribe"); + send_json( + &mut stream, + serde_json::json!({ "jsonrpc": "2.0", "result": 42, "id": subscribe["id"].clone() }), + ) + .await; + send_json( + &mut stream, + serde_json::json!({ + "jsonrpc": "2.0", + "method": "slotNotification", + "params": { + "result": { "slot": 8, "parent": 7, "root": 6 }, + "subscription": 42 + } + }), + ) + .await; + let unsubscribe = next_text(&mut stream).await; + assert_eq!(unsubscribe["method"], "slotUnsubscribe"); + assert_eq!(unsubscribe["params"][0], 42); + send_json( + &mut stream, + serde_json::json!({ "jsonrpc": "2.0", "result": true, "id": unsubscribe["id"].clone() }), + ) + .await; + let close = match stream.next().await { + std::option::Option::Some(std::result::Result::Ok(message)) => message, + other => panic!("close read failed: {other:?}"), + }; + assert!(matches!(close, tokio_tungstenite::tungstenite::Message::Close(_))); + }); + let client = match crate::WsClient::new(endpoint(format!("ws://{address}"))) { + std::result::Result::Ok(client) => client, + std::result::Result::Err(error) => panic!("client failed: {error}"), + }; + let session = match crate::WsSession::connect( + client, + crate::StandardWsCapabilities::default(), + crate::WsReconnectPolicy::disabled(), + ) + .await + { + std::result::Result::Ok(session) => session, + std::result::Result::Err(error) => panic!("session failed: {error}"), + }; + let mut events = session.subscribe_events(); + let acknowledgement = match session + .subscribe(crate::StandardWsRequest::Slot(crate::SlotSubscribeRequest)) + .await + { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("subscribe failed: {error}"), + }; + assert_eq!(acknowledgement.subscription.remote_subscription_id, Some(42)); + let notification = tokio::time::timeout(std::time::Duration::from_secs(2), async { + loop { + let event = match events.recv().await { + std::result::Result::Ok(event) => event, + std::result::Result::Err(error) => panic!("event failed: {error}"), + }; + if let crate::WsSessionEvent::Notification { notification, .. } = event { + return notification; + } + } + }) + .await; + let notification = match notification { + std::result::Result::Ok(notification) => notification, + std::result::Result::Err(_) => panic!("notification timed out"), + }; + assert!(matches!(notification.as_ref(), crate::StandardWsNotification::Slot(_))); + let unsubscribe = match session.unsubscribe(42).await { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("unsubscribe failed: {error}"), + }; + assert_eq!(unsubscribe.subscription.remote_subscription_id, Some(42)); + assert!(session.disconnect().await.is_ok()); + if let std::result::Result::Err(error) = server.await { + panic!("server task failed: {error}"); + } + } + + #[tokio::test] + async fn unsupported_unstable_method_is_disabled_after_first_server_rejection() { + let listener = match tokio::net::TcpListener::bind("127.0.0.1:0").await { + std::result::Result::Ok(listener) => listener, + std::result::Result::Err(error) => panic!("listener failed: {error}"), + }; + let address = match listener.local_addr() { + std::result::Result::Ok(address) => address, + std::result::Result::Err(error) => panic!("local address failed: {error}"), + }; + let server = tokio::spawn(async move { + let (socket, _peer) = match listener.accept().await { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("accept failed: {error}"), + }; + let mut stream = match tokio_tungstenite::accept_async(socket).await { + std::result::Result::Ok(stream) => stream, + std::result::Result::Err(error) => panic!("handshake failed: {error}"), + }; + let subscribe = next_text(&mut stream).await; + assert_eq!(subscribe["method"], "blockSubscribe"); + send_json( + &mut stream, + serde_json::json!({ + "jsonrpc": "2.0", + "error": { + "code": -32601, + "message": "Method not found" + }, + "id": subscribe["id"].clone() + }), + ) + .await; + let close = match stream.next().await { + std::option::Option::Some(std::result::Result::Ok(message)) => message, + other => panic!("close read failed: {other:?}"), + }; + assert!(matches!(close, tokio_tungstenite::tungstenite::Message::Close(_))); + }); + let client = match crate::WsClient::new(endpoint(format!("ws://{address}"))) { + std::result::Result::Ok(client) => client, + std::result::Result::Err(error) => panic!("client failed: {error}"), + }; + let session = match crate::WsSession::connect( + client, + crate::StandardWsCapabilities { + block_subscribe: true, + slots_updates_subscribe: false, + vote_subscribe: false, + }, + crate::WsReconnectPolicy::disabled(), + ) + .await + { + std::result::Result::Ok(session) => session, + std::result::Result::Err(error) => panic!("session failed: {error}"), + }; + let request = crate::StandardWsRequest::Block(crate::BlockSubscribeRequest { + filter: crate::WsBlockFilter::All, + config: std::option::Option::None, + }); + assert!(session.subscribe(request.clone()).await.is_err()); + let snapshot = session.snapshot().await; + assert!(!snapshot.capabilities.block_subscribe); + assert!(session.subscribe(request).await.is_err()); + assert!(session.disconnect().await.is_ok()); + if let std::result::Result::Err(error) = server.await { + panic!("server task failed: {error}"); + } + } + + #[tokio::test] + async fn terminal_signature_notification_removes_one_shot_subscription() { + let listener = match tokio::net::TcpListener::bind("127.0.0.1:0").await { + std::result::Result::Ok(listener) => listener, + std::result::Result::Err(error) => panic!("listener failed: {error}"), + }; + let address = match listener.local_addr() { + std::result::Result::Ok(address) => address, + std::result::Result::Err(error) => panic!("local address failed: {error}"), + }; + let server = tokio::spawn(async move { + let (socket, _peer) = match listener.accept().await { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("accept failed: {error}"), + }; + let mut stream = match tokio_tungstenite::accept_async(socket).await { + std::result::Result::Ok(stream) => stream, + std::result::Result::Err(error) => panic!("handshake failed: {error}"), + }; + let subscribe = next_text(&mut stream).await; + assert_eq!(subscribe["method"], "signatureSubscribe"); + send_json( + &mut stream, + serde_json::json!({ + "jsonrpc": "2.0", + "result": 77, + "id": subscribe["id"].clone() + }), + ) + .await; + send_json( + &mut stream, + serde_json::json!({ + "jsonrpc": "2.0", + "method": "signatureNotification", + "params": { + "result": { + "context": { "slot": 9 }, + "value": { "err": null } + }, + "subscription": 77 + } + }), + ) + .await; + let close = match stream.next().await { + std::option::Option::Some(std::result::Result::Ok(message)) => message, + other => panic!("close read failed: {other:?}"), + }; + assert!(matches!(close, tokio_tungstenite::tungstenite::Message::Close(_))); + }); + let client = match crate::WsClient::new(endpoint(format!("ws://{address}"))) { + std::result::Result::Ok(client) => client, + std::result::Result::Err(error) => panic!("client failed: {error}"), + }; + let session = match crate::WsSession::connect( + client, + crate::StandardWsCapabilities::default(), + crate::WsReconnectPolicy::disabled(), + ) + .await + { + std::result::Result::Ok(session) => session, + std::result::Result::Err(error) => panic!("session failed: {error}"), + }; + let mut events = session.subscribe_events(); + let acknowledgement = match session + .subscribe(crate::StandardWsRequest::Signature(crate::SignatureSubscribeRequest { + signature: "1111111111111111111111111111111111111111111111111111111111111111" + .to_string(), + config: std::option::Option::None, + })) + .await + { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("subscribe failed: {error}"), + }; + assert_eq!(acknowledgement.subscription.remote_subscription_id, Some(77)); + let removed = tokio::time::timeout(std::time::Duration::from_secs(2), async { + loop { + let event = match events.recv().await { + std::result::Result::Ok(event) => event, + std::result::Result::Err(error) => panic!("event failed: {error}"), + }; + if let crate::WsSessionEvent::SubscriptionRemoved(subscription) = event { + return subscription; + } + } + }) + .await; + let removed = match removed { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => panic!("terminal removal timed out"), + }; + assert_eq!(removed.remote_subscription_id, Some(77)); + assert!(session.snapshot().await.subscriptions.is_empty()); + assert!(session.disconnect().await.is_ok()); + if let std::result::Result::Err(error) = server.await { + panic!("server task failed: {error}"); + } + } + + #[tokio::test] + async fn reconnect_restores_subscription_with_new_remote_id() { + let listener = match tokio::net::TcpListener::bind("127.0.0.1:0").await { + std::result::Result::Ok(listener) => listener, + std::result::Result::Err(error) => panic!("listener failed: {error}"), + }; + let address = match listener.local_addr() { + std::result::Result::Ok(address) => address, + std::result::Result::Err(error) => panic!("local address failed: {error}"), + }; + let server = tokio::spawn(async move { + let (first_socket, _peer) = match listener.accept().await { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("first accept failed: {error}"), + }; + let mut first = match tokio_tungstenite::accept_async(first_socket).await { + std::result::Result::Ok(stream) => stream, + std::result::Result::Err(error) => panic!("first handshake failed: {error}"), + }; + let subscribe = next_text(&mut first).await; + send_json( + &mut first, + serde_json::json!({ "jsonrpc": "2.0", "result": 10, "id": subscribe["id"].clone() }), + ) + .await; + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + if let std::result::Result::Err(error) = + first.send(tokio_tungstenite::tungstenite::Message::Close(None)).await + { + panic!("first close failed: {error}"); + } + let (second_socket, _peer) = match listener.accept().await { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("second accept failed: {error}"), + }; + let mut second = match tokio_tungstenite::accept_async(second_socket).await { + std::result::Result::Ok(stream) => stream, + std::result::Result::Err(error) => panic!("second handshake failed: {error}"), + }; + let resubscribe = next_text(&mut second).await; + assert_eq!(resubscribe["method"], "slotSubscribe"); + send_json( + &mut second, + serde_json::json!({ "jsonrpc": "2.0", "result": 20, "id": resubscribe["id"].clone() }), + ) + .await; + let disconnect_unsubscribe = next_text(&mut second).await; + assert_eq!(disconnect_unsubscribe["method"], "slotUnsubscribe"); + assert_eq!(disconnect_unsubscribe["params"][0], 20); + let close = match second.next().await { + std::option::Option::Some(std::result::Result::Ok(message)) => message, + other => panic!("second close read failed: {other:?}"), + }; + assert!(matches!(close, tokio_tungstenite::tungstenite::Message::Close(_))); + }); + let client = match crate::WsClient::new(endpoint(format!("ws://{address}"))) { + std::result::Result::Ok(client) => client, + std::result::Result::Err(error) => panic!("client failed: {error}"), + }; + let policy = match crate::WsReconnectPolicy::bounded(2, 10, 20) { + std::result::Result::Ok(policy) => policy, + std::result::Result::Err(error) => panic!("policy failed: {error}"), + }; + let session = match crate::WsSession::connect( + client, + crate::StandardWsCapabilities::default(), + policy, + ) + .await + { + std::result::Result::Ok(session) => session, + std::result::Result::Err(error) => panic!("session failed: {error}"), + }; + let mut events = session.subscribe_events(); + let acknowledgement = match session + .subscribe(crate::StandardWsRequest::Slot(crate::SlotSubscribeRequest)) + .await + { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("subscribe failed: {error}"), + }; + assert_eq!(acknowledgement.subscription.remote_subscription_id, Some(10)); + let remapped = tokio::time::timeout(std::time::Duration::from_secs(3), async { + loop { + let event = match events.recv().await { + std::result::Result::Ok(event) => event, + std::result::Result::Err(error) => panic!("event failed: {error}"), + }; + if let crate::WsSessionEvent::SubscriptionRemapped { + previous_remote_subscription_id, + remote_subscription_id, + .. + } = event + { + return (previous_remote_subscription_id, remote_subscription_id); + } + } + }) + .await; + let remapped = match remapped { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => panic!("remap timed out"), + }; + assert_eq!(remapped, (Some(10), 20)); + let snapshot = session.snapshot().await; + assert_eq!(snapshot.reconnect_count, 1); + assert_eq!(snapshot.subscriptions[0].remote_subscription_id, Some(20)); + assert!(session.disconnect().await.is_ok()); + if let std::result::Result::Err(error) = server.await { + panic!("server task failed: {error}"); + } + } +}