Sasedev/khadhroony-solana-project
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v0.2.7-pre.005

Browse Source
This commit is contained in:
SinuS von SifriduS
2026-08-22 19:16:41 +02:00
parent 34637848eb
commit 6e3a0fa034
10 changed files with 836 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files

439
crates/ksp-onchain-transport-lib/src/ws_session.rs
View File

@@ -1,5 +1,5 @@
// file: crates/ksp-onchain-transport-lib/src/ws_session.rs
// version: 2
// version: 3
use futures_util::SinkExt; // rust-rules: trait-import
use futures_util::StreamExt; // rust-rules: trait-import
@@ -9,13 +9,15 @@ static NEXT_WS_SESSION_ID: std::sync::atomic::AtomicU64 = std::sync::atomic::Ato
/// Shareable handle for one explicitly created physical WebSocket session.
///
/// The handle never exposes the sensitive endpoint URL or the underlying socket. All socket I/O is owned by one internal actor task and all caller
/// interaction is serialized through a bounded command queue.
/// interaction is serialized through bounded channels.
#[derive(Clone)]
pub struct WsSession {
id: crate::WsSessionId,
command_tx: tokio::sync::mpsc::Sender<WsSessionCommand>,
shutdown_tx: tokio::sync::watch::Sender<std::option::Option<tokio::time::Instant>>,
snapshot_rx: tokio::sync::watch::Receiver<crate::WsSessionSnapshot>,
command_timeout: std::time::Duration,
close_timeout: std::time::Duration,
}
impl WsSession {
@@ -46,9 +48,11 @@ impl WsSession {
std::vec::Vec::new(),
);
let (command_tx, command_rx) = tokio::sync::mpsc::channel(endpoint.session().command_queue_capacity());
let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(std::option::Option::None::<tokio::time::Instant>);
let (snapshot_tx, snapshot_rx) = tokio::sync::watch::channel(initial_snapshot);
let (startup_tx, startup_rx) = tokio::sync::oneshot::channel();
let command_timeout = endpoint.session().command_timeout();
let close_timeout = endpoint.session().close_timeout();
ksp_logging_lib::debug!(
target: crate::TRACING_TARGET,
session_id = id.get(),
@@ -58,25 +62,25 @@ impl WsSession {
protocol = endpoint.protocol().as_str(),
"starting physical WebSocket session actor"
);
let join_handle = tokio::spawn(run_ws_session_actor(id, endpoint, command_rx, snapshot_tx, startup_tx));
let join_handle = tokio::spawn(run_ws_session_actor(id, endpoint, command_rx, shutdown_rx, snapshot_tx, startup_tx));
let startup_wait = tokio::time::timeout(command_timeout, startup_rx).await;
match startup_wait {
return match startup_wait {
std::result::Result::Ok(std::result::Result::Ok(std::result::Result::Ok(()))) => {
return std::result::Result::Ok(Self { id, command_tx, snapshot_rx, command_timeout });
std::result::Result::Ok(Self { id, command_tx, shutdown_tx, snapshot_rx, command_timeout, close_timeout })
},
std::result::Result::Ok(std::result::Result::Ok(std::result::Result::Err(error))) => {
join_handle.abort();
return std::result::Result::Err(error);
std::result::Result::Err(error)
},
std::result::Result::Ok(std::result::Result::Err(_)) => {
join_handle.abort();
return std::result::Result::Err(ws_session_closed_error(id, "WebSocket session actor ended during startup"));
std::result::Result::Err(ws_session_closed_error(id, "WebSocket session actor ended during startup"))
},
std::result::Result::Err(_) => {
join_handle.abort();
return std::result::Result::Err(ws_timeout_error(id, "WebSocket handshake exceeded the configured command timeout"));
std::result::Result::Err(ws_timeout_error(id, "WebSocket handshake exceeded the configured command timeout"))
},
}
};
}
/// Returns the stable local session identity.
@@ -97,11 +101,57 @@ impl WsSession {
return self.snapshot_rx.borrow().state();
}
/// Explicitly closes this physical session under the configured close timeout.
///
/// Shutdown is session-wide: calling this method through any clone requests the actor to enter `Closing`, cancels pending JSON-RPC requests, sends a
/// best-effort WebSocket Close frame and waits for the safe lifecycle snapshot to reach `Closed`. The shutdown signal is independent from the bounded
/// command queue so a saturated command queue cannot prevent close from being requested.
pub async fn close(&self) -> ksp_core_lib::Result<()> {
match self.state() {
crate::WsSessionState::Closed => return std::result::Result::Ok(()),
crate::WsSessionState::Failed => {
return std::result::Result::Err(ws_session_closed_error(self.id, "WebSocket session is already in terminal failed state"));
},
_ => {},
}
let deadline = tokio::time::Instant::now() + self.close_timeout;
self.shutdown_tx.send_replace(std::option::Option::Some(deadline));
let mut snapshot_rx = self.snapshot_rx.clone();
loop {
match snapshot_rx.borrow().state() {
crate::WsSessionState::Closed => return std::result::Result::Ok(()),
crate::WsSessionState::Failed => {
return std::result::Result::Err(ws_session_closed_error(self.id, "WebSocket session failed while explicit shutdown was in progress"));
},
_ => {},
}
let changed = tokio::time::timeout_at(deadline, snapshot_rx.changed()).await;
match changed {
std::result::Result::Ok(std::result::Result::Ok(())) => {},
std::result::Result::Ok(std::result::Result::Err(_)) => {
return match snapshot_rx.borrow().state() {
crate::WsSessionState::Closed => std::result::Result::Ok(()),
_ => std::result::Result::Err(ws_session_closed_error(self.id, "WebSocket session actor ended before publishing Closed state")),
};
},
std::result::Result::Err(_) => {
if snapshot_rx.borrow().state() == crate::WsSessionState::Closed {
return std::result::Result::Ok(());
}
return std::result::Result::Err(ws_timeout_error(self.id, "WebSocket explicit shutdown exceeded the configured close timeout"));
},
}
}
}
/// Executes one internal JSON-RPC request through the actor-owned socket.
///
/// This remains crate-private in `0.2.7`; standard subscriptions consume it without exposing a public raw provider-extension escape hatch.
#[allow(dead_code)] // Staged in pre.004 and consumed by the subscription engine starting in pre.006.
pub(crate) async fn execute_json_rpc(&self, method: &'static str, params: std::vec::Vec<serde_json::Value>) -> ksp_core_lib::Result<serde_json::Value> {
if self.state() != crate::WsSessionState::Active {
return std::result::Result::Err(ws_session_closed_error(self.id, "WebSocket session is not active"));
}
let (response_tx, response_rx) = tokio::sync::oneshot::channel();
let command = WsSessionCommand::ExecuteJsonRpc { method, params, response_tx };
let send_wait = tokio::time::timeout(self.command_timeout, self.command_tx.send(command)).await;
@@ -114,14 +164,10 @@ impl WsSession {
return std::result::Result::Err(ws_timeout_error(self.id, "WebSocket session command queue remained unavailable until timeout"));
},
}
let response_wait = tokio::time::timeout(self.command_timeout, response_rx).await;
return match response_wait {
std::result::Result::Ok(std::result::Result::Ok(result)) => result,
std::result::Result::Ok(std::result::Result::Err(_)) => {
std::result::Result::Err(ws_session_closed_error(self.id, "WebSocket session ended before the JSON-RPC response was delivered"))
},
return match response_rx.await {
std::result::Result::Ok(result) => result,
std::result::Result::Err(_) => {
std::result::Result::Err(ws_timeout_error(self.id, "WebSocket JSON-RPC request exceeded the configured command timeout"))
std::result::Result::Err(ws_session_closed_error(self.id, "WebSocket session ended before the JSON-RPC response was delivered"))
},
};
}
@@ -147,10 +193,18 @@ struct PendingWsRequest {
response_tx: tokio::sync::oneshot::Sender<ksp_core_lib::Result<serde_json::Value>>,
}
enum WsActorIoOutcome {
Continue,
RemoteClosed,
ShutdownRequested { deadline: tokio::time::Instant },
Failed { code: ksp_core_lib::ErrorCode, pending_message: &'static str },
}
async fn run_ws_session_actor(
id: crate::WsSessionId,
endpoint: crate::WsEndpointSettings,
mut command_rx: tokio::sync::mpsc::Receiver<WsSessionCommand>,
mut shutdown_rx: tokio::sync::watch::Receiver<std::option::Option<tokio::time::Instant>>,
snapshot_tx: tokio::sync::watch::Sender<crate::WsSessionSnapshot>,
startup_tx: tokio::sync::oneshot::Sender<ksp_core_lib::Result<()>>,
) {
@@ -213,35 +267,81 @@ async fn run_ws_session_actor(
let mut next_request_id = 1_u64;
let mut pending = std::collections::BTreeMap::<u64, PendingWsRequest>::new();
loop {
prune_cancelled_pending(id, &mut pending);
let timeout_deadline = next_pending_deadline(&pending);
tokio::select! {
biased;
shutdown_changed = shutdown_rx.changed() => {
let deadline = resolve_shutdown_deadline(&shutdown_rx, shutdown_changed, endpoint.session().close_timeout());
close_session_actor(id, &endpoint, &snapshot_tx, &mut websocket, &mut pending, deadline).await;
return;
},
maybe_command = command_rx.recv() => {
let command = match maybe_command {
std::option::Option::Some(command) => command,
std::option::Option::None => {
let deadline = tokio::time::Instant::now() + endpoint.session().close_timeout();
ksp_logging_lib::trace!(target: crate::TRACING_TARGET, session_id = id.get(), "all WebSocket session handles dropped; closing actor");
let _ = websocket.send(tokio_tungstenite::tungstenite::Message::Close(std::option::Option::None)).await;
close_session_actor(id, &endpoint, &snapshot_tx, &mut websocket, &mut pending, deadline).await;
return;
},
};
let command_outcome = handle_session_command(
id,
&endpoint,
&mut websocket,
&mut pending,
&mut next_request_id,
&mut shutdown_rx,
command,
)
.await;
match command_outcome {
WsActorIoOutcome::Continue => {
publish_snapshot(&snapshot_tx, id, &endpoint, crate::WsSessionState::Active, pending.len());
},
WsActorIoOutcome::RemoteClosed => {
publish_snapshot(&snapshot_tx, id, &endpoint, crate::WsSessionState::Closed, pending.len());
fail_all_pending(&mut pending, id, crate::ERROR_CODE_WS_SESSION_CLOSED, "WebSocket session closed before pending response delivery");
return;
},
};
let command_outcome = handle_session_command(id, &endpoint, &mut websocket, &mut pending, &mut next_request_id, command).await;
if !command_outcome {
publish_snapshot(&snapshot_tx, id, &endpoint, crate::WsSessionState::Failed, pending.len());
fail_all_pending(&mut pending, id, crate::ERROR_CODE_WS_CONNECTION_FAILED, "WebSocket connection failed while writing a request");
return;
WsActorIoOutcome::ShutdownRequested { deadline } => {
close_session_actor(id, &endpoint, &snapshot_tx, &mut websocket, &mut pending, deadline).await;
return;
},
WsActorIoOutcome::Failed { code, pending_message } => {
publish_snapshot(&snapshot_tx, id, &endpoint, crate::WsSessionState::Failed, pending.len());
fail_all_pending(&mut pending, id, code, pending_message);
return;
},
}
publish_snapshot(&snapshot_tx, id, &endpoint, crate::WsSessionState::Active, pending.len());
},
maybe_message = websocket.next() => {
let keep_running = handle_socket_message(id, &endpoint, maybe_message, &mut websocket, &mut pending).await;
if !keep_running {
publish_snapshot(&snapshot_tx, id, &endpoint, crate::WsSessionState::Failed, pending.len());
fail_all_pending(&mut pending, id, crate::ERROR_CODE_WS_CONNECTION_FAILED, "WebSocket connection ended before pending response delivery");
return;
let socket_outcome = handle_socket_message(id, &endpoint, maybe_message, &mut websocket, &mut pending, &mut shutdown_rx).await;
match socket_outcome {
WsActorIoOutcome::Continue => {
publish_snapshot(&snapshot_tx, id, &endpoint, crate::WsSessionState::Active, pending.len());
},
WsActorIoOutcome::RemoteClosed => {
publish_snapshot(&snapshot_tx, id, &endpoint, crate::WsSessionState::Closed, pending.len());
fail_all_pending(
&mut pending,
id,
crate::ERROR_CODE_WS_SESSION_CLOSED,
"Remote peer closed WebSocket session before pending response delivery",
);
return;
},
WsActorIoOutcome::ShutdownRequested { deadline } => {
close_session_actor(id, &endpoint, &snapshot_tx, &mut websocket, &mut pending, deadline).await;
return;
},
WsActorIoOutcome::Failed { code, pending_message } => {
publish_snapshot(&snapshot_tx, id, &endpoint, crate::WsSessionState::Failed, pending.len());
fail_all_pending(&mut pending, id, code, pending_message);
return;
},
}
publish_snapshot(&snapshot_tx, id, &endpoint, crate::WsSessionState::Active, pending.len());
},
() = tokio::time::sleep_until(timeout_deadline) => {
expire_pending_requests(id, &mut pending);
@@ -251,18 +351,20 @@ async fn run_ws_session_actor(
}
}
#[allow(clippy::too_many_arguments)]
async fn handle_session_command<S>(
id: crate::WsSessionId,
endpoint: &crate::WsEndpointSettings,
websocket: &mut tokio_tungstenite::WebSocketStream<S>,
pending: &mut std::collections::BTreeMap<u64, PendingWsRequest>,
next_request_id: &mut u64,
shutdown_rx: &mut tokio::sync::watch::Receiver<std::option::Option<tokio::time::Instant>>,
command: WsSessionCommand,
) -> bool
) -> WsActorIoOutcome
where
S: tokio::io::AsyncRead + tokio::io::AsyncWrite + std::marker::Unpin,
{
match command {
return match command {
WsSessionCommand::ExecuteJsonRpc { method, params, response_tx } => {
if pending.len() >= endpoint.session().max_pending_requests() {
let error = ksp_core_lib::Error::new(crate::ERROR_CODE_WS_BACKPRESSURE_OVERFLOW, "WebSocket pending JSON-RPC request capacity is exhausted")
@@ -276,7 +378,7 @@ where
max_pending_requests = endpoint.session().max_pending_requests(),
"rejected WebSocket JSON-RPC request because pending capacity is exhausted"
);
return true;
return WsActorIoOutcome::Continue;
}
let request_id = *next_request_id;
let incremented = request_id.checked_add(1);
@@ -286,7 +388,7 @@ where
let error = ksp_core_lib::Error::new(crate::ERROR_CODE_WS_PROTOCOL_ERROR, "WebSocket JSON-RPC request identifier space is exhausted")
.with_context("session_id", id.get().to_string());
let _ = response_tx.send(std::result::Result::Err(error));
return true;
return WsActorIoOutcome::Continue;
},
};
let request_result = crate::JsonRpcRequest::new(request_id, method, params);
@@ -294,7 +396,7 @@ where
std::result::Result::Ok(request) => request,
std::result::Result::Err(error) => {
let _ = response_tx.send(std::result::Result::Err(error));
return true;
return WsActorIoOutcome::Continue;
},
};
let payload_result = request.to_json_string();
@@ -302,11 +404,30 @@ where
std::result::Result::Ok(payload) => payload,
std::result::Result::Err(error) => {
let _ = response_tx.send(std::result::Result::Err(error));
return true;
return WsActorIoOutcome::Continue;
},
};
let deadline = tokio::time::Instant::now() + endpoint.session().command_timeout();
pending.insert(request_id, PendingWsRequest { method, deadline, response_tx });
let payload_size = payload.len();
if payload_size > endpoint.session().max_message_size_bytes() || payload_size > endpoint.session().max_frame_size_bytes() {
let error =
ksp_core_lib::Error::new(crate::ERROR_CODE_INVALID_RPC_PARAMETERS, "WebSocket JSON-RPC request exceeds the configured outbound size bound")
.with_context("session_id", id.get().to_string())
.with_context("method", method)
.with_context("payload_size_bytes", payload_size.to_string())
.with_context("max_message_size_bytes", endpoint.session().max_message_size_bytes().to_string())
.with_context("max_frame_size_bytes", endpoint.session().max_frame_size_bytes().to_string());
let _ = response_tx.send(std::result::Result::Err(error));
ksp_logging_lib::debug!(
target: crate::TRACING_TARGET,
session_id = id.get(),
method,
payload_size_bytes = payload_size,
max_message_size_bytes = endpoint.session().max_message_size_bytes(),
max_frame_size_bytes = endpoint.session().max_frame_size_bytes(),
"rejected oversized WebSocket JSON-RPC request before socket write"
);
return WsActorIoOutcome::Continue;
}
ksp_logging_lib::trace!(
target: crate::TRACING_TARGET,
session_id = id.get(),
@@ -315,20 +436,65 @@ where
pending_request_count = pending.len(),
"sending WebSocket JSON-RPC request"
);
let send_result = websocket.send(tokio_tungstenite::tungstenite::Message::Text(payload.into())).await;
if send_result.is_err() {
ksp_logging_lib::warn!(
target: crate::TRACING_TARGET,
session_id = id.get(),
request_id,
method,
"WebSocket JSON-RPC request write failed"
);
return false;
let message = tokio_tungstenite::tungstenite::Message::Text(payload.into());
let send_result = tokio::select! {
biased;
shutdown_changed = shutdown_rx.changed() => {
let deadline = resolve_shutdown_deadline(shutdown_rx, shutdown_changed, endpoint.session().close_timeout());
let error = ws_session_closed_error(id, "WebSocket JSON-RPC request cancelled by session shutdown").with_context("method", method);
let _ = response_tx.send(std::result::Result::Err(error));
return WsActorIoOutcome::ShutdownRequested { deadline };
},
send_result = websocket.send(message) => send_result,
() = tokio::time::sleep(endpoint.session().command_timeout()) => {
let error = ws_timeout_error(id, "WebSocket JSON-RPC request write exceeded the configured command timeout").with_context("method", method);
let _ = response_tx.send(std::result::Result::Err(error));
ksp_logging_lib::warn!(
target: crate::TRACING_TARGET,
session_id = id.get(),
request_id,
method,
"WebSocket JSON-RPC request write timed out"
);
return WsActorIoOutcome::Failed {
code: crate::ERROR_CODE_WS_CONNECTION_FAILED,
pending_message: "WebSocket connection failed while a request write timed out",
};
},
};
match send_result {
std::result::Result::Ok(()) => {},
std::result::Result::Err(tokio_tungstenite::tungstenite::Error::WriteBufferFull(_))
| std::result::Result::Err(tokio_tungstenite::tungstenite::Error::Capacity(_)) => {
let error = ksp_core_lib::Error::new(crate::ERROR_CODE_WS_BACKPRESSURE_OVERFLOW, "WebSocket write capacity is exhausted")
.with_context("session_id", id.get().to_string())
.with_context("method", method);
let _ = response_tx.send(std::result::Result::Err(error));
ksp_logging_lib::warn!(
target: crate::TRACING_TARGET,
session_id = id.get(),
request_id,
method,
"WebSocket request write capacity is exhausted"
);
return WsActorIoOutcome::Continue;
},
std::result::Result::Err(_) => {
let error =
ws_connection_error(id, endpoint, "WebSocket connection failed while writing a JSON-RPC request").with_context("method", method);
let _ = response_tx.send(std::result::Result::Err(error));
ksp_logging_lib::warn!(target: crate::TRACING_TARGET, session_id = id.get(), request_id, method, "WebSocket JSON-RPC request write failed");
return WsActorIoOutcome::Failed {
code: crate::ERROR_CODE_WS_CONNECTION_FAILED,
pending_message: "WebSocket connection failed while writing a request",
};
},
}
return true;
let deadline = tokio::time::Instant::now() + endpoint.session().command_timeout();
pending.insert(request_id, PendingWsRequest { method, deadline, response_tx });
WsActorIoOutcome::Continue
},
}
};
}
async fn handle_socket_message<S>(
@@ -337,70 +503,103 @@ async fn handle_socket_message<S>(
maybe_message: std::option::Option<std::result::Result<tokio_tungstenite::tungstenite::Message, tokio_tungstenite::tungstenite::Error>>,
websocket: &mut tokio_tungstenite::WebSocketStream<S>,
pending: &mut std::collections::BTreeMap<u64, PendingWsRequest>,
) -> bool
shutdown_rx: &mut tokio::sync::watch::Receiver<std::option::Option<tokio::time::Instant>>,
) -> WsActorIoOutcome
where
S: tokio::io::AsyncRead + tokio::io::AsyncWrite + std::marker::Unpin,
{
let message = match maybe_message {
std::option::Option::Some(std::result::Result::Ok(message)) => message,
std::option::Option::Some(std::result::Result::Err(_)) => {
std::option::Option::Some(std::result::Result::Err(error)) => {
let code = classify_websocket_read_error(&error);
ksp_logging_lib::warn!(
target: crate::TRACING_TARGET,
session_id = id.get(),
endpoint_name = endpoint.name(),
protocol_error = code == crate::ERROR_CODE_WS_PROTOCOL_ERROR,
"physical WebSocket read failed"
);
return false;
return WsActorIoOutcome::Failed { code, pending_message: "WebSocket read failed before pending response delivery" };
},
std::option::Option::None => {
ksp_logging_lib::warn!(
target: crate::TRACING_TARGET,
session_id = id.get(),
endpoint_name = endpoint.name(),
"physical WebSocket stream ended"
"physical WebSocket stream ended without a Close frame"
);
return false;
return WsActorIoOutcome::Failed {
code: crate::ERROR_CODE_WS_CONNECTION_FAILED,
pending_message: "WebSocket connection ended before pending response delivery",
};
},
};
return match message {
tokio_tungstenite::tungstenite::Message::Text(text) => handle_text_message(id, text.as_str(), pending),
tokio_tungstenite::tungstenite::Message::Binary(_) => {
ksp_logging_lib::warn!(target: crate::TRACING_TARGET, session_id = id.get(), "received unexpected binary WebSocket message");
false
WsActorIoOutcome::Failed {
code: crate::ERROR_CODE_WS_PROTOCOL_ERROR,
pending_message: "WebSocket wire data violated the expected JSON text protocol",
}
},
tokio_tungstenite::tungstenite::Message::Ping(payload) => {
ksp_logging_lib::trace!(target: crate::TRACING_TARGET, session_id = id.get(), "received WebSocket ping control frame");
return websocket.send(tokio_tungstenite::tungstenite::Message::Pong(payload)).await.is_ok();
tokio_tungstenite::tungstenite::Message::Ping(_) => {
ksp_logging_lib::trace!(target: crate::TRACING_TARGET, session_id = id.get(), "received WebSocket ping control frame; flushing automatic Pong");
let flush_result = tokio::select! {
biased;
shutdown_changed = shutdown_rx.changed() => {
let deadline = resolve_shutdown_deadline(shutdown_rx, shutdown_changed, endpoint.session().close_timeout());
return WsActorIoOutcome::ShutdownRequested { deadline };
},
flush_result = websocket.flush() => flush_result,
() = tokio::time::sleep(endpoint.session().command_timeout()) => {
return WsActorIoOutcome::Failed {
code: crate::ERROR_CODE_WS_CONNECTION_FAILED,
pending_message: "WebSocket connection failed while flushing automatic Pong",
};
},
};
return match flush_result {
std::result::Result::Ok(()) => WsActorIoOutcome::Continue,
std::result::Result::Err(_) => WsActorIoOutcome::Failed {
code: crate::ERROR_CODE_WS_CONNECTION_FAILED,
pending_message: "WebSocket connection failed while flushing automatic Pong",
},
};
},
tokio_tungstenite::tungstenite::Message::Pong(_) => {
ksp_logging_lib::trace!(target: crate::TRACING_TARGET, session_id = id.get(), "received WebSocket pong control frame");
true
WsActorIoOutcome::Continue
},
tokio_tungstenite::tungstenite::Message::Close(_) => {
ksp_logging_lib::debug!(target: crate::TRACING_TARGET, session_id = id.get(), "remote peer closed physical WebSocket session");
false
let _ = tokio::time::timeout(endpoint.session().close_timeout(), websocket.flush()).await;
WsActorIoOutcome::RemoteClosed
},
tokio_tungstenite::tungstenite::Message::Frame(_) => {
ksp_logging_lib::trace!(target: crate::TRACING_TARGET, session_id = id.get(), "ignored internal WebSocket frame event");
true
WsActorIoOutcome::Continue
},
};
}
fn handle_text_message(id: crate::WsSessionId, text: &str, pending: &mut std::collections::BTreeMap<u64, PendingWsRequest>) -> bool {
fn handle_text_message(id: crate::WsSessionId, text: &str, pending: &mut std::collections::BTreeMap<u64, PendingWsRequest>) -> WsActorIoOutcome {
let decoded = serde_json::from_str::<serde_json::Value>(text);
let value = match decoded {
std::result::Result::Ok(value) => value,
std::result::Result::Err(_) => {
ksp_logging_lib::warn!(target: crate::TRACING_TARGET, session_id = id.get(), "received malformed JSON WebSocket message");
return false;
return WsActorIoOutcome::Failed { code: crate::ERROR_CODE_WS_PROTOCOL_ERROR, pending_message: "WebSocket JSON payload was malformed" };
},
};
let object = match value.as_object() {
std::option::Option::Some(object) => object,
std::option::Option::None => {
ksp_logging_lib::warn!(target: crate::TRACING_TARGET, session_id = id.get(), "received non-object JSON WebSocket message");
return false;
return WsActorIoOutcome::Failed {
code: crate::ERROR_CODE_WS_PROTOCOL_ERROR,
pending_message: "WebSocket JSON-RPC payload was not an object",
};
},
};
if !object.contains_key("id") {
@@ -410,16 +609,22 @@ fn handle_text_message(id: crate::WsSessionId, text: &str, pending: &mut std::co
session_id = id.get(),
"received WebSocket notification before subscription registry activation; safely ignored"
);
return true;
return WsActorIoOutcome::Continue;
}
ksp_logging_lib::warn!(target: crate::TRACING_TARGET, session_id = id.get(), "received structurally invalid WebSocket JSON-RPC message");
return false;
return WsActorIoOutcome::Failed {
code: crate::ERROR_CODE_WS_PROTOCOL_ERROR,
pending_message: "WebSocket JSON-RPC payload violated structural invariants",
};
}
let response_id = match object.get("id").and_then(serde_json::Value::as_u64) {
std::option::Option::Some(response_id) => response_id,
std::option::Option::None => {
ksp_logging_lib::warn!(target: crate::TRACING_TARGET, session_id = id.get(), "received WebSocket JSON-RPC response with invalid id");
return false;
return WsActorIoOutcome::Failed {
code: crate::ERROR_CODE_WS_PROTOCOL_ERROR,
pending_message: "WebSocket JSON-RPC response contained an invalid id",
};
},
};
let pending_request = match pending.remove(&response_id) {
@@ -431,24 +636,30 @@ fn handle_text_message(id: crate::WsSessionId, text: &str, pending: &mut std::co
response_id,
"ignored unknown or stale WebSocket JSON-RPC response id"
);
return true;
return WsActorIoOutcome::Continue;
},
};
let parsed = crate::parse_json_rpc_response_value(value, response_id);
let (result, keep_running) = match parsed {
let (result, outcome) = match parsed {
std::result::Result::Ok(response) => {
let result = match response.into_result() {
std::result::Result::Ok(value) => std::result::Result::Ok(value),
std::result::Result::Err(error) => std::result::Result::Err(error.with_context("method", pending_request.method)),
};
(result, true)
(result, WsActorIoOutcome::Continue)
},
std::result::Result::Err(_) => {
let error = ksp_core_lib::Error::new(crate::ERROR_CODE_WS_PROTOCOL_ERROR, "WebSocket JSON-RPC response violates protocol invariants")
.with_context("session_id", id.get().to_string())
.with_context("request_id", response_id.to_string())
.with_context("method", pending_request.method);
(std::result::Result::Err(error), false)
(
std::result::Result::Err(error),
WsActorIoOutcome::Failed {
code: crate::ERROR_CODE_WS_PROTOCOL_ERROR,
pending_message: "WebSocket JSON-RPC response violated protocol invariants",
},
)
},
};
let _ = pending_request.response_tx.send(result);
@@ -460,7 +671,69 @@ fn handle_text_message(id: crate::WsSessionId, text: &str, pending: &mut std::co
pending_request_count = pending.len(),
"dispatched WebSocket JSON-RPC response to pending request"
);
return keep_running;
return outcome;
}
async fn close_session_actor<S>(
id: crate::WsSessionId,
endpoint: &crate::WsEndpointSettings,
snapshot_tx: &tokio::sync::watch::Sender<crate::WsSessionSnapshot>,
websocket: &mut tokio_tungstenite::WebSocketStream<S>,
pending: &mut std::collections::BTreeMap<u64, PendingWsRequest>,
deadline: tokio::time::Instant,
) where
S: tokio::io::AsyncRead + tokio::io::AsyncWrite + std::marker::Unpin,
{
publish_snapshot(snapshot_tx, id, endpoint, crate::WsSessionState::Closing, pending.len());
fail_all_pending(pending, id, crate::ERROR_CODE_WS_SESSION_CLOSED, "WebSocket session shutdown cancelled the pending request");
publish_snapshot(snapshot_tx, id, endpoint, crate::WsSessionState::Closing, 0);
ksp_logging_lib::debug!(
target: crate::TRACING_TARGET,
session_id = id.get(),
endpoint_name = endpoint.name(),
"closing physical WebSocket session"
);
let now = tokio::time::Instant::now();
let remaining = deadline.saturating_duration_since(now);
let io_deadline = now + (remaining / 2);
let close_wait = tokio::time::timeout_at(io_deadline, websocket.send(tokio_tungstenite::tungstenite::Message::Close(std::option::Option::None))).await;
match close_wait {
std::result::Result::Ok(std::result::Result::Ok(())) => {
ksp_logging_lib::trace!(target: crate::TRACING_TARGET, session_id = id.get(), "sent WebSocket Close control frame");
},
std::result::Result::Ok(std::result::Result::Err(_)) => {
ksp_logging_lib::debug!(target: crate::TRACING_TARGET, session_id = id.get(), "best-effort WebSocket Close frame could not be sent");
},
std::result::Result::Err(_) => {
ksp_logging_lib::debug!(target: crate::TRACING_TARGET, session_id = id.get(), "best-effort WebSocket Close frame reached close deadline");
},
}
publish_snapshot(snapshot_tx, id, endpoint, crate::WsSessionState::Closed, 0);
ksp_logging_lib::debug!(target: crate::TRACING_TARGET, session_id = id.get(), endpoint_name = endpoint.name(), "physical WebSocket session is closed");
}
fn classify_websocket_read_error(error: &tokio_tungstenite::tungstenite::Error) -> ksp_core_lib::ErrorCode {
return match error {
tokio_tungstenite::tungstenite::Error::Capacity(_)
| tokio_tungstenite::tungstenite::Error::Protocol(_)
| tokio_tungstenite::tungstenite::Error::Utf8(_)
| tokio_tungstenite::tungstenite::Error::AttackAttempt => crate::ERROR_CODE_WS_PROTOCOL_ERROR,
_ => crate::ERROR_CODE_WS_CONNECTION_FAILED,
};
}
fn resolve_shutdown_deadline(
shutdown_rx: &tokio::sync::watch::Receiver<std::option::Option<tokio::time::Instant>>,
shutdown_changed: std::result::Result<(), tokio::sync::watch::error::RecvError>,
close_timeout: std::time::Duration,
) -> tokio::time::Instant {
if shutdown_changed.is_ok() {
let requested = shutdown_rx.borrow().to_owned();
if let std::option::Option::Some(deadline) = requested {
return deadline;
}
}
return tokio::time::Instant::now() + close_timeout;
}
fn next_pending_deadline(pending: &std::collections::BTreeMap<u64, PendingWsRequest>) -> tokio::time::Instant {
@@ -500,6 +773,26 @@ fn expire_pending_requests(id: crate::WsSessionId, pending: &mut std::collection
}
}
fn prune_cancelled_pending(id: crate::WsSessionId, pending: &mut std::collections::BTreeMap<u64, PendingWsRequest>) {
let mut cancelled_ids = std::vec::Vec::new();
for (request_id, request) in pending.iter() {
if request.response_tx.is_closed() {
cancelled_ids.push(*request_id);
}
}
for request_id in cancelled_ids {
if let std::option::Option::Some(request) = pending.remove(&request_id) {
ksp_logging_lib::trace!(
target: crate::TRACING_TARGET,
session_id = id.get(),
request_id,
method = request.method,
"removed abandoned WebSocket JSON-RPC request after caller cancellation"
);
}
}
}
fn fail_all_pending(
pending: &mut std::collections::BTreeMap<u64, PendingWsRequest>,
id: crate::WsSessionId,