// file: crates/ksp-onchain-transport-lib/src/ws_settings.rs // version: 6 const DEFAULT_WS_CLOSE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5); const DEFAULT_WS_COMMAND_QUEUE_CAPACITY: usize = 128; const DEFAULT_WS_COMMAND_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); const DEFAULT_WS_MAX_ACTIVE_SUBSCRIPTIONS: usize = 1_024; const DEFAULT_WS_MAX_FRAME_SIZE_BYTES: usize = 16 * 1024 * 1024; const DEFAULT_WS_MAX_MESSAGE_SIZE_BYTES: usize = 64 * 1024 * 1024; const DEFAULT_WS_MAX_PENDING_REQUESTS: usize = 128; const DEFAULT_WS_MAX_WRITE_BUFFER_SIZE_BYTES: usize = 1024 * 1024; const DEFAULT_WS_NOTIFICATION_QUEUE_CAPACITY: usize = 256; const DEFAULT_WS_RECONNECT_INITIAL_BACKOFF: std::time::Duration = std::time::Duration::from_millis(250); const DEFAULT_WS_RECONNECT_MAX_BACKOFF: std::time::Duration = std::time::Duration::from_secs(5); const DEFAULT_WS_RECONNECT_MAX_RETRIES: u32 = 5; /// Runtime WebSocket endpoint URL owned by Transport. /// /// The actual URL can contain provider credentials. Its [`std::fmt::Debug`] implementation is intentionally redacted. #[derive(Clone, Eq, PartialEq)] pub struct WsEndpointUrl { value: std::string::String, } impl WsEndpointUrl { /// Parses and validates one WebSocket endpoint URL. pub fn parse(value: impl std::convert::Into) -> ksp_core_lib::Result { ksp_logging_lib::trace!(target: crate::TRACING_TARGET, "validating WebSocket endpoint URL"); let value = value.into(); let parsed = match reqwest::Url::parse(value.as_str()) { std::result::Result::Ok(parsed) => parsed, std::result::Result::Err(error) => { ksp_logging_lib::warn!(target: crate::TRACING_TARGET, field = "ws_endpoints.url", "rejected invalid WebSocket endpoint URL"); return std::result::Result::Err( ksp_core_lib::Error::new(crate::ERROR_CODE_INVALID_SETTINGS, "WebSocket endpoint URL is invalid") .with_context("field", "ws_endpoints.url") .with_source(error), ); }, }; if parsed.scheme() != "ws" && parsed.scheme() != "wss" { ksp_logging_lib::warn!( target: crate::TRACING_TARGET, field = "ws_endpoints.url", scheme = parsed.scheme(), "rejected WebSocket endpoint URL with unsupported scheme" ); return std::result::Result::Err( ksp_core_lib::Error::new(crate::ERROR_CODE_INVALID_SETTINGS, "WebSocket endpoint URL must use ws or wss") .with_context("field", "ws_endpoints.url") .with_context("scheme", parsed.scheme()), ); } if parsed.host_str().is_none() { ksp_logging_lib::warn!(target: crate::TRACING_TARGET, field = "ws_endpoints.url", "rejected WebSocket endpoint URL without host"); return std::result::Result::Err( ksp_core_lib::Error::new(crate::ERROR_CODE_INVALID_SETTINGS, "WebSocket endpoint URL must contain a host") .with_context("field", "ws_endpoints.url"), ); } ksp_logging_lib::trace!(target: crate::TRACING_TARGET, scheme = parsed.scheme(), "validated WebSocket endpoint URL syntax"); return std::result::Result::Ok(Self { value }); } /// Returns the sensitive runtime URL text. /// /// Callers must not write this value to logs, generic diagnostics or snapshots. #[must_use] pub fn as_str(&self) -> &str { return self.value.as_str(); } } impl std::fmt::Debug for WsEndpointUrl { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { return formatter.write_str("WsEndpointUrl()"); } } /// Open provider descriptor used by WebSocket endpoint settings. #[derive(Clone, Debug, Eq, Hash, PartialEq)] pub struct WsProviderName { value: std::string::String, } impl WsProviderName { /// Creates an open provider descriptor. Validation is performed by [`WsTransportSettings::validate`]. #[must_use] pub fn new(value: impl std::convert::Into) -> Self { return Self { value: value.into() }; } /// Returns the provider descriptor text. #[must_use] pub fn as_str(&self) -> &str { return self.value.as_str(); } } /// Open cluster or network descriptor used by WebSocket endpoint settings. #[derive(Clone, Debug, Eq, Hash, PartialEq)] pub struct WsClusterName { value: std::string::String, } impl WsClusterName { /// Creates an open cluster descriptor. Validation is performed by [`WsTransportSettings::validate`]. #[must_use] pub fn new(value: impl std::convert::Into) -> Self { return Self { value: value.into() }; } /// Returns the cluster descriptor text. #[must_use] pub fn as_str(&self) -> &str { return self.value.as_str(); } } /// WebSocket protocol family understood by KSP Transport. /// /// The protocol discriminator belongs specifically to the WebSocket runtime. Provider products using another transport, including a future Helius /// LaserStream gRPC backend, require a distinct transport-owned descriptor instead of reusing this enum. #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] #[non_exhaustive] pub enum WsProtocolKind { /// Standard Solana JSON-RPC WebSocket PubSub. SolanaStandard, /// Helius LaserStream WebSocket protocol surface. HeliusLaserStream, } impl WsProtocolKind { /// Returns the stable KSP descriptor for this protocol family. #[must_use] pub const fn as_str(self) -> &'static str { return match self { Self::SolanaStandard => "solana_standard", Self::HeliusLaserStream => "helius_laserstream", }; } } /// Bounded reconnect settings owned by the WebSocket transport runtime. #[derive(Clone, Debug, Eq, PartialEq)] pub struct WsReconnectSettings { max_retries: u32, initial_backoff: std::time::Duration, max_backoff: std::time::Duration, } impl WsReconnectSettings { /// Creates bounded reconnect settings. #[must_use] pub const fn new(max_retries: u32, initial_backoff: std::time::Duration, max_backoff: std::time::Duration) -> Self { return Self { max_retries, initial_backoff, max_backoff }; } /// Returns the number of reconnect attempts allowed after the connection is lost. #[must_use] pub const fn max_retries(&self) -> u32 { return self.max_retries; } /// Returns the initial reconnect backoff. #[must_use] pub const fn initial_backoff(&self) -> std::time::Duration { return self.initial_backoff; } /// Returns the maximum reconnect backoff. #[must_use] pub const fn max_backoff(&self) -> std::time::Duration { return self.max_backoff; } } impl std::default::Default for WsReconnectSettings { fn default() -> Self { return Self::new(DEFAULT_WS_RECONNECT_MAX_RETRIES, DEFAULT_WS_RECONNECT_INITIAL_BACKOFF, DEFAULT_WS_RECONNECT_MAX_BACKOFF); } } /// Policy controlling whether logical subscriptions are restored after a successful reconnect. #[derive(Clone, Copy, Debug, Default, Eq, Hash, PartialEq)] pub enum WsResubscribePolicy { /// Never restore subscriptions automatically after the physical connection is replaced. Never, /// Restore subscriptions that are still logically desired when reconnect completes. #[default] ActiveSubscriptions, } /// Runtime limits and lifecycle settings for one physical WebSocket session. #[derive(Clone, Debug, Eq, PartialEq)] pub struct WsSessionSettings { command_timeout: std::time::Duration, close_timeout: std::time::Duration, reconnect: crate::WsReconnectSettings, resubscribe: crate::WsResubscribePolicy, command_queue_capacity: usize, notification_queue_capacity: usize, max_active_subscriptions: usize, max_pending_requests: usize, max_message_size_bytes: usize, max_frame_size_bytes: usize, max_write_buffer_size_bytes: usize, } impl WsSessionSettings { /// Creates complete runtime settings for one physical WebSocket session. #[must_use] #[allow(clippy::too_many_arguments)] pub const fn new( command_timeout: std::time::Duration, close_timeout: std::time::Duration, reconnect: crate::WsReconnectSettings, resubscribe: crate::WsResubscribePolicy, command_queue_capacity: usize, notification_queue_capacity: usize, max_active_subscriptions: usize, max_pending_requests: usize, max_message_size_bytes: usize, max_frame_size_bytes: usize, max_write_buffer_size_bytes: usize, ) -> Self { return Self { command_timeout, close_timeout, reconnect, resubscribe, command_queue_capacity, notification_queue_capacity, max_active_subscriptions, max_pending_requests, max_message_size_bytes, max_frame_size_bytes, max_write_buffer_size_bytes, }; } /// Returns the deadline applied to bounded session commands and JSON-RPC control requests. #[must_use] pub const fn command_timeout(&self) -> std::time::Duration { return self.command_timeout; } /// Returns the total bounded close/shutdown deadline. #[must_use] pub const fn close_timeout(&self) -> std::time::Duration { return self.close_timeout; } /// Returns the reconnect policy. #[must_use] pub const fn reconnect(&self) -> &crate::WsReconnectSettings { return &self.reconnect; } /// Returns the resubscribe policy. #[must_use] pub const fn resubscribe(&self) -> crate::WsResubscribePolicy { return self.resubscribe; } /// Returns the bounded session-command queue capacity. #[must_use] pub const fn command_queue_capacity(&self) -> usize { return self.command_queue_capacity; } /// Returns the bounded notification queue capacity allocated per logical subscription. #[must_use] pub const fn notification_queue_capacity(&self) -> usize { return self.notification_queue_capacity; } /// Returns the maximum number of logical subscriptions allowed on one physical session. #[must_use] pub const fn max_active_subscriptions(&self) -> usize { return self.max_active_subscriptions; } /// Returns the maximum number of JSON-RPC requests allowed to await responses concurrently. #[must_use] pub const fn max_pending_requests(&self) -> usize { return self.max_pending_requests; } /// Returns the maximum accepted complete WebSocket message size in bytes. #[must_use] pub const fn max_message_size_bytes(&self) -> usize { return self.max_message_size_bytes; } /// Returns the maximum accepted WebSocket frame size in bytes. #[must_use] pub const fn max_frame_size_bytes(&self) -> usize { return self.max_frame_size_bytes; } /// Returns the maximum WebSocket write-buffer size in bytes. #[must_use] pub const fn max_write_buffer_size_bytes(&self) -> usize { return self.max_write_buffer_size_bytes; } /// Validates runtime bounds without reading Config or environment state. pub fn validate(&self) -> ksp_core_lib::Result<()> { ksp_logging_lib::trace!(target: crate::TRACING_TARGET, "validating WebSocket session settings"); if self.command_timeout.is_zero() { return ws_invalid_settings("WebSocket command timeout must be greater than zero", "ws_session.command_timeout"); } if self.close_timeout.is_zero() { return ws_invalid_settings("WebSocket close timeout must be greater than zero", "ws_session.close_timeout"); } if self.reconnect.initial_backoff().is_zero() { return ws_invalid_settings("initial WebSocket reconnect backoff must be greater than zero", "ws_session.reconnect.initial_backoff"); } if self.reconnect.max_backoff().is_zero() { return ws_invalid_settings("maximum WebSocket reconnect backoff must be greater than zero", "ws_session.reconnect.max_backoff"); } if self.reconnect.max_backoff() < self.reconnect.initial_backoff() { return ws_invalid_settings( "maximum WebSocket reconnect backoff must not be lower than initial reconnect backoff", "ws_session.reconnect.max_backoff", ); } if let std::result::Result::Err(error) = validate_non_zero_bound(self.command_queue_capacity, "ws_session.command_queue_capacity") { return std::result::Result::Err(error); } if let std::result::Result::Err(error) = validate_non_zero_bound(self.notification_queue_capacity, "ws_session.notification_queue_capacity") { return std::result::Result::Err(error); } if let std::result::Result::Err(error) = validate_non_zero_bound(self.max_active_subscriptions, "ws_session.max_active_subscriptions") { return std::result::Result::Err(error); } if let std::result::Result::Err(error) = validate_non_zero_bound(self.max_pending_requests, "ws_session.max_pending_requests") { return std::result::Result::Err(error); } if let std::result::Result::Err(error) = validate_non_zero_bound(self.max_message_size_bytes, "ws_session.max_message_size_bytes") { return std::result::Result::Err(error); } if let std::result::Result::Err(error) = validate_non_zero_bound(self.max_frame_size_bytes, "ws_session.max_frame_size_bytes") { return std::result::Result::Err(error); } if let std::result::Result::Err(error) = validate_non_zero_bound(self.max_write_buffer_size_bytes, "ws_session.max_write_buffer_size_bytes") { return std::result::Result::Err(error); } ksp_logging_lib::debug!( target: crate::TRACING_TARGET, command_queue_capacity = self.command_queue_capacity, notification_queue_capacity = self.notification_queue_capacity, max_active_subscriptions = self.max_active_subscriptions, max_pending_requests = self.max_pending_requests, max_message_size_bytes = self.max_message_size_bytes, max_frame_size_bytes = self.max_frame_size_bytes, max_write_buffer_size_bytes = self.max_write_buffer_size_bytes, reconnect_max_retries = self.reconnect.max_retries(), resubscribe = self.resubscribe.as_str(), "validated WebSocket session settings" ); return std::result::Result::Ok(()); } } impl std::default::Default for WsSessionSettings { fn default() -> Self { return Self::new( DEFAULT_WS_COMMAND_TIMEOUT, DEFAULT_WS_CLOSE_TIMEOUT, crate::WsReconnectSettings::default(), crate::WsResubscribePolicy::default(), DEFAULT_WS_COMMAND_QUEUE_CAPACITY, DEFAULT_WS_NOTIFICATION_QUEUE_CAPACITY, DEFAULT_WS_MAX_ACTIVE_SUBSCRIPTIONS, DEFAULT_WS_MAX_PENDING_REQUESTS, DEFAULT_WS_MAX_MESSAGE_SIZE_BYTES, DEFAULT_WS_MAX_FRAME_SIZE_BYTES, DEFAULT_WS_MAX_WRITE_BUFFER_SIZE_BYTES, ); } } impl WsResubscribePolicy { /// Returns the stable KSP descriptor for this policy. #[must_use] pub const fn as_str(self) -> &'static str { return match self { Self::Never => "never", Self::ActiveSubscriptions => "active_subscriptions", }; } } /// Runtime settings for one named WebSocket endpoint. #[derive(Clone, Debug, Eq, PartialEq)] pub struct WsEndpointSettings { name: std::string::String, enabled: bool, provider: crate::WsProviderName, cluster: crate::WsClusterName, protocol: crate::WsProtocolKind, subscription_capabilities: std::option::Option>, url: crate::WsEndpointUrl, session: crate::WsSessionSettings, } impl WsEndpointSettings { /// Creates explicit settings for one logical WebSocket endpoint. #[must_use] pub fn new( name: impl std::convert::Into, enabled: bool, provider: crate::WsProviderName, cluster: crate::WsClusterName, protocol: crate::WsProtocolKind, url: crate::WsEndpointUrl, session: crate::WsSessionSettings, ) -> Self { return Self { name: name.into(), enabled, provider, cluster, protocol, subscription_capabilities: std::option::Option::None, url, session, }; } /// Declares the exact WebSocket subscription families supported by this endpoint. /// /// An endpoint created without this builder remains valid for backward compatibility, but its capabilities are considered undeclared and therefore /// never constitute positive support evidence for a subscription family. Validation rejects an explicitly declared empty set, duplicates and /// capabilities that cannot belong to the selected protocol surface. #[must_use] pub fn with_subscription_capabilities(mut self, capabilities: std::vec::Vec) -> Self { self.subscription_capabilities = std::option::Option::Some(capabilities); return self; } /// Returns the explicitly declared WebSocket subscription capabilities, or `None` when the endpoint still uses the legacy undeclared form. #[must_use] pub fn subscription_capabilities(&self) -> std::option::Option<&[crate::WsSubscriptionKind]> { return self.subscription_capabilities.as_deref(); } /// Returns whether this endpoint explicitly declares support for one WebSocket subscription family. /// /// Undeclared legacy capability state is fail-closed and therefore returns `false`. #[must_use] pub fn supports_subscription(&self, kind: crate::WsSubscriptionKind) -> bool { return ws_protocol_supports_subscription(self.protocol, kind) && self.subscription_capabilities().is_some_and(|capabilities| return capabilities.contains(&kind)); } /// Returns the logical endpoint name. #[must_use] pub fn name(&self) -> &str { return self.name.as_str(); } /// Returns whether this endpoint can be used to create physical sessions. #[must_use] pub const fn enabled(&self) -> bool { return self.enabled; } /// Returns the open provider descriptor. #[must_use] pub const fn provider(&self) -> &crate::WsProviderName { return &self.provider; } /// Returns the open cluster descriptor. #[must_use] pub const fn cluster(&self) -> &crate::WsClusterName { return &self.cluster; } /// Returns the WebSocket protocol family. #[must_use] pub const fn protocol(&self) -> crate::WsProtocolKind { return self.protocol; } /// Returns the sensitive WebSocket endpoint URL wrapper. #[must_use] pub const fn url(&self) -> &crate::WsEndpointUrl { return &self.url; } /// Returns the effective settings applied to every physical session explicitly created from this endpoint. #[must_use] pub const fn session(&self) -> &crate::WsSessionSettings { return &self.session; } } /// Complete runtime settings consumed by the KSP WebSocket transport foundation. #[derive(Clone, Debug, Eq, PartialEq)] pub struct WsTransportSettings { endpoints: std::vec::Vec, } impl WsTransportSettings { /// Creates complete WebSocket transport runtime settings. #[must_use] pub fn new(endpoints: std::vec::Vec) -> Self { return Self { endpoints }; } /// Returns configured WebSocket endpoints in declaration order. #[must_use] pub fn endpoints(&self) -> &[crate::WsEndpointSettings] { return self.endpoints.as_slice(); } /// Validates structural runtime invariants without reading Config or environment state. pub fn validate(&self) -> ksp_core_lib::Result<()> { ksp_logging_lib::trace!(target: crate::TRACING_TARGET, endpoint_count = self.endpoints.len(), "validating WebSocket transport settings"); if self.endpoints.is_empty() { return ws_invalid_settings("at least one WebSocket endpoint must be configured", "ws_endpoints"); } let mut enabled_endpoint_count = 0_usize; for (endpoint_index, endpoint) in self.endpoints.iter().enumerate() { if let std::result::Result::Err(error) = validate_ws_endpoint(endpoint, endpoint_index) { return std::result::Result::Err(error); } if endpoint.enabled() { enabled_endpoint_count += 1; } for previous in &self.endpoints[..endpoint_index] { if previous.name() == endpoint.name() { return std::result::Result::Err( ksp_core_lib::Error::new(crate::ERROR_CODE_INVALID_SETTINGS, "WebSocket endpoint names must be unique") .with_context("field", format!("ws_endpoints[{endpoint_index}].name")) .with_context("endpoint_name", endpoint.name()), ); } } } if enabled_endpoint_count == 0 { return ws_invalid_settings("at least one WebSocket endpoint must be enabled", "ws_endpoints.enabled"); } ksp_logging_lib::debug!( target: crate::TRACING_TARGET, endpoint_count = self.endpoints.len(), enabled_endpoint_count, "validated WebSocket transport settings" ); return std::result::Result::Ok(()); } } fn validate_ws_endpoint(endpoint: &crate::WsEndpointSettings, endpoint_index: usize) -> ksp_core_lib::Result<()> { let name_field = format!("ws_endpoints[{endpoint_index}].name"); if let std::result::Result::Err(error) = validate_ws_descriptor(endpoint.name(), name_field.as_str()) { return std::result::Result::Err(error); } let provider_field = format!("ws_endpoints[{endpoint_index}].provider"); if let std::result::Result::Err(error) = validate_ws_descriptor(endpoint.provider().as_str(), provider_field.as_str()) { return std::result::Result::Err(error); } let cluster_field = format!("ws_endpoints[{endpoint_index}].cluster"); if let std::result::Result::Err(error) = validate_ws_descriptor(endpoint.cluster().as_str(), cluster_field.as_str()) { return std::result::Result::Err(error); } if let std::option::Option::Some(capabilities) = endpoint.subscription_capabilities() { if capabilities.is_empty() { return ws_invalid_settings( "explicit WebSocket subscription capabilities must not be empty", format!("ws_endpoints[{endpoint_index}].subscription_capabilities").as_str(), ); } for (capability_index, capability) in capabilities.iter().copied().enumerate() { let field = format!("ws_endpoints[{endpoint_index}].subscription_capabilities[{capability_index}]"); if !ws_protocol_supports_subscription(endpoint.protocol(), capability) { return std::result::Result::Err( ksp_core_lib::Error::new(crate::ERROR_CODE_INVALID_SETTINGS, "WebSocket subscription capability is incompatible with endpoint protocol") .with_context("field", field) .with_context("subscription_kind", capability.as_str()) .with_context("protocol", endpoint.protocol().as_str()), ); } if capabilities[..capability_index].contains(&capability) { return std::result::Result::Err( ksp_core_lib::Error::new(crate::ERROR_CODE_INVALID_SETTINGS, "WebSocket subscription capabilities must be unique") .with_context("field", field) .with_context("subscription_kind", capability.as_str()), ); } } } if let std::result::Result::Err(error) = endpoint.session().validate() { return std::result::Result::Err(error); } ksp_logging_lib::trace!( target: crate::TRACING_TARGET, endpoint_name = endpoint.name(), provider = endpoint.provider().as_str(), cluster = endpoint.cluster().as_str(), protocol = endpoint.protocol().as_str(), enabled = endpoint.enabled(), "validated WebSocket endpoint settings" ); return std::result::Result::Ok(()); } const fn ws_protocol_supports_subscription(protocol: crate::WsProtocolKind, kind: crate::WsSubscriptionKind) -> bool { return match protocol { crate::WsProtocolKind::SolanaStandard => matches!( kind, crate::WsSubscriptionKind::Account | crate::WsSubscriptionKind::Block | crate::WsSubscriptionKind::Logs | crate::WsSubscriptionKind::Program | crate::WsSubscriptionKind::Root | crate::WsSubscriptionKind::Signature | crate::WsSubscriptionKind::Slot | crate::WsSubscriptionKind::SlotsUpdates | crate::WsSubscriptionKind::Vote ), crate::WsProtocolKind::HeliusLaserStream => matches!( kind, crate::WsSubscriptionKind::Account | crate::WsSubscriptionKind::Logs | crate::WsSubscriptionKind::Program | crate::WsSubscriptionKind::Root | crate::WsSubscriptionKind::Signature | crate::WsSubscriptionKind::Slot | crate::WsSubscriptionKind::SlotsUpdates | crate::WsSubscriptionKind::HeliusTransaction ), }; } fn validate_ws_descriptor(value: &str, field: &str) -> ksp_core_lib::Result<()> { if value.trim().is_empty() { return ws_invalid_settings("WebSocket transport descriptor must not be empty", field); } if value.trim() != value { return ws_invalid_settings("WebSocket transport descriptor must not contain leading or trailing whitespace", field); } return std::result::Result::Ok(()); } fn validate_non_zero_bound(value: usize, field: &str) -> ksp_core_lib::Result<()> { if value == 0 { return ws_invalid_settings("WebSocket runtime bound must be greater than zero", field); } return std::result::Result::Ok(()); } fn ws_invalid_settings(message: &str, field: &str) -> ksp_core_lib::Result<()> { ksp_logging_lib::warn!(target: crate::TRACING_TARGET, field = field, reason = message, "rejected WebSocket transport settings"); return std::result::Result::Err(ksp_core_lib::Error::new(crate::ERROR_CODE_INVALID_SETTINGS, message).with_context("field", field)); } #[cfg(test)] #[path = "../unit_tests/ws_settings.rs"] mod tests;