v0.2.7-pre.002

This commit is contained in:
2026-08-22 16:39:13 +02:00
parent cf4b28df2b
commit b64a799c85
11 changed files with 1491 additions and 29 deletions

View File

@@ -1,5 +1,5 @@
// file: crates/ksp-onchain-transport-lib/src/constants.rs
// version: 1
// version: 2
//! Transport-owned tracing constants.

View File

@@ -1,5 +1,5 @@
// file: crates/ksp-onchain-transport-lib/src/lib.rs
// version: 21
// version: 22
#![warn(missing_docs)]
#![deny(unreachable_pub)]
@@ -16,6 +16,8 @@
//! modern/legacy `getTransaction` coverage. `0.2.4` completes the HTTP surface with all ten Blocks and five Economics wrappers, including complete
//! modern/legacy `getBlock`, positional inflation rewards, runtime-provided economics values and the final `KSP-TRANSPORT-007` compliance target.
//! The candidate surface therefore exposes typed wrappers for all 52 current audited Solana HTTP methods while retaining 14 removed historical descriptors.
//! `0.2.7-pre.002` adds the provider-neutral WebSocket settings foundation, redacted endpoint URLs, explicit protocol-family discrimination, local session/subscription
//! identities, observable lifecycle states and safe snapshots. Physical sockets and subscription execution are intentionally deferred to later prereleases.
mod client;
mod constants;
@@ -34,6 +36,8 @@ mod rpc_method;
mod rpc_tokens;
mod rpc_transactions;
mod settings;
mod ws_lifecycle;
mod ws_settings;
/// Passive runtime availability reported for one logical HTTP endpoint.
pub use self::client::HttpEndpointAvailability;
@@ -291,6 +295,38 @@ pub use self::settings::HttpRoleLimits;
pub use self::settings::HttpRoleName;
/// Complete runtime settings consumed by the Solana HTTP transport foundation.
pub use self::settings::HttpTransportSettings;
/// Stable local identity assigned to one physical WebSocket session.
pub use self::ws_lifecycle::WsSessionId;
/// Safe runtime snapshot for one physical WebSocket session.
pub use self::ws_lifecycle::WsSessionSnapshot;
/// Observable lifecycle state of one physical WebSocket session.
pub use self::ws_lifecycle::WsSessionState;
/// Stable local identity assigned to one logical WebSocket subscription.
pub use self::ws_lifecycle::WsSubscriptionId;
/// Standard Solana subscription family represented by one logical WebSocket subscription.
pub use self::ws_lifecycle::WsSubscriptionKind;
/// Safe lifecycle projection for one logical WebSocket subscription.
pub use self::ws_lifecycle::WsSubscriptionSnapshot;
/// Observable lifecycle state of one logical WebSocket subscription.
pub use self::ws_lifecycle::WsSubscriptionState;
/// Open cluster or network descriptor used by WebSocket endpoint settings.
pub use self::ws_settings::WsClusterName;
/// Runtime settings for one named WebSocket endpoint.
pub use self::ws_settings::WsEndpointSettings;
/// Runtime WebSocket endpoint URL with redacted diagnostics.
pub use self::ws_settings::WsEndpointUrl;
/// WebSocket protocol family understood by KSP Transport.
pub use self::ws_settings::WsProtocolKind;
/// Open provider descriptor used by WebSocket endpoint settings.
pub use self::ws_settings::WsProviderName;
/// Bounded reconnect settings owned by the WebSocket transport runtime.
pub use self::ws_settings::WsReconnectSettings;
/// Policy controlling logical resubscription after reconnect.
pub use self::ws_settings::WsResubscribePolicy;
/// Runtime limits and lifecycle settings for one physical WebSocket session.
pub use self::ws_settings::WsSessionSettings;
/// Complete runtime settings consumed by the KSP WebSocket transport foundation.
pub use self::ws_settings::WsTransportSettings;
/// Owning tracing target for events emitted by the on-chain transport crate.
pub(crate) use self::constants::TRACING_TARGET;

View File

@@ -0,0 +1,281 @@
// file: crates/ksp-onchain-transport-lib/src/ws_lifecycle.rs
// version: 1
/// Stable local identity assigned to one physical WebSocket session.
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct WsSessionId(std::num::NonZeroU64);
impl WsSessionId {
/// Creates a session identity from a non-zero local value.
#[must_use]
pub const fn new(value: std::num::NonZeroU64) -> Self {
return Self(value);
}
/// Returns the stable local numeric value.
#[must_use]
pub const fn get(self) -> u64 {
return self.0.get();
}
}
/// Stable local identity assigned to one logical WebSocket subscription.
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct WsSubscriptionId(std::num::NonZeroU64);
impl WsSubscriptionId {
/// Creates a subscription identity from a non-zero local value.
#[must_use]
pub const fn new(value: std::num::NonZeroU64) -> Self {
return Self(value);
}
/// Returns the stable local numeric value.
#[must_use]
pub const fn get(self) -> u64 {
return self.0.get();
}
}
/// Observable lifecycle state of one physical WebSocket session.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum WsSessionState {
/// No physical connection is currently active and no connection attempt is running.
Disconnected,
/// The actor is establishing the physical connection.
Connecting,
/// The physical connection is active.
Active,
/// The actor is reconnecting after an unexpected physical disconnect.
Reconnecting {
/// One-based reconnect attempt currently in progress or waiting for backoff.
attempt: u32,
},
/// Explicit shutdown has started and new subscriptions are refused.
Closing,
/// Explicit shutdown completed.
Closed,
/// The session reached a terminal failure state.
Failed,
}
/// Observable lifecycle state of one logical WebSocket subscription.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum WsSubscriptionState {
/// The local subscription exists but its initial subscribe request has not completed.
Requested,
/// The logical subscription is bound to an active remote subscription.
Active,
/// The logical subscription is being restored after reconnect.
Resubscribing,
/// Local cancellation has won and remote cleanup is in progress when possible.
Cancelling,
/// The logical subscription reached a non-error terminal state.
Closed,
/// The logical subscription reached a terminal failure state.
Failed,
}
/// Standard Solana subscription family represented by one logical WebSocket subscription.
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
#[non_exhaustive]
pub enum WsSubscriptionKind {
/// `accountSubscribe` family.
Account,
/// `blockSubscribe` family.
Block,
/// `logsSubscribe` family.
Logs,
/// `programSubscribe` family.
Program,
/// `rootSubscribe` family.
Root,
/// `signatureSubscribe` family.
Signature,
/// `slotSubscribe` family.
Slot,
/// `slotsUpdatesSubscribe` family.
SlotsUpdates,
/// `voteSubscribe` family.
Vote,
}
impl WsSubscriptionKind {
/// Returns the stable KSP descriptor for this standard subscription family.
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::Account => "account",
Self::Block => "block",
Self::Logs => "logs",
Self::Program => "program",
Self::Root => "root",
Self::Signature => "signature",
Self::Slot => "slot",
Self::SlotsUpdates => "slots_updates",
Self::Vote => "vote",
}
}
}
/// Safe lifecycle projection for one logical WebSocket subscription.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct WsSubscriptionSnapshot {
id: crate::WsSubscriptionId,
kind: crate::WsSubscriptionKind,
state: crate::WsSubscriptionState,
remote_bound: bool,
}
impl WsSubscriptionSnapshot {
/// Creates one safe subscription lifecycle projection for Transport runtime internals.
#[must_use]
pub(crate) const fn new(id: crate::WsSubscriptionId, kind: crate::WsSubscriptionKind, state: crate::WsSubscriptionState, remote_bound: bool) -> Self {
return Self { id, kind, state, remote_bound };
}
/// Returns the stable local subscription identity.
#[must_use]
pub const fn id(&self) -> crate::WsSubscriptionId {
return self.id;
}
/// Returns the standard subscription family.
#[must_use]
pub const fn kind(&self) -> crate::WsSubscriptionKind {
return self.kind;
}
/// Returns the current logical lifecycle state.
#[must_use]
pub const fn state(&self) -> crate::WsSubscriptionState {
return self.state;
}
/// Returns whether a current remote subscription ID is bound internally.
///
/// The remote ID itself is deliberately absent because it is ephemeral across reconnects.
#[must_use]
pub const fn remote_bound(&self) -> bool {
return self.remote_bound;
}
}
/// Safe runtime snapshot for one physical WebSocket session.
///
/// The snapshot deliberately contains logical endpoint metadata and local identities only. It never stores the endpoint URL, credentials, request payloads or
/// raw notifications.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct WsSessionSnapshot {
id: crate::WsSessionId,
endpoint_name: std::string::String,
provider: crate::WsProviderName,
cluster: crate::WsClusterName,
protocol: crate::WsProtocolKind,
state: crate::WsSessionState,
pending_request_count: usize,
continuity_gap_count: u64,
overflow_count: u64,
subscriptions: std::vec::Vec<crate::WsSubscriptionSnapshot>,
}
impl WsSessionSnapshot {
/// Creates one safe session projection for Transport runtime internals.
#[must_use]
#[allow(clippy::too_many_arguments)]
pub(crate) fn new(
id: crate::WsSessionId,
endpoint_name: impl std::convert::Into<std::string::String>,
provider: crate::WsProviderName,
cluster: crate::WsClusterName,
protocol: crate::WsProtocolKind,
state: crate::WsSessionState,
pending_request_count: usize,
continuity_gap_count: u64,
overflow_count: u64,
subscriptions: std::vec::Vec<crate::WsSubscriptionSnapshot>,
) -> Self {
return Self {
id,
endpoint_name: endpoint_name.into(),
provider,
cluster,
protocol,
state,
pending_request_count,
continuity_gap_count,
overflow_count,
subscriptions,
};
}
/// Returns the stable local session identity.
#[must_use]
pub const fn id(&self) -> crate::WsSessionId {
return self.id;
}
/// Returns the safe logical endpoint name.
#[must_use]
pub fn endpoint_name(&self) -> &str {
return self.endpoint_name.as_str();
}
/// Returns the provider descriptor without endpoint credentials.
#[must_use]
pub const fn provider(&self) -> &crate::WsProviderName {
return &self.provider;
}
/// Returns the 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 current physical session lifecycle state.
#[must_use]
pub const fn state(&self) -> crate::WsSessionState {
return self.state;
}
/// Returns the number of JSON-RPC requests currently awaiting responses.
#[must_use]
pub const fn pending_request_count(&self) -> usize {
return self.pending_request_count;
}
/// Returns the number of observed physical continuity gaps for this session.
#[must_use]
pub const fn continuity_gap_count(&self) -> u64 {
return self.continuity_gap_count;
}
/// Returns the cumulative number of notification queue overflows observed by this session.
#[must_use]
pub const fn overflow_count(&self) -> u64 {
return self.overflow_count;
}
/// Returns safe lifecycle projections for logical subscriptions owned by this session.
#[must_use]
pub fn subscriptions(&self) -> &[crate::WsSubscriptionSnapshot] {
return self.subscriptions.as_slice();
}
/// Returns the number of logical subscriptions currently projected by the session.
#[must_use]
pub fn subscription_count(&self) -> usize {
return self.subscriptions.len();
}
}
#[cfg(test)]
#[path = "../unit_tests/ws_lifecycle.rs"]
mod tests;

View File

@@ -0,0 +1,544 @@
// file: crates/ksp-onchain-transport-lib/src/ws_settings.rs
// version: 2
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<std::string::String>) -> ksp_core_lib::Result<Self> {
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(<redacted>)");
}
}
/// 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<std::string::String>) -> 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<std::string::String>) -> 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.
///
/// `0.2.7` exposes only standard Solana WebSocket. The non-exhaustive contract allows later provider-specific families without changing the common endpoint
/// container or injecting provider-only options into [`WsSessionSettings`].
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
#[non_exhaustive]
pub enum WsProtocolKind {
/// Standard Solana JSON-RPC WebSocket PubSub.
SolanaStandard,
}
impl WsProtocolKind {
/// Returns the stable KSP descriptor for this protocol family.
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::SolanaStandard => "solana_standard",
}
}
}
/// 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",
);
}
validate_non_zero_bound(self.command_queue_capacity, "ws_session.command_queue_capacity")?;
validate_non_zero_bound(self.notification_queue_capacity, "ws_session.notification_queue_capacity")?;
validate_non_zero_bound(self.max_active_subscriptions, "ws_session.max_active_subscriptions")?;
validate_non_zero_bound(self.max_pending_requests, "ws_session.max_pending_requests")?;
validate_non_zero_bound(self.max_message_size_bytes, "ws_session.max_message_size_bytes")?;
validate_non_zero_bound(self.max_frame_size_bytes, "ws_session.max_frame_size_bytes")?;
validate_non_zero_bound(self.max_write_buffer_size_bytes, "ws_session.max_write_buffer_size_bytes")?;
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 {
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,
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<std::string::String>,
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, url, session };
}
/// 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<crate::WsEndpointSettings>,
}
impl WsTransportSettings {
/// Creates complete WebSocket transport runtime settings.
#[must_use]
pub fn new(endpoints: std::vec::Vec<crate::WsEndpointSettings>) -> 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() {
validate_ws_endpoint(endpoint, endpoint_index)?;
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<()> {
validate_ws_descriptor(endpoint.name(), format!("ws_endpoints[{endpoint_index}].name").as_str())?;
validate_ws_descriptor(endpoint.provider().as_str(), format!("ws_endpoints[{endpoint_index}].provider").as_str())?;
validate_ws_descriptor(endpoint.cluster().as_str(), format!("ws_endpoints[{endpoint_index}].cluster").as_str())?;
endpoint.session().validate()?;
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(());
}
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;

View File

@@ -1,5 +1,5 @@
// file: crates/ksp-onchain-transport-lib/tests/public_api.rs
// version: 24
// version: 25
//! Integration tests for the public `ksp-onchain-transport-lib` consumer contract.
@@ -521,3 +521,31 @@ fn public_v0_2_4_pre_009_all_52_current_typed_wrappers_and_legacy_forms_are_avai
let _get_block_legacy = ksp_onchain_transport_lib::HttpTransportPool::get_block_legacy;
let _get_transaction_legacy = ksp_onchain_transport_lib::HttpTransportPool::get_transaction_legacy;
}
#[test]
fn public_v0_2_7_pre_002_websocket_settings_and_lifecycle_contracts_are_available_from_crate_root() {
let url = ksp_onchain_transport_lib::WsEndpointUrl::parse("wss://api.devnet.solana.com").expect("public WebSocket URL fixture must parse");
let session_settings = ksp_onchain_transport_lib::WsSessionSettings::default();
let endpoint = ksp_onchain_transport_lib::WsEndpointSettings::new(
"devnet_public",
true,
ksp_onchain_transport_lib::WsProviderName::new("solana-public"),
ksp_onchain_transport_lib::WsClusterName::new("devnet"),
ksp_onchain_transport_lib::WsProtocolKind::SolanaStandard,
url,
session_settings,
);
let settings = ksp_onchain_transport_lib::WsTransportSettings::new(std::vec![endpoint]);
assert!(settings.validate().is_ok());
assert_eq!(settings.endpoints()[0].protocol(), ksp_onchain_transport_lib::WsProtocolKind::SolanaStandard);
let session_id = ksp_onchain_transport_lib::WsSessionId::new(std::num::NonZeroU64::new(1).expect("public test ID must be non-zero"));
let subscription_id = ksp_onchain_transport_lib::WsSubscriptionId::new(std::num::NonZeroU64::new(2).expect("public test ID must be non-zero"));
assert_eq!(session_id.get(), 1);
assert_eq!(subscription_id.get(), 2);
assert_eq!(
ksp_onchain_transport_lib::WsSessionState::Reconnecting { attempt: 1 },
ksp_onchain_transport_lib::WsSessionState::Reconnecting { attempt: 1 }
);
assert_eq!(ksp_onchain_transport_lib::WsSubscriptionKind::Slot.as_str(), "slot");
assert_eq!(ksp_onchain_transport_lib::WsSubscriptionState::Requested, ksp_onchain_transport_lib::WsSubscriptionState::Requested);
}

View File

@@ -0,0 +1,70 @@
// file: crates/ksp-onchain-transport-lib/unit_tests/ws_lifecycle.rs
// version: 1
fn non_zero(value: u64) -> std::num::NonZeroU64 {
return std::num::NonZeroU64::new(value).expect("test ID must be non-zero");
}
#[test]
fn websocket_local_ids_preserve_ordering_and_numeric_identity() {
let first_session = crate::WsSessionId::new(non_zero(1));
let second_session = crate::WsSessionId::new(non_zero(2));
let subscription = crate::WsSubscriptionId::new(non_zero(7));
assert!(first_session < second_session);
assert_eq!(subscription.get(), 7);
}
#[test]
fn websocket_state_models_expose_concurrent_lifecycle_states() {
assert_eq!(crate::WsSessionState::Reconnecting { attempt: 3 }, crate::WsSessionState::Reconnecting { attempt: 3 });
assert_eq!(crate::WsSubscriptionState::Resubscribing, crate::WsSubscriptionState::Resubscribing);
assert_ne!(crate::WsSubscriptionState::Cancelling, crate::WsSubscriptionState::Closed);
}
#[test]
fn websocket_subscription_kinds_cover_all_nine_standard_families() {
let kinds = [
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,
];
assert_eq!(kinds.len(), 9);
assert_eq!(kinds[7].as_str(), "slots_updates");
}
#[test]
fn websocket_snapshots_expose_safe_metadata_without_remote_ids_or_urls() {
let subscription = crate::WsSubscriptionSnapshot::new(
crate::WsSubscriptionId::new(non_zero(9)),
crate::WsSubscriptionKind::Slot,
crate::WsSubscriptionState::Active,
true,
);
let snapshot = crate::WsSessionSnapshot::new(
crate::WsSessionId::new(non_zero(3)),
"devnet_public",
crate::WsProviderName::new("solana-public"),
crate::WsClusterName::new("devnet"),
crate::WsProtocolKind::SolanaStandard,
crate::WsSessionState::Active,
2,
1,
0,
std::vec![subscription],
);
assert_eq!(snapshot.id().get(), 3);
assert_eq!(snapshot.endpoint_name(), "devnet_public");
assert_eq!(snapshot.pending_request_count(), 2);
assert_eq!(snapshot.continuity_gap_count(), 1);
assert_eq!(snapshot.subscription_count(), 1);
assert!(snapshot.subscriptions()[0].remote_bound());
let rendered = format!("{snapshot:?}");
assert!(!rendered.contains("wss://"));
assert!(!rendered.contains("remote_subscription_id"));
}

View File

@@ -0,0 +1,148 @@
// file: crates/ksp-onchain-transport-lib/unit_tests/ws_settings.rs
// version: 1
fn valid_endpoint(name: &str, url_text: &str) -> crate::WsEndpointSettings {
return crate::WsEndpointSettings::new(
name,
true,
crate::WsProviderName::new("solana-public"),
crate::WsClusterName::new("devnet"),
crate::WsProtocolKind::SolanaStandard,
crate::WsEndpointUrl::parse(url_text).expect("test WebSocket URL must parse"),
crate::WsSessionSettings::default(),
);
}
#[test]
fn websocket_endpoint_url_accepts_ws_and_wss() {
assert!(crate::WsEndpointUrl::parse("wss://api.devnet.solana.com").is_ok());
assert!(crate::WsEndpointUrl::parse("ws://127.0.0.1:8900").is_ok());
}
#[test]
fn websocket_endpoint_url_rejects_http_schemes() {
let result = crate::WsEndpointUrl::parse("https://api.devnet.solana.com");
let error = result.expect_err("HTTP URL must not be accepted by WebSocket settings");
assert_eq!(error.code(), crate::ERROR_CODE_INVALID_SETTINGS);
}
#[test]
fn websocket_endpoint_url_debug_redacts_secret_material() {
let url = crate::WsEndpointUrl::parse("wss://user:password@provider.invalid/path?api-key=SECRET-CANARY").expect("test URL must parse");
let rendered = format!("{url:?}");
assert_eq!(rendered, "WsEndpointUrl(<redacted>)");
assert!(!rendered.contains("SECRET-CANARY"));
assert!(!rendered.contains("provider.invalid"));
assert!(!rendered.contains("password"));
}
#[test]
fn websocket_endpoint_url_errors_do_not_echo_sensitive_url() {
let result = crate::WsEndpointUrl::parse("http://user:password@provider.invalid/path?api-key=SECRET-CANARY");
let error = result.expect_err("unsupported scheme must fail");
let rendered = format!("{error:?}");
assert!(!rendered.contains("SECRET-CANARY"));
assert!(!rendered.contains("provider.invalid"));
assert!(!rendered.contains("password"));
}
#[test]
fn websocket_protocol_kind_is_extensible_but_only_standard_is_available_now() {
assert_eq!(crate::WsProtocolKind::SolanaStandard.as_str(), "solana_standard");
}
#[test]
fn websocket_session_defaults_are_bounded_and_validate() {
let settings = crate::WsSessionSettings::default();
assert!(settings.validate().is_ok());
assert!(settings.command_queue_capacity() > 0);
assert!(settings.notification_queue_capacity() > 0);
assert!(settings.max_active_subscriptions() > 0);
assert!(settings.max_pending_requests() > 0);
assert!(settings.max_message_size_bytes() > 0);
assert!(settings.max_frame_size_bytes() > 0);
assert!(settings.max_write_buffer_size_bytes() > 0);
assert_eq!(settings.resubscribe(), crate::WsResubscribePolicy::ActiveSubscriptions);
}
#[test]
fn websocket_session_settings_reject_zero_runtime_bounds() {
let defaults = crate::WsSessionSettings::default();
let settings = crate::WsSessionSettings::new(
defaults.command_timeout(),
defaults.close_timeout(),
defaults.reconnect().clone(),
defaults.resubscribe(),
0,
defaults.notification_queue_capacity(),
defaults.max_active_subscriptions(),
defaults.max_pending_requests(),
defaults.max_message_size_bytes(),
defaults.max_frame_size_bytes(),
defaults.max_write_buffer_size_bytes(),
);
let error = settings.validate().expect_err("zero command queue capacity must fail");
assert_eq!(error.code(), crate::ERROR_CODE_INVALID_SETTINGS);
}
#[test]
fn websocket_session_settings_reject_reversed_reconnect_backoff() {
let defaults = crate::WsSessionSettings::default();
let settings = crate::WsSessionSettings::new(
defaults.command_timeout(),
defaults.close_timeout(),
crate::WsReconnectSettings::new(2, std::time::Duration::from_secs(2), std::time::Duration::from_secs(1)),
defaults.resubscribe(),
defaults.command_queue_capacity(),
defaults.notification_queue_capacity(),
defaults.max_active_subscriptions(),
defaults.max_pending_requests(),
defaults.max_message_size_bytes(),
defaults.max_frame_size_bytes(),
defaults.max_write_buffer_size_bytes(),
);
assert!(settings.validate().is_err());
}
#[test]
fn websocket_transport_settings_validate_unique_enabled_endpoints() {
let settings = crate::WsTransportSettings::new(std::vec![
valid_endpoint("devnet_primary", "wss://api.devnet.solana.com"),
valid_endpoint("devnet_secondary", "wss://example.invalid/ws"),
]);
assert!(settings.validate().is_ok());
}
#[test]
fn websocket_transport_settings_reject_duplicate_endpoint_names() {
let settings =
crate::WsTransportSettings::new(std::vec![valid_endpoint("duplicate", "wss://one.invalid/ws"), valid_endpoint("duplicate", "wss://two.invalid/ws"),]);
let error = settings.validate().expect_err("duplicate endpoint names must fail");
assert_eq!(error.code(), crate::ERROR_CODE_INVALID_SETTINGS);
}
#[test]
fn websocket_transport_settings_require_one_enabled_endpoint() {
let endpoint = crate::WsEndpointSettings::new(
"disabled",
false,
crate::WsProviderName::new("provider"),
crate::WsClusterName::new("devnet"),
crate::WsProtocolKind::SolanaStandard,
crate::WsEndpointUrl::parse("wss://provider.invalid/ws").expect("test URL must parse"),
crate::WsSessionSettings::default(),
);
let settings = crate::WsTransportSettings::new(std::vec![endpoint]);
assert!(settings.validate().is_err());
}
#[test]
fn websocket_transport_settings_debug_never_exposes_endpoint_url() {
let settings =
crate::WsTransportSettings::new(std::vec![valid_endpoint("secret_endpoint", "wss://user:password@provider.invalid/path?api-key=SECRET-CANARY",)]);
let rendered = format!("{settings:?}");
assert!(rendered.contains("WsEndpointUrl(<redacted>)"));
assert!(!rendered.contains("SECRET-CANARY"));
assert!(!rendered.contains("provider.invalid"));
assert!(!rendered.contains("password"));
}