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

v0.2.7-pre.008

Browse Source
This commit is contained in:
SinuS von SifriduS
2026-08-22 21:44:18 +02:00
parent 4c540d67a7
commit 6fefc64e75
13 changed files with 804 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files

326
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: 7
// version: 8
use futures_util::SinkExt; // rust-rules: trait-import
use futures_util::StreamExt; // rust-rules: trait-import
@@ -275,10 +275,16 @@ enum WsActorIoOutcome {
Continue,
RemoteClosed,
ShutdownRequested { deadline: tokio::time::Instant },
StaleSubscribeAck { kind: crate::WsSubscriptionKind, remote_id: u64 },
BestEffortUnsubscribe { kind: crate::WsSubscriptionKind, remote_id: u64 },
Failed { code: ksp_core_lib::ErrorCode, pending_message: &'static str },
}
#[derive(Clone, Copy, Default)]
struct WsRuntimeCounters {
continuity_gap_count: u64,
overflow_count: u64,
}
enum WsReconnectOutcome {
Connected { websocket: std::boxed::Box<WsPhysicalStream> },
ShutdownRequested { deadline: tokio::time::Instant },
@@ -323,7 +329,7 @@ async fn run_ws_session_actor(
let (mut websocket, handshake_status) = match connect_wait {
std::result::Result::Ok(std::result::Result::Ok((websocket, response))) => (websocket, response.status().as_u16()),
std::result::Result::Ok(std::result::Result::Err(_)) => {
publish_snapshot(&snapshot_tx, id, &endpoint, crate::WsSessionState::Failed, 0, 0, &std::collections::BTreeMap::new());
publish_snapshot(&snapshot_tx, id, &endpoint, crate::WsSessionState::Failed, 0, WsRuntimeCounters::default(), &std::collections::BTreeMap::new());
let error = ws_connection_error(id, &endpoint, "WebSocket connection or handshake failed");
let _ = startup_tx.send(std::result::Result::Err(error));
ksp_logging_lib::warn!(
@@ -337,7 +343,7 @@ async fn run_ws_session_actor(
return;
},
std::result::Result::Err(_) => {
publish_snapshot(&snapshot_tx, id, &endpoint, crate::WsSessionState::Failed, 0, 0, &std::collections::BTreeMap::new());
publish_snapshot(&snapshot_tx, id, &endpoint, crate::WsSessionState::Failed, 0, WsRuntimeCounters::default(), &std::collections::BTreeMap::new());
let error = ws_timeout_error(id, "WebSocket connection handshake timed out");
let _ = startup_tx.send(std::result::Result::Err(error));
ksp_logging_lib::warn!(
@@ -350,7 +356,16 @@ async fn run_ws_session_actor(
},
};
let mut continuity_gap_count = 0_u64;
publish_snapshot(&snapshot_tx, id, &endpoint, crate::WsSessionState::Active, 0, continuity_gap_count, &std::collections::BTreeMap::new());
let mut overflow_count = 0_u64;
publish_snapshot(
&snapshot_tx,
id,
&endpoint,
crate::WsSessionState::Active,
0,
WsRuntimeCounters { continuity_gap_count, overflow_count },
&std::collections::BTreeMap::new(),
);
let _ = startup_tx.send(std::result::Result::Ok(()));
ksp_logging_lib::debug!(
target: crate::TRACING_TARGET,
@@ -388,6 +403,7 @@ async fn run_ws_session_actor(
&mut subscriptions,
&mut remote_to_local,
continuity_gap_count,
overflow_count,
deadline,
)
.await;
@@ -417,6 +433,7 @@ async fn run_ws_session_actor(
&mut pending,
&mut subscriptions,
&mut remote_to_local,
&mut overflow_count,
&mut shutdown_rx,
)
.await
@@ -428,10 +445,18 @@ async fn run_ws_session_actor(
};
match outcome {
WsActorIoOutcome::Continue => {
publish_snapshot(&snapshot_tx, id, &endpoint, crate::WsSessionState::Active, pending.len(), continuity_gap_count, &subscriptions);
publish_snapshot(
&snapshot_tx,
id,
&endpoint,
crate::WsSessionState::Active,
pending.len(),
WsRuntimeCounters { continuity_gap_count, overflow_count },
&subscriptions,
);
},
WsActorIoOutcome::StaleSubscribeAck { kind, remote_id } => {
let cleanup = best_effort_stale_unsubscribe(id, &endpoint, &mut websocket, &mut shutdown_rx, kind, remote_id).await;
WsActorIoOutcome::BestEffortUnsubscribe { kind, remote_id } => {
let cleanup = best_effort_remote_unsubscribe(id, &endpoint, &mut websocket, &mut shutdown_rx, kind, remote_id).await;
if let WsActorIoOutcome::ShutdownRequested { deadline } = cleanup {
close_session_actor(
id,
@@ -442,12 +467,21 @@ async fn run_ws_session_actor(
&mut subscriptions,
&mut remote_to_local,
continuity_gap_count,
overflow_count,
deadline,
)
.await;
return;
}
publish_snapshot(&snapshot_tx, id, &endpoint, crate::WsSessionState::Active, pending.len(), continuity_gap_count, &subscriptions);
publish_snapshot(
&snapshot_tx,
id,
&endpoint,
crate::WsSessionState::Active,
pending.len(),
WsRuntimeCounters { continuity_gap_count, overflow_count },
&subscriptions,
);
},
WsActorIoOutcome::ShutdownRequested { deadline } => {
close_session_actor(
@@ -459,6 +493,7 @@ async fn run_ws_session_actor(
&mut subscriptions,
&mut remote_to_local,
continuity_gap_count,
overflow_count,
deadline,
)
.await;
@@ -476,6 +511,7 @@ async fn run_ws_session_actor(
&mut subscriptions,
&mut remote_to_local,
&mut continuity_gap_count,
&mut overflow_count,
crate::ERROR_CODE_WS_CONNECTION_FAILED,
"Remote peer closed WebSocket session before pending response delivery",
)
@@ -491,16 +527,35 @@ async fn run_ws_session_actor(
&mut subscriptions,
&mut remote_to_local,
continuity_gap_count,
overflow_count,
deadline,
);
return;
},
WsReconnectOutcome::HandlesDropped => {
finish_disconnected_close(id, &endpoint, &snapshot_tx, &mut pending, &mut subscriptions, &mut remote_to_local, continuity_gap_count);
finish_disconnected_close(
id,
&endpoint,
&snapshot_tx,
&mut pending,
&mut subscriptions,
&mut remote_to_local,
continuity_gap_count,
overflow_count,
);
return;
},
WsReconnectOutcome::Exhausted => {
finish_reconnect_exhaustion(id, &endpoint, &snapshot_tx, &mut pending, &mut subscriptions, &mut remote_to_local, continuity_gap_count);
finish_reconnect_exhaustion(
id,
&endpoint,
&snapshot_tx,
&mut pending,
&mut subscriptions,
&mut remote_to_local,
continuity_gap_count,
overflow_count,
);
return;
},
}
@@ -517,6 +572,7 @@ async fn run_ws_session_actor(
&mut subscriptions,
&mut remote_to_local,
&mut continuity_gap_count,
&mut overflow_count,
code,
pending_message,
)
@@ -532,16 +588,35 @@ async fn run_ws_session_actor(
&mut subscriptions,
&mut remote_to_local,
continuity_gap_count,
overflow_count,
deadline,
);
return;
},
WsReconnectOutcome::HandlesDropped => {
finish_disconnected_close(id, &endpoint, &snapshot_tx, &mut pending, &mut subscriptions, &mut remote_to_local, continuity_gap_count);
finish_disconnected_close(
id,
&endpoint,
&snapshot_tx,
&mut pending,
&mut subscriptions,
&mut remote_to_local,
continuity_gap_count,
overflow_count,
);
return;
},
WsReconnectOutcome::Exhausted => {
finish_reconnect_exhaustion(id, &endpoint, &snapshot_tx, &mut pending, &mut subscriptions, &mut remote_to_local, continuity_gap_count);
finish_reconnect_exhaustion(
id,
&endpoint,
&snapshot_tx,
&mut pending,
&mut subscriptions,
&mut remote_to_local,
continuity_gap_count,
overflow_count,
);
return;
},
}
@@ -570,11 +645,12 @@ async fn recover_websocket_session(
subscriptions: &mut std::collections::BTreeMap<u64, crate::WsSubscriptionRuntime>,
remote_to_local: &mut std::collections::BTreeMap<u64, crate::WsSubscriptionId>,
continuity_gap_count: &mut u64,
overflow_count: &mut u64,
code: ksp_core_lib::ErrorCode,
pending_message: &'static str,
) -> WsReconnectOutcome {
fail_pending_for_reconnect(pending, id, code, pending_message, subscriptions, remote_to_local);
prepare_subscriptions_for_reconnect(endpoint.session().resubscribe(), subscriptions, remote_to_local);
prepare_subscriptions_for_reconnect(endpoint.session().resubscribe(), subscriptions, remote_to_local, code);
*continuity_gap_count = (*continuity_gap_count).saturating_add(1);
let max_retries = endpoint.session().reconnect().max_retries();
if max_retries == 0 {
@@ -582,7 +658,15 @@ async fn recover_websocket_session(
}
let mut attempt = 1_u32;
while attempt <= max_retries {
publish_snapshot(snapshot_tx, id, endpoint, crate::WsSessionState::Reconnecting { attempt }, pending.len(), *continuity_gap_count, subscriptions);
publish_snapshot(
snapshot_tx,
id,
endpoint,
crate::WsSessionState::Reconnecting { attempt },
pending.len(),
WsRuntimeCounters { continuity_gap_count: *continuity_gap_count, overflow_count: *overflow_count },
subscriptions,
);
ksp_logging_lib::warn!(
target: crate::TRACING_TARGET,
session_id = id.get(),
@@ -631,12 +715,21 @@ async fn recover_websocket_session(
subscriptions,
remote_to_local,
*continuity_gap_count,
overflow_count,
attempt,
)
.await;
match restore {
WsActorIoOutcome::Continue => {
publish_snapshot(snapshot_tx, id, endpoint, crate::WsSessionState::Active, pending.len(), *continuity_gap_count, subscriptions);
publish_snapshot(
snapshot_tx,
id,
endpoint,
crate::WsSessionState::Active,
pending.len(),
WsRuntimeCounters { continuity_gap_count: *continuity_gap_count, overflow_count: *overflow_count },
subscriptions,
);
ksp_logging_lib::debug!(
target: crate::TRACING_TARGET,
session_id = id.get(),
@@ -657,13 +750,13 @@ async fn recover_websocket_session(
subscriptions,
remote_to_local,
);
prepare_subscriptions_for_reconnect(endpoint.session().resubscribe(), subscriptions, remote_to_local);
prepare_subscriptions_for_reconnect(endpoint.session().resubscribe(), subscriptions, remote_to_local, crate::ERROR_CODE_WS_CONNECTION_FAILED);
},
WsActorIoOutcome::Failed { code: restore_code, pending_message: restore_message } => {
fail_pending_for_reconnect(pending, id, restore_code, restore_message, subscriptions, remote_to_local);
prepare_subscriptions_for_reconnect(endpoint.session().resubscribe(), subscriptions, remote_to_local);
prepare_subscriptions_for_reconnect(endpoint.session().resubscribe(), subscriptions, remote_to_local, restore_code);
},
WsActorIoOutcome::StaleSubscribeAck { .. } => {},
WsActorIoOutcome::BestEffortUnsubscribe { .. } => {},
}
attempt = attempt.saturating_add(1);
}
@@ -683,6 +776,7 @@ async fn restore_subscriptions_after_reconnect(
subscriptions: &mut std::collections::BTreeMap<u64, crate::WsSubscriptionRuntime>,
remote_to_local: &mut std::collections::BTreeMap<u64, crate::WsSubscriptionId>,
continuity_gap_count: u64,
overflow_count: &mut u64,
attempt: u32,
) -> WsActorIoOutcome {
let restore_ids = subscriptions
@@ -700,13 +794,14 @@ async fn restore_subscriptions_after_reconnect(
let (request_id, deadline) = match write_result {
std::result::Result::Ok(result) => result,
std::result::Result::Err((error, WsActorIoOutcome::Continue)) => {
close_local_subscription(subscriptions, remote_to_local, subscription_id, crate::WsSubscriptionState::Failed);
let error_code = error.code();
fail_local_subscription(subscriptions, remote_to_local, subscription_id, error_code);
ksp_logging_lib::warn!(
target: crate::TRACING_TARGET,
session_id = id.get(),
subscription_id = subscription_id.get(),
subscription_kind = kind.as_str(),
error_code = error.code().code(),
error_code = error_code.code(),
"logical WebSocket resubscribe could not be queued"
);
continue;
@@ -752,7 +847,7 @@ async fn restore_subscriptions_after_reconnect(
}
},
maybe_message = websocket.next() => {
handle_socket_message(id, endpoint, maybe_message, websocket, pending, subscriptions, remote_to_local, shutdown_rx).await
handle_socket_message(id, endpoint, maybe_message, websocket, pending, subscriptions, remote_to_local, overflow_count, shutdown_rx).await
},
() = tokio::time::sleep_until(timeout_deadline) => {
expire_pending_requests(id, pending, subscriptions, remote_to_local);
@@ -761,15 +856,23 @@ async fn restore_subscriptions_after_reconnect(
};
match outcome {
WsActorIoOutcome::Continue => {},
WsActorIoOutcome::StaleSubscribeAck { kind: stale_kind, remote_id } => {
let cleanup = best_effort_stale_unsubscribe(id, endpoint, websocket, shutdown_rx, stale_kind, remote_id).await;
WsActorIoOutcome::BestEffortUnsubscribe { kind: stale_kind, remote_id } => {
let cleanup = best_effort_remote_unsubscribe(id, endpoint, websocket, shutdown_rx, stale_kind, remote_id).await;
if let WsActorIoOutcome::ShutdownRequested { deadline } = cleanup {
return WsActorIoOutcome::ShutdownRequested { deadline };
}
},
_ => return outcome,
}
publish_snapshot(snapshot_tx, id, endpoint, crate::WsSessionState::Reconnecting { attempt }, pending.len(), continuity_gap_count, subscriptions);
publish_snapshot(
snapshot_tx,
id,
endpoint,
crate::WsSessionState::Reconnecting { attempt },
pending.len(),
WsRuntimeCounters { continuity_gap_count, overflow_count: *overflow_count },
subscriptions,
);
}
}
return WsActorIoOutcome::Continue;
@@ -791,7 +894,7 @@ async fn handle_reconnecting_command_with_socket(
});
close_local_subscription(subscriptions, remote_to_local, subscription_id, crate::WsSubscriptionState::Closed);
if let std::option::Option::Some((kind, remote_id)) = cleanup {
let cleanup_outcome = best_effort_stale_unsubscribe(id, endpoint, websocket, shutdown_rx, kind, remote_id).await;
let cleanup_outcome = best_effort_remote_unsubscribe(id, endpoint, websocket, shutdown_rx, kind, remote_id).await;
if let WsActorIoOutcome::ShutdownRequested { deadline } = cleanup_outcome {
let _ = response_tx.send(std::result::Result::Ok(false));
return WsActorIoOutcome::ShutdownRequested { deadline };
@@ -943,9 +1046,11 @@ fn prepare_subscriptions_for_reconnect(
policy: crate::WsResubscribePolicy,
subscriptions: &mut std::collections::BTreeMap<u64, crate::WsSubscriptionRuntime>,
remote_to_local: &mut std::collections::BTreeMap<u64, crate::WsSubscriptionId>,
failure_code: ksp_core_lib::ErrorCode,
) {
remote_to_local.clear();
let mut terminal = std::vec::Vec::new();
let mut failed = std::vec::Vec::new();
let mut already_terminal = std::vec::Vec::new();
for runtime in subscriptions.values_mut() {
runtime.remote_id = std::option::Option::None;
match runtime.state {
@@ -953,19 +1058,18 @@ fn prepare_subscriptions_for_reconnect(
if policy == crate::WsResubscribePolicy::ActiveSubscriptions {
runtime.set_state(crate::WsSubscriptionState::Resubscribing);
} else {
runtime.set_state(crate::WsSubscriptionState::Failed);
terminal.push(runtime.id);
failed.push(runtime.id);
}
},
crate::WsSubscriptionState::Requested | crate::WsSubscriptionState::Cancelling => {
runtime.set_state(crate::WsSubscriptionState::Failed);
terminal.push(runtime.id);
},
crate::WsSubscriptionState::Closed | crate::WsSubscriptionState::Failed => terminal.push(runtime.id),
crate::WsSubscriptionState::Requested | crate::WsSubscriptionState::Cancelling => failed.push(runtime.id),
crate::WsSubscriptionState::Closed | crate::WsSubscriptionState::Failed => already_terminal.push(runtime.id),
}
}
for subscription_id in terminal {
close_local_subscription(subscriptions, remote_to_local, subscription_id, crate::WsSubscriptionState::Failed);
for subscription_id in failed {
fail_local_subscription(subscriptions, remote_to_local, subscription_id, failure_code);
}
for subscription_id in already_terminal {
subscriptions.remove(&subscription_id.get());
}
}
@@ -998,7 +1102,7 @@ fn fail_pending_for_reconnect(
let _ = response_tx.send(std::result::Result::Err(error));
},
PendingWsResponse::Subscribe { subscription_id, response_tx } => {
close_local_subscription(subscriptions, remote_to_local, subscription_id, crate::WsSubscriptionState::Failed);
fail_local_subscription(subscriptions, remote_to_local, subscription_id, code);
let _ = response_tx.send(std::result::Result::Err(error));
},
PendingWsResponse::Resubscribe { .. } => {},
@@ -1010,7 +1114,7 @@ fn fail_pending_for_reconnect(
}
}
async fn best_effort_stale_unsubscribe(
async fn best_effort_remote_unsubscribe(
id: crate::WsSessionId,
endpoint: &crate::WsEndpointSettings,
websocket: &mut WsPhysicalStream,
@@ -1041,7 +1145,7 @@ async fn best_effort_stale_unsubscribe(
target: crate::TRACING_TARGET,
session_id = id.get(),
subscription_kind = kind.as_str(),
"sent best-effort unsubscribe for stale remote subscription acknowledgement"
"sent best-effort remote WebSocket unsubscribe"
);
}
return WsActorIoOutcome::Continue;
@@ -1055,12 +1159,21 @@ fn finish_disconnected_shutdown(
subscriptions: &mut std::collections::BTreeMap<u64, crate::WsSubscriptionRuntime>,
remote_to_local: &mut std::collections::BTreeMap<u64, crate::WsSubscriptionId>,
continuity_gap_count: u64,
overflow_count: u64,
_deadline: tokio::time::Instant,
) {
publish_snapshot(snapshot_tx, id, endpoint, crate::WsSessionState::Closing, pending.len(), continuity_gap_count, subscriptions);
publish_snapshot(
snapshot_tx,
id,
endpoint,
crate::WsSessionState::Closing,
pending.len(),
WsRuntimeCounters { continuity_gap_count, overflow_count },
subscriptions,
);
fail_all_pending(pending, id, crate::ERROR_CODE_WS_SESSION_CLOSED, "WebSocket session shutdown cancelled pending request");
terminate_all_subscriptions(subscriptions, remote_to_local, crate::WsSubscriptionState::Closed);
publish_snapshot(snapshot_tx, id, endpoint, crate::WsSessionState::Closed, 0, continuity_gap_count, subscriptions);
publish_snapshot(snapshot_tx, id, endpoint, crate::WsSessionState::Closed, 0, WsRuntimeCounters { continuity_gap_count, overflow_count }, subscriptions);
}
fn finish_disconnected_close(
@@ -1071,10 +1184,11 @@ fn finish_disconnected_close(
subscriptions: &mut std::collections::BTreeMap<u64, crate::WsSubscriptionRuntime>,
remote_to_local: &mut std::collections::BTreeMap<u64, crate::WsSubscriptionId>,
continuity_gap_count: u64,
overflow_count: u64,
) {
fail_all_pending(pending, id, crate::ERROR_CODE_WS_SESSION_CLOSED, "WebSocket session handles were dropped during reconnect");
terminate_all_subscriptions(subscriptions, remote_to_local, crate::WsSubscriptionState::Closed);
publish_snapshot(snapshot_tx, id, endpoint, crate::WsSessionState::Closed, 0, continuity_gap_count, subscriptions);
publish_snapshot(snapshot_tx, id, endpoint, crate::WsSessionState::Closed, 0, WsRuntimeCounters { continuity_gap_count, overflow_count }, subscriptions);
}
fn finish_reconnect_exhaustion(
@@ -1085,10 +1199,11 @@ fn finish_reconnect_exhaustion(
subscriptions: &mut std::collections::BTreeMap<u64, crate::WsSubscriptionRuntime>,
remote_to_local: &mut std::collections::BTreeMap<u64, crate::WsSubscriptionId>,
continuity_gap_count: u64,
overflow_count: u64,
) {
fail_all_pending(pending, id, crate::ERROR_CODE_WS_CONNECTION_FAILED, "WebSocket reconnect budget was exhausted");
terminate_all_subscriptions(subscriptions, remote_to_local, crate::WsSubscriptionState::Failed);
publish_snapshot(snapshot_tx, id, endpoint, crate::WsSessionState::Failed, 0, continuity_gap_count, subscriptions);
fail_all_subscriptions(subscriptions, remote_to_local, crate::ERROR_CODE_WS_CONNECTION_FAILED);
publish_snapshot(snapshot_tx, id, endpoint, crate::WsSessionState::Failed, 0, WsRuntimeCounters { continuity_gap_count, overflow_count }, subscriptions);
ksp_logging_lib::warn!(
target: crate::TRACING_TARGET,
session_id = id.get(),
@@ -1154,6 +1269,7 @@ where
},
};
let (state_tx, _) = tokio::sync::watch::channel(crate::WsSubscriptionState::Requested);
let (terminal_error_tx, _) = tokio::sync::watch::channel(std::option::Option::None::<ksp_core_lib::ErrorCode>);
let resubscribe_params = params.clone();
subscriptions.insert(
subscription_id.get(),
@@ -1164,6 +1280,7 @@ where
params: resubscribe_params,
remote_id: std::option::Option::None,
state_tx,
terminal_error_tx,
dispatcher,
},
);
@@ -1182,9 +1299,8 @@ where
WsActorIoOutcome::Continue
},
std::result::Result::Err((error, outcome)) => {
if let std::option::Option::Some(mut runtime) = subscriptions.remove(&subscription_id.get()) {
runtime.set_state(crate::WsSubscriptionState::Failed);
}
let error_code = error.code();
fail_local_subscription(subscriptions, remote_to_local, subscription_id, error_code);
let _ = response_tx.send(std::result::Result::Err(error));
outcome
},
@@ -1367,6 +1483,7 @@ async fn handle_socket_message<S>(
pending: &mut std::collections::BTreeMap<u64, PendingWsRequest>,
subscriptions: &mut std::collections::BTreeMap<u64, crate::WsSubscriptionRuntime>,
remote_to_local: &mut std::collections::BTreeMap<u64, crate::WsSubscriptionId>,
overflow_count: &mut u64,
shutdown_rx: &mut tokio::sync::watch::Receiver<std::option::Option<tokio::time::Instant>>,
) -> WsActorIoOutcome
where
@@ -1399,7 +1516,7 @@ where
},
};
return match message {
tokio_tungstenite::tungstenite::Message::Text(text) => handle_text_message(id, text.as_str(), pending, subscriptions, remote_to_local),
tokio_tungstenite::tungstenite::Message::Text(text) => handle_text_message(id, text.as_str(), pending, subscriptions, remote_to_local, overflow_count),
tokio_tungstenite::tungstenite::Message::Binary(_) => {
ksp_logging_lib::warn!(target: crate::TRACING_TARGET, session_id = id.get(), "received unexpected binary WebSocket message");
WsActorIoOutcome::Failed {
@@ -1453,6 +1570,7 @@ fn handle_text_message(
pending: &mut std::collections::BTreeMap<u64, PendingWsRequest>,
subscriptions: &mut std::collections::BTreeMap<u64, crate::WsSubscriptionRuntime>,
remote_to_local: &mut std::collections::BTreeMap<u64, crate::WsSubscriptionId>,
overflow_count: &mut u64,
) -> WsActorIoOutcome {
let decoded = serde_json::from_str::<serde_json::Value>(text);
let value = match decoded {
@@ -1473,7 +1591,7 @@ fn handle_text_message(
},
};
if !object.contains_key("id") {
return handle_subscription_notification(id, object, subscriptions, remote_to_local);
return handle_subscription_notification(id, object, subscriptions, remote_to_local, overflow_count);
}
let response_id = match object.get("id").and_then(serde_json::Value::as_u64) {
std::option::Option::Some(response_id) => response_id,
@@ -1538,7 +1656,7 @@ fn dispatch_pending_response(
let remote_id = match value.as_u64() {
std::option::Option::Some(remote_id) => remote_id,
std::option::Option::None => {
close_local_subscription(subscriptions, remote_to_local, subscription_id, crate::WsSubscriptionState::Failed);
fail_local_subscription(subscriptions, remote_to_local, subscription_id, crate::ERROR_CODE_WS_PROTOCOL_ERROR);
let error = ksp_core_lib::Error::new(
crate::ERROR_CODE_WS_PROTOCOL_ERROR,
"WebSocket subscribe response did not contain a numeric remote subscription id",
@@ -1553,7 +1671,7 @@ fn dispatch_pending_response(
},
};
if remote_to_local.contains_key(&remote_id) {
close_local_subscription(subscriptions, remote_to_local, subscription_id, crate::WsSubscriptionState::Failed);
fail_local_subscription(subscriptions, remote_to_local, subscription_id, crate::ERROR_CODE_WS_PROTOCOL_ERROR);
let error = ksp_core_lib::Error::new(crate::ERROR_CODE_WS_PROTOCOL_ERROR, "WebSocket endpoint reused an active remote subscription id")
.with_context("session_id", id.get().to_string())
.with_context("subscription_id", subscription_id.get().to_string());
@@ -1568,7 +1686,12 @@ fn dispatch_pending_response(
runtime.remote_id = std::option::Option::Some(remote_id);
runtime.set_state(crate::WsSubscriptionState::Active);
remote_to_local.insert(remote_id, subscription_id);
crate::WsSubscriptionRegistration::new(subscription_id, runtime.kind, runtime.state_tx.subscribe())
crate::WsSubscriptionRegistration::new(
subscription_id,
runtime.kind,
runtime.state_tx.subscribe(),
runtime.terminal_error_tx.subscribe(),
)
},
std::option::Option::None => {
let error = ksp_core_lib::Error::new(
@@ -1593,7 +1716,8 @@ fn dispatch_pending_response(
);
},
std::result::Result::Err(error) => {
close_local_subscription(subscriptions, remote_to_local, subscription_id, crate::WsSubscriptionState::Failed);
let error_code = error.code();
fail_local_subscription(subscriptions, remote_to_local, subscription_id, error_code);
let _ = response_tx.send(std::result::Result::Err(error.with_context("method", pending_request.method)));
},
},
@@ -1602,7 +1726,7 @@ fn dispatch_pending_response(
let remote_id = match value.as_u64() {
std::option::Option::Some(remote_id) => remote_id,
std::option::Option::None => {
close_local_subscription(subscriptions, remote_to_local, subscription_id, crate::WsSubscriptionState::Failed);
fail_local_subscription(subscriptions, remote_to_local, subscription_id, crate::ERROR_CODE_WS_PROTOCOL_ERROR);
return WsActorIoOutcome::Failed {
code: crate::ERROR_CODE_WS_PROTOCOL_ERROR,
pending_message: "WebSocket resubscribe response violated protocol invariants",
@@ -1610,7 +1734,7 @@ fn dispatch_pending_response(
},
};
if remote_to_local.contains_key(&remote_id) {
close_local_subscription(subscriptions, remote_to_local, subscription_id, crate::WsSubscriptionState::Failed);
fail_local_subscription(subscriptions, remote_to_local, subscription_id, crate::ERROR_CODE_WS_PROTOCOL_ERROR);
return WsActorIoOutcome::Failed {
code: crate::ERROR_CODE_WS_PROTOCOL_ERROR,
pending_message: "WebSocket endpoint reused an active remote subscription id during resubscribe",
@@ -1626,7 +1750,7 @@ fn dispatch_pending_response(
subscription_kind = kind.as_str(),
"late WebSocket resubscribe acknowledgement lost to local cancellation"
);
return WsActorIoOutcome::StaleSubscribeAck { kind, remote_id };
return WsActorIoOutcome::BestEffortUnsubscribe { kind, remote_id };
}
if let std::option::Option::Some(runtime) = subscriptions.get_mut(&subscription_id.get()) {
runtime.remote_id = std::option::Option::Some(remote_id);
@@ -1641,13 +1765,15 @@ fn dispatch_pending_response(
"logical WebSocket subscription restored with a new remote binding"
);
},
std::result::Result::Err(_) => {
close_local_subscription(subscriptions, remote_to_local, subscription_id, crate::WsSubscriptionState::Failed);
std::result::Result::Err(error) => {
let error_code = error.code();
fail_local_subscription(subscriptions, remote_to_local, subscription_id, error_code);
ksp_logging_lib::warn!(
target: crate::TRACING_TARGET,
session_id = id.get(),
subscription_id = subscription_id.get(),
subscription_kind = kind.as_str(),
error_code = error_code.code(),
"remote WebSocket resubscribe returned an application error"
);
},
@@ -1700,6 +1826,7 @@ fn handle_subscription_notification(
object: &serde_json::Map<std::string::String, serde_json::Value>,
subscriptions: &mut std::collections::BTreeMap<u64, crate::WsSubscriptionRuntime>,
remote_to_local: &mut std::collections::BTreeMap<u64, crate::WsSubscriptionId>,
overflow_count: &mut u64,
) -> WsActorIoOutcome {
if object.get("jsonrpc").and_then(serde_json::Value::as_str) != std::option::Option::Some("2.0") {
return WsActorIoOutcome::Failed {
@@ -1767,38 +1894,50 @@ fn handle_subscription_notification(
subscription_kind = subscription_kind.as_str(),
"WebSocket notification method mismatched the registered subscription family"
);
close_local_subscription(subscriptions, remote_to_local, local_id, crate::WsSubscriptionState::Failed);
return WsActorIoOutcome::Continue;
fail_local_subscription(subscriptions, remote_to_local, local_id, crate::ERROR_CODE_WS_PROTOCOL_ERROR);
return WsActorIoOutcome::BestEffortUnsubscribe { kind: subscription_kind, remote_id };
}
let subscription_kind = runtime.kind;
let dispatch = (runtime.dispatcher)(result);
match dispatch {
crate::WsNotificationDispatchOutcome::Delivered => {},
return match dispatch {
crate::WsNotificationDispatchOutcome::Delivered => WsActorIoOutcome::Continue,
crate::WsNotificationDispatchOutcome::ReceiverClosed => {
close_local_subscription(subscriptions, remote_to_local, local_id, crate::WsSubscriptionState::Closed);
ksp_logging_lib::debug!(
target: crate::TRACING_TARGET,
session_id = id.get(),
subscription_id = local_id.get(),
subscription_kind = subscription_kind.as_str(),
"logical WebSocket subscription receiver was dropped; scheduling remote cleanup"
);
WsActorIoOutcome::BestEffortUnsubscribe { kind: subscription_kind, remote_id }
},
crate::WsNotificationDispatchOutcome::QueueFull => {
*overflow_count = (*overflow_count).saturating_add(1);
ksp_logging_lib::warn!(
target: crate::TRACING_TARGET,
session_id = id.get(),
subscription_id = local_id.get(),
subscription_kind = subscription_kind.as_str(),
"bounded WebSocket subscription notification queue is full"
overflow_count = *overflow_count,
"bounded WebSocket subscription notification queue overflowed; failing only the slow subscription"
);
close_local_subscription(subscriptions, remote_to_local, local_id, crate::WsSubscriptionState::Failed);
fail_local_subscription(subscriptions, remote_to_local, local_id, crate::ERROR_CODE_WS_BACKPRESSURE_OVERFLOW);
WsActorIoOutcome::BestEffortUnsubscribe { kind: subscription_kind, remote_id }
},
crate::WsNotificationDispatchOutcome::DecodeFailed => {
crate::WsNotificationDispatchOutcome::DecodeFailed { code } => {
ksp_logging_lib::warn!(
target: crate::TRACING_TARGET,
session_id = id.get(),
subscription_id = local_id.get(),
subscription_kind = subscription_kind.as_str(),
error_code = code.code(),
"typed WebSocket notification decoding failed for one subscription"
);
close_local_subscription(subscriptions, remote_to_local, local_id, crate::WsSubscriptionState::Failed);
fail_local_subscription(subscriptions, remote_to_local, local_id, code);
WsActorIoOutcome::BestEffortUnsubscribe { kind: subscription_kind, remote_id }
},
}
return WsActorIoOutcome::Continue;
};
}
async fn close_session_actor<S>(
@@ -1810,14 +1949,23 @@ async fn close_session_actor<S>(
subscriptions: &mut std::collections::BTreeMap<u64, crate::WsSubscriptionRuntime>,
remote_to_local: &mut std::collections::BTreeMap<u64, crate::WsSubscriptionId>,
continuity_gap_count: u64,
overflow_count: u64,
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(), continuity_gap_count, subscriptions);
publish_snapshot(
snapshot_tx,
id,
endpoint,
crate::WsSessionState::Closing,
pending.len(),
WsRuntimeCounters { continuity_gap_count, overflow_count },
subscriptions,
);
fail_all_pending(pending, id, crate::ERROR_CODE_WS_SESSION_CLOSED, "WebSocket session shutdown cancelled the pending request");
terminate_all_subscriptions(subscriptions, remote_to_local, crate::WsSubscriptionState::Closed);
publish_snapshot(snapshot_tx, id, endpoint, crate::WsSessionState::Closing, 0, continuity_gap_count, subscriptions);
publish_snapshot(snapshot_tx, id, endpoint, crate::WsSessionState::Closing, 0, WsRuntimeCounters { continuity_gap_count, overflow_count }, subscriptions);
ksp_logging_lib::debug!(
target: crate::TRACING_TARGET,
session_id = id.get(),
@@ -1839,7 +1987,7 @@ async fn close_session_actor<S>(
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, continuity_gap_count, subscriptions);
publish_snapshot(snapshot_tx, id, endpoint, crate::WsSessionState::Closed, 0, WsRuntimeCounters { continuity_gap_count, overflow_count }, subscriptions);
ksp_logging_lib::debug!(target: crate::TRACING_TARGET, session_id = id.get(), endpoint_name = endpoint.name(), "physical WebSocket session is closed");
}
@@ -1921,11 +2069,11 @@ fn expire_pending_response(
let _ = response_tx.send(std::result::Result::Err(error));
},
PendingWsResponse::Subscribe { subscription_id, response_tx } => {
close_local_subscription(subscriptions, remote_to_local, subscription_id, crate::WsSubscriptionState::Failed);
fail_local_subscription(subscriptions, remote_to_local, subscription_id, crate::ERROR_CODE_TIMEOUT);
let _ = response_tx.send(std::result::Result::Err(error));
},
PendingWsResponse::Resubscribe { subscription_id, .. } => {
close_local_subscription(subscriptions, remote_to_local, subscription_id, crate::WsSubscriptionState::Failed);
fail_local_subscription(subscriptions, remote_to_local, subscription_id, crate::ERROR_CODE_TIMEOUT);
},
PendingWsResponse::Unsubscribe { subscription_id, response_tx } => {
close_local_subscription(subscriptions, remote_to_local, subscription_id, crate::WsSubscriptionState::Closed);
@@ -2026,6 +2174,32 @@ fn terminate_all_subscriptions(
remote_to_local.clear();
}
fn fail_all_subscriptions(
subscriptions: &mut std::collections::BTreeMap<u64, crate::WsSubscriptionRuntime>,
remote_to_local: &mut std::collections::BTreeMap<u64, crate::WsSubscriptionId>,
error_code: ksp_core_lib::ErrorCode,
) {
for runtime in subscriptions.values_mut() {
runtime.remote_id = std::option::Option::None;
runtime.fail_with_code(error_code);
}
remote_to_local.clear();
}
fn fail_local_subscription(
subscriptions: &mut std::collections::BTreeMap<u64, crate::WsSubscriptionRuntime>,
remote_to_local: &mut std::collections::BTreeMap<u64, crate::WsSubscriptionId>,
subscription_id: crate::WsSubscriptionId,
error_code: ksp_core_lib::ErrorCode,
) {
if let std::option::Option::Some(mut runtime) = subscriptions.remove(&subscription_id.get()) {
if let std::option::Option::Some(remote_id) = runtime.remote_id.take() {
remote_to_local.remove(&remote_id);
}
runtime.fail_with_code(error_code);
}
}
fn close_local_subscription(
subscriptions: &mut std::collections::BTreeMap<u64, crate::WsSubscriptionRuntime>,
remote_to_local: &mut std::collections::BTreeMap<u64, crate::WsSubscriptionId>,
@@ -2046,7 +2220,7 @@ fn publish_snapshot(
endpoint: &crate::WsEndpointSettings,
state: crate::WsSessionState,
pending_request_count: usize,
continuity_gap_count: u64,
counters: WsRuntimeCounters,
subscriptions: &std::collections::BTreeMap<u64, crate::WsSubscriptionRuntime>,
) {
let subscription_snapshots = subscriptions.values().map(crate::WsSubscriptionRuntime::snapshot).collect();
@@ -2058,8 +2232,8 @@ fn publish_snapshot(
endpoint.protocol(),
state,
pending_request_count,
continuity_gap_count,
0,
counters.continuity_gap_count,
counters.overflow_count,
subscription_snapshots,
);
snapshot_tx.send_replace(snapshot);