574 lines
33 KiB
Rust
574 lines
33 KiB
Rust
// file: crates/ksp-onchain-transport-lib/unit_tests/ws_session.rs
|
|
// version: 4
|
|
|
|
use futures_util::SinkExt; // rust-rules: trait-import
|
|
use futures_util::StreamExt; // rust-rules: trait-import
|
|
|
|
fn local_endpoint(url: &str) -> crate::WsEndpointSettings {
|
|
return local_endpoint_with_session(url, crate::WsSessionSettings::default());
|
|
}
|
|
|
|
fn local_endpoint_with_session(url: &str, session: crate::WsSessionSettings) -> crate::WsEndpointSettings {
|
|
return crate::WsEndpointSettings::new(
|
|
"local_ws",
|
|
true,
|
|
crate::WsProviderName::new("local-fixture"),
|
|
crate::WsClusterName::new("local"),
|
|
crate::WsProtocolKind::SolanaStandard,
|
|
crate::WsEndpointUrl::parse(url).expect("local test WebSocket URL must parse"),
|
|
session,
|
|
);
|
|
}
|
|
|
|
fn session_settings(
|
|
command_timeout: std::time::Duration,
|
|
close_timeout: std::time::Duration,
|
|
max_pending_requests: usize,
|
|
max_message_size_bytes: usize,
|
|
max_frame_size_bytes: usize,
|
|
max_write_buffer_size_bytes: usize,
|
|
) -> crate::WsSessionSettings {
|
|
let defaults = crate::WsSessionSettings::default();
|
|
return crate::WsSessionSettings::new(
|
|
command_timeout,
|
|
close_timeout,
|
|
defaults.reconnect().clone(),
|
|
defaults.resubscribe(),
|
|
defaults.command_queue_capacity(),
|
|
defaults.notification_queue_capacity(),
|
|
defaults.max_active_subscriptions(),
|
|
max_pending_requests,
|
|
max_message_size_bytes,
|
|
max_frame_size_bytes,
|
|
max_write_buffer_size_bytes,
|
|
);
|
|
}
|
|
|
|
async fn bind_local_listener() -> (tokio::net::TcpListener, std::string::String) {
|
|
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.expect("local listener must bind");
|
|
let address = listener.local_addr().expect("local listener must expose address");
|
|
return (listener, format!("ws://{address}"));
|
|
}
|
|
|
|
async fn read_request(websocket: &mut tokio_tungstenite::WebSocketStream<tokio::net::TcpStream>) -> serde_json::Value {
|
|
let message = websocket.next().await.expect("request message must exist").expect("request message must decode");
|
|
let text = message.to_text().expect("request must be text");
|
|
return serde_json::from_str(text).expect("request must contain JSON");
|
|
}
|
|
|
|
async fn send_result(websocket: &mut tokio_tungstenite::WebSocketStream<tokio::net::TcpStream>, request: &serde_json::Value, result: serde_json::Value) {
|
|
let id = request.get("id").and_then(serde_json::Value::as_u64).expect("request id must be numeric");
|
|
let response = serde_json::json!({"jsonrpc":"2.0","id":id,"result":result});
|
|
websocket.send(tokio_tungstenite::tungstenite::Message::Text(response.to_string().into())).await.expect("local response must send");
|
|
}
|
|
|
|
async fn wait_for_state(session: &crate::WsSession, expected: crate::WsSessionState) {
|
|
let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(1);
|
|
loop {
|
|
if session.state() == expected {
|
|
return;
|
|
}
|
|
assert!(tokio::time::Instant::now() < deadline, "session did not reach expected state: {expected:?}");
|
|
tokio::time::sleep(std::time::Duration::from_millis(5)).await;
|
|
}
|
|
}
|
|
|
|
async fn wait_for_pending_count(session: &crate::WsSession, expected: usize) {
|
|
let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(1);
|
|
loop {
|
|
if session.snapshot().pending_request_count() == expected {
|
|
return;
|
|
}
|
|
assert!(tokio::time::Instant::now() < deadline, "session did not reach pending count {expected}");
|
|
tokio::time::sleep(std::time::Duration::from_millis(5)).await;
|
|
}
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn websocket_session_connects_and_round_trips_internal_json_rpc() {
|
|
let (listener, url) = bind_local_listener().await;
|
|
let server = tokio::spawn(async move {
|
|
let (stream, _) = listener.accept().await.expect("local server must accept client");
|
|
let mut websocket = tokio_tungstenite::accept_async(stream).await.expect("local WebSocket handshake must succeed");
|
|
let request = read_request(&mut websocket).await;
|
|
assert_eq!(request.get("method").and_then(serde_json::Value::as_str), std::option::Option::Some("getVersion"));
|
|
send_result(&mut websocket, &request, serde_json::json!({"solana-core":"fixture"})).await;
|
|
});
|
|
let session = crate::WsSession::connect(local_endpoint(url.as_str())).await.expect("client handshake must succeed");
|
|
assert_eq!(session.state(), crate::WsSessionState::Active);
|
|
assert_eq!(session.snapshot().pending_request_count(), 0);
|
|
let result = session.execute_json_rpc("getVersion", std::vec::Vec::new()).await.expect("fixture JSON-RPC call must succeed");
|
|
assert_eq!(result.get("solana-core").and_then(serde_json::Value::as_str), std::option::Option::Some("fixture"));
|
|
tokio::task::yield_now().await;
|
|
assert_eq!(session.snapshot().pending_request_count(), 0);
|
|
server.await.expect("local server task must complete");
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn websocket_sessions_are_physical_and_distinct_even_for_the_same_url() {
|
|
let (listener, url) = bind_local_listener().await;
|
|
let server = tokio::spawn(async move {
|
|
let first = listener.accept().await.expect("first local client must connect");
|
|
let first_ws = tokio_tungstenite::accept_async(first.0).await.expect("first handshake must succeed");
|
|
let second = listener.accept().await.expect("second local client must connect");
|
|
let second_ws = tokio_tungstenite::accept_async(second.0).await.expect("second handshake must succeed");
|
|
return (first_ws, second_ws);
|
|
});
|
|
let first = crate::WsSession::connect(local_endpoint(url.as_str())).await.expect("first client must connect");
|
|
let second = crate::WsSession::connect(local_endpoint(url.as_str())).await.expect("second client must connect");
|
|
assert_ne!(first.id(), second.id());
|
|
assert_eq!(first.snapshot().endpoint_name(), second.snapshot().endpoint_name());
|
|
let _ = server.await.expect("local server task must complete");
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn websocket_pending_map_dispatches_out_of_order_responses_by_request_id() {
|
|
let (listener, url) = bind_local_listener().await;
|
|
let server = tokio::spawn(async move {
|
|
let (stream, _) = listener.accept().await.expect("local server must accept client");
|
|
let mut websocket = tokio_tungstenite::accept_async(stream).await.expect("local WebSocket handshake must succeed");
|
|
let first = read_request(&mut websocket).await;
|
|
let second = read_request(&mut websocket).await;
|
|
send_result(&mut websocket, &second, serde_json::json!("second")).await;
|
|
send_result(&mut websocket, &first, serde_json::json!("first")).await;
|
|
});
|
|
let session = crate::WsSession::connect(local_endpoint(url.as_str())).await.expect("client handshake must succeed");
|
|
let first_session = session.clone();
|
|
let second_session = session.clone();
|
|
let (first, second) = tokio::join!(
|
|
first_session.execute_json_rpc("firstMethod", std::vec::Vec::new()),
|
|
second_session.execute_json_rpc("secondMethod", std::vec::Vec::new())
|
|
);
|
|
assert_eq!(first.expect("first response must dispatch"), serde_json::json!("first"));
|
|
assert_eq!(second.expect("second response must dispatch"), serde_json::json!("second"));
|
|
tokio::task::yield_now().await;
|
|
assert_eq!(session.snapshot().pending_request_count(), 0);
|
|
server.await.expect("local server task must complete");
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn websocket_rpc_application_error_does_not_fail_the_physical_session() {
|
|
let (listener, url) = bind_local_listener().await;
|
|
let server = tokio::spawn(async move {
|
|
let (stream, _) = listener.accept().await.expect("local server must accept client");
|
|
let mut websocket = tokio_tungstenite::accept_async(stream).await.expect("local WebSocket handshake must succeed");
|
|
let request = read_request(&mut websocket).await;
|
|
let id = request.get("id").and_then(serde_json::Value::as_u64).expect("request id must be numeric");
|
|
let response = serde_json::json!({"jsonrpc":"2.0","id":id,"error":{"code":-32602,"message":"fixture invalid params"}});
|
|
websocket.send(tokio_tungstenite::tungstenite::Message::Text(response.to_string().into())).await.expect("local error response must send");
|
|
let request = read_request(&mut websocket).await;
|
|
send_result(&mut websocket, &request, serde_json::json!(true)).await;
|
|
});
|
|
let session = crate::WsSession::connect(local_endpoint(url.as_str())).await.expect("client handshake must succeed");
|
|
let error = session.execute_json_rpc("badMethod", std::vec::Vec::new()).await.expect_err("RPC application error must surface");
|
|
assert_eq!(error.code(), crate::ERROR_CODE_RPC_APPLICATION_ERROR);
|
|
assert_eq!(session.state(), crate::WsSessionState::Active);
|
|
let result = session.execute_json_rpc("goodMethod", std::vec::Vec::new()).await.expect("session must remain usable");
|
|
assert_eq!(result, serde_json::json!(true));
|
|
server.await.expect("local server task must complete");
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn websocket_connection_errors_do_not_echo_url_credentials() {
|
|
let defaults = crate::WsSessionSettings::default();
|
|
let short_session = crate::WsSessionSettings::new(
|
|
std::time::Duration::from_millis(150),
|
|
defaults.close_timeout(),
|
|
defaults.reconnect().clone(),
|
|
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(),
|
|
);
|
|
let endpoint = crate::WsEndpointSettings::new(
|
|
"redaction_fixture",
|
|
true,
|
|
crate::WsProviderName::new("fixture-provider"),
|
|
crate::WsClusterName::new("local"),
|
|
crate::WsProtocolKind::SolanaStandard,
|
|
crate::WsEndpointUrl::parse("ws://user:password@127.0.0.1:1/private?api-key=SECRET-CANARY").expect("test URL must parse"),
|
|
short_session,
|
|
);
|
|
let error = crate::WsSession::connect(endpoint).await.expect_err("unreachable local endpoint must fail");
|
|
let rendered = format!("{error:?}");
|
|
assert!(!rendered.contains("SECRET-CANARY"));
|
|
assert!(!rendered.contains("password"));
|
|
assert!(!rendered.contains("private?"));
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn websocket_pending_request_capacity_rejects_only_the_excess_request() {
|
|
let (listener, url) = bind_local_listener().await;
|
|
let (first_seen_tx, first_seen_rx) = tokio::sync::oneshot::channel();
|
|
let (release_tx, release_rx) = tokio::sync::oneshot::channel();
|
|
let server = tokio::spawn(async move {
|
|
let (stream, _) = listener.accept().await.expect("local server must accept client");
|
|
let mut websocket = tokio_tungstenite::accept_async(stream).await.expect("local WebSocket handshake must succeed");
|
|
let first = read_request(&mut websocket).await;
|
|
let _ = first_seen_tx.send(());
|
|
let _ = release_rx.await;
|
|
send_result(&mut websocket, &first, serde_json::json!("first")).await;
|
|
});
|
|
let settings = session_settings(std::time::Duration::from_secs(1), std::time::Duration::from_millis(200), 1, 64 * 1024, 64 * 1024, 64 * 1024);
|
|
let session = crate::WsSession::connect(local_endpoint_with_session(url.as_str(), settings)).await.expect("client handshake must succeed");
|
|
let first_session = session.clone();
|
|
let first = tokio::spawn(async move {
|
|
return first_session.execute_json_rpc("firstMethod", std::vec::Vec::new()).await;
|
|
});
|
|
first_seen_rx.await.expect("server must observe first request");
|
|
wait_for_pending_count(&session, 1).await;
|
|
let second_error = session.execute_json_rpc("secondMethod", std::vec::Vec::new()).await.expect_err("second pending request must be rejected");
|
|
assert_eq!(second_error.code(), crate::ERROR_CODE_WS_BACKPRESSURE_OVERFLOW);
|
|
assert_eq!(session.state(), crate::WsSessionState::Active);
|
|
release_tx.send(()).expect("first request release signal must send");
|
|
assert_eq!(first.await.expect("first request task must join").expect("first request must complete"), serde_json::json!("first"));
|
|
wait_for_pending_count(&session, 0).await;
|
|
server.await.expect("local server task must complete");
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn websocket_oversized_outbound_request_is_rejected_before_socket_write() {
|
|
let (listener, url) = bind_local_listener().await;
|
|
let server = tokio::spawn(async move {
|
|
let (stream, _) = listener.accept().await.expect("local server must accept client");
|
|
let mut websocket = tokio_tungstenite::accept_async(stream).await.expect("local WebSocket handshake must succeed");
|
|
let request = read_request(&mut websocket).await;
|
|
assert_eq!(request.get("method").and_then(serde_json::Value::as_str), std::option::Option::Some("smallMethod"));
|
|
send_result(&mut websocket, &request, serde_json::json!(true)).await;
|
|
});
|
|
let settings = session_settings(std::time::Duration::from_secs(1), std::time::Duration::from_millis(200), 8, 256, 128, 1024);
|
|
let session = crate::WsSession::connect(local_endpoint_with_session(url.as_str(), settings)).await.expect("client handshake must succeed");
|
|
let oversized = serde_json::json!({"value":"X".repeat(512)});
|
|
let error = session.execute_json_rpc("oversizedMethod", std::vec![oversized]).await.expect_err("oversized outbound request must be rejected");
|
|
assert_eq!(error.code(), crate::ERROR_CODE_INVALID_RPC_PARAMETERS);
|
|
assert_eq!(session.state(), crate::WsSessionState::Active);
|
|
let result = session.execute_json_rpc("smallMethod", std::vec::Vec::new()).await.expect("small request must still succeed");
|
|
assert_eq!(result, serde_json::json!(true));
|
|
server.await.expect("local server task must complete");
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn websocket_oversized_inbound_frame_fails_before_json_decode() {
|
|
let (listener, url) = bind_local_listener().await;
|
|
let server = tokio::spawn(async move {
|
|
let (stream, _) = listener.accept().await.expect("local server must accept client");
|
|
let mut websocket = tokio_tungstenite::accept_async(stream).await.expect("local WebSocket handshake must succeed");
|
|
websocket.send(tokio_tungstenite::tungstenite::Message::Text("X".repeat(512).into())).await.expect("oversized fixture message must send");
|
|
tokio::time::sleep(std::time::Duration::from_millis(200)).await;
|
|
});
|
|
let settings = session_settings(std::time::Duration::from_secs(1), std::time::Duration::from_millis(200), 8, 128, 64, 1024);
|
|
let session = crate::WsSession::connect(local_endpoint_with_session(url.as_str(), settings)).await.expect("client handshake must succeed");
|
|
wait_for_state(&session, crate::WsSessionState::Failed).await;
|
|
server.await.expect("local server task must complete");
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn websocket_pending_request_timeout_purges_capacity_without_failing_session() {
|
|
let (listener, url) = bind_local_listener().await;
|
|
let server = tokio::spawn(async move {
|
|
let (stream, _) = listener.accept().await.expect("local server must accept client");
|
|
let mut websocket = tokio_tungstenite::accept_async(stream).await.expect("local WebSocket handshake must succeed");
|
|
let _ = read_request(&mut websocket).await;
|
|
tokio::time::sleep(std::time::Duration::from_millis(400)).await;
|
|
});
|
|
let settings = session_settings(std::time::Duration::from_millis(100), std::time::Duration::from_millis(100), 1, 64 * 1024, 64 * 1024, 64 * 1024);
|
|
let session = crate::WsSession::connect(local_endpoint_with_session(url.as_str(), settings)).await.expect("client handshake must succeed");
|
|
let wait = tokio::time::timeout(std::time::Duration::from_secs(1), session.execute_json_rpc("timeoutMethod", std::vec::Vec::new())).await;
|
|
let error = wait.expect("request timeout fixture must remain bounded").expect_err("remote silence must time out the request");
|
|
assert_eq!(error.code(), crate::ERROR_CODE_TIMEOUT);
|
|
wait_for_pending_count(&session, 0).await;
|
|
assert_eq!(session.state(), crate::WsSessionState::Active);
|
|
server.await.expect("local server task must complete");
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn websocket_ping_flushes_automatic_pong_and_keeps_session_active() {
|
|
let (listener, url) = bind_local_listener().await;
|
|
let (pong_tx, pong_rx) = tokio::sync::oneshot::channel();
|
|
let server = tokio::spawn(async move {
|
|
let (stream, _) = listener.accept().await.expect("local server must accept client");
|
|
let mut websocket = tokio_tungstenite::accept_async(stream).await.expect("local WebSocket handshake must succeed");
|
|
websocket.send(tokio_tungstenite::tungstenite::Message::Ping(b"ping-canary".to_vec().into())).await.expect("fixture ping must send");
|
|
let pong = websocket.next().await.expect("pong frame must arrive").expect("pong frame must decode");
|
|
match pong {
|
|
tokio_tungstenite::tungstenite::Message::Pong(payload) => assert_eq!(payload.as_ref(), b"ping-canary"),
|
|
other => panic!("expected Pong frame, got {other:?}"),
|
|
}
|
|
let _ = pong_tx.send(());
|
|
let request = read_request(&mut websocket).await;
|
|
send_result(&mut websocket, &request, serde_json::json!(true)).await;
|
|
});
|
|
let session = crate::WsSession::connect(local_endpoint(url.as_str())).await.expect("client handshake must succeed");
|
|
tokio::time::timeout(std::time::Duration::from_secs(1), pong_rx)
|
|
.await
|
|
.expect("automatic Pong must remain bounded")
|
|
.expect("fixture server must observe Pong");
|
|
assert_eq!(session.state(), crate::WsSessionState::Active);
|
|
let result = session.execute_json_rpc("afterPing", std::vec::Vec::new()).await.expect("session must remain usable after Ping/Pong");
|
|
assert_eq!(result, serde_json::json!(true));
|
|
server.await.expect("local server task must complete");
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn websocket_remote_close_transitions_to_closed_instead_of_failed() {
|
|
let (listener, url) = bind_local_listener().await;
|
|
let server = tokio::spawn(async move {
|
|
let (stream, _) = listener.accept().await.expect("local server must accept client");
|
|
let mut websocket = tokio_tungstenite::accept_async(stream).await.expect("local WebSocket handshake must succeed");
|
|
websocket.send(tokio_tungstenite::tungstenite::Message::Close(std::option::Option::None)).await.expect("fixture Close must send");
|
|
let _ = tokio::time::timeout(std::time::Duration::from_millis(200), websocket.next()).await;
|
|
});
|
|
let session = crate::WsSession::connect(local_endpoint(url.as_str())).await.expect("client handshake must succeed");
|
|
wait_for_state(&session, crate::WsSessionState::Closed).await;
|
|
assert!(session.close().await.is_ok());
|
|
server.await.expect("local server task must complete");
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn websocket_explicit_close_cancels_pending_and_finishes_with_hostile_peer() {
|
|
let (listener, url) = bind_local_listener().await;
|
|
let (request_seen_tx, request_seen_rx) = tokio::sync::oneshot::channel();
|
|
let server = tokio::spawn(async move {
|
|
let (stream, _) = listener.accept().await.expect("local server must accept client");
|
|
let mut websocket = tokio_tungstenite::accept_async(stream).await.expect("local WebSocket handshake must succeed");
|
|
let _ = read_request(&mut websocket).await;
|
|
let _ = request_seen_tx.send(());
|
|
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
|
|
});
|
|
let settings = session_settings(std::time::Duration::from_secs(1), std::time::Duration::from_millis(150), 8, 64 * 1024, 64 * 1024, 64 * 1024);
|
|
let session = crate::WsSession::connect(local_endpoint_with_session(url.as_str(), settings)).await.expect("client handshake must succeed");
|
|
let request_session = session.clone();
|
|
let pending = tokio::spawn(async move {
|
|
return request_session.execute_json_rpc("neverRespond", std::vec::Vec::new()).await;
|
|
});
|
|
request_seen_rx.await.expect("hostile server must observe request");
|
|
wait_for_pending_count(&session, 1).await;
|
|
let close_result = tokio::time::timeout(std::time::Duration::from_millis(300), session.close()).await;
|
|
close_result.expect("explicit close must remain bounded").expect("explicit close must finish cleanly");
|
|
assert_eq!(session.state(), crate::WsSessionState::Closed);
|
|
assert_eq!(session.snapshot().pending_request_count(), 0);
|
|
let pending_error = pending.await.expect("pending request task must join").expect_err("shutdown must cancel pending request");
|
|
assert_eq!(pending_error.code(), crate::ERROR_CODE_WS_SESSION_CLOSED);
|
|
server.await.expect("hostile server task must complete");
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn websocket_repeated_connect_close_cycles_are_bounded() {
|
|
const CYCLES: usize = 8;
|
|
let (listener, url) = bind_local_listener().await;
|
|
let server = tokio::spawn(async move {
|
|
for _ in 0..CYCLES {
|
|
let (stream, _) = listener.accept().await.expect("local server must accept client");
|
|
let mut websocket = tokio_tungstenite::accept_async(stream).await.expect("local WebSocket handshake must succeed");
|
|
let message = websocket.next().await.expect("client Close must arrive").expect("client Close must decode");
|
|
assert!(matches!(message, tokio_tungstenite::tungstenite::Message::Close(_)));
|
|
let _ = websocket.flush().await;
|
|
}
|
|
});
|
|
let settings = session_settings(std::time::Duration::from_secs(1), std::time::Duration::from_millis(200), 8, 64 * 1024, 64 * 1024, 64 * 1024);
|
|
for _ in 0..CYCLES {
|
|
let session = crate::WsSession::connect(local_endpoint_with_session(url.as_str(), settings.clone())).await.expect("client must connect");
|
|
session.close().await.expect("client close must succeed");
|
|
assert_eq!(session.state(), crate::WsSessionState::Closed);
|
|
}
|
|
tokio::time::timeout(std::time::Duration::from_secs(2), server)
|
|
.await
|
|
.expect("repeated close server must remain bounded")
|
|
.expect("repeated close server task must join");
|
|
}
|
|
|
|
async fn send_notification(
|
|
websocket: &mut tokio_tungstenite::WebSocketStream<tokio::net::TcpStream>,
|
|
method: &str,
|
|
remote_subscription_id: u64,
|
|
result: serde_json::Value,
|
|
) {
|
|
let notification = serde_json::json!({
|
|
"jsonrpc": "2.0",
|
|
"method": method,
|
|
"params": {
|
|
"result": result,
|
|
"subscription": remote_subscription_id
|
|
}
|
|
});
|
|
websocket.send(tokio_tungstenite::tungstenite::Message::Text(notification.to_string().into())).await.expect("local notification must send");
|
|
}
|
|
|
|
async fn wait_for_subscription_state<T>(subscription: &crate::WsSubscription<T>, expected: crate::WsSubscriptionState) {
|
|
let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(1);
|
|
loop {
|
|
if subscription.state() == expected {
|
|
return;
|
|
}
|
|
assert!(tokio::time::Instant::now() < deadline, "subscription did not reach expected state: {expected:?}");
|
|
tokio::time::sleep(std::time::Duration::from_millis(5)).await;
|
|
}
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn websocket_generic_subscription_registers_remote_id_dispatches_typed_notification_and_unsubscribes() {
|
|
let (listener, url) = bind_local_listener().await;
|
|
let server = tokio::spawn(async move {
|
|
let (stream, _) = listener.accept().await.expect("local server must accept client");
|
|
let mut websocket = tokio_tungstenite::accept_async(stream).await.expect("local WebSocket handshake must succeed");
|
|
let subscribe = read_request(&mut websocket).await;
|
|
assert_eq!(subscribe.get("method").and_then(serde_json::Value::as_str), std::option::Option::Some("slotSubscribe"));
|
|
send_result(&mut websocket, &subscribe, serde_json::json!(41)).await;
|
|
send_notification(&mut websocket, "slotNotification", 41, serde_json::json!({"slot": 9001})).await;
|
|
let unsubscribe = read_request(&mut websocket).await;
|
|
assert_eq!(unsubscribe.get("method").and_then(serde_json::Value::as_str), std::option::Option::Some("slotUnsubscribe"));
|
|
assert_eq!(unsubscribe.get("params"), std::option::Option::Some(&serde_json::json!([41])));
|
|
send_result(&mut websocket, &unsubscribe, serde_json::json!(true)).await;
|
|
});
|
|
let session = crate::WsSession::connect(local_endpoint(url.as_str())).await.expect("client handshake must succeed");
|
|
let mut subscription = session
|
|
.subscribe_typed(crate::WsSubscriptionKind::Slot, std::vec::Vec::new(), |value| {
|
|
return std::result::Result::Ok(value);
|
|
})
|
|
.await
|
|
.expect("generic slot subscribe must succeed");
|
|
assert_eq!(subscription.id().get(), 1);
|
|
assert_eq!(subscription.kind(), crate::WsSubscriptionKind::Slot);
|
|
assert_eq!(subscription.state(), crate::WsSubscriptionState::Active);
|
|
let snapshot = session.snapshot();
|
|
assert_eq!(snapshot.subscription_count(), 1);
|
|
assert_eq!(snapshot.subscriptions()[0].id(), subscription.id());
|
|
assert!(snapshot.subscriptions()[0].remote_bound());
|
|
let notification = subscription.recv().await.expect("typed notification channel must remain open").expect("notification must decode");
|
|
assert_eq!(notification, serde_json::json!({"slot": 9001}));
|
|
assert!(subscription.unsubscribe().await.expect("unsubscribe must succeed"));
|
|
assert_eq!(subscription.state(), crate::WsSubscriptionState::Closed);
|
|
tokio::task::yield_now().await;
|
|
assert_eq!(session.snapshot().subscription_count(), 0);
|
|
server.await.expect("local server task must complete");
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn websocket_remote_subscription_mapping_routes_multiple_families_to_stable_local_ids() {
|
|
let (listener, url) = bind_local_listener().await;
|
|
let server = tokio::spawn(async move {
|
|
let (stream, _) = listener.accept().await.expect("local server must accept client");
|
|
let mut websocket = tokio_tungstenite::accept_async(stream).await.expect("local WebSocket handshake must succeed");
|
|
let first = read_request(&mut websocket).await;
|
|
send_result(&mut websocket, &first, serde_json::json!(77)).await;
|
|
let second = read_request(&mut websocket).await;
|
|
send_result(&mut websocket, &second, serde_json::json!(12)).await;
|
|
send_notification(&mut websocket, "logsNotification", 12, serde_json::json!({"family": "logs"})).await;
|
|
send_notification(&mut websocket, "accountNotification", 77, serde_json::json!({"family": "account"})).await;
|
|
});
|
|
let session = crate::WsSession::connect(local_endpoint(url.as_str())).await.expect("client handshake must succeed");
|
|
let mut account = session
|
|
.subscribe_typed(crate::WsSubscriptionKind::Account, std::vec![serde_json::json!("account")], |value| {
|
|
return std::result::Result::Ok(value);
|
|
})
|
|
.await
|
|
.expect("account subscribe must succeed");
|
|
let mut logs = session
|
|
.subscribe_typed(crate::WsSubscriptionKind::Logs, std::vec![serde_json::json!("all")], |value| {
|
|
return std::result::Result::Ok(value);
|
|
})
|
|
.await
|
|
.expect("logs subscribe must succeed");
|
|
assert_eq!(account.id().get(), 1);
|
|
assert_eq!(logs.id().get(), 2);
|
|
assert_eq!(logs.recv().await.expect("logs notification must exist").expect("logs notification must decode"), serde_json::json!({"family": "logs"}));
|
|
assert_eq!(
|
|
account.recv().await.expect("account notification must exist").expect("account notification must decode"),
|
|
serde_json::json!({"family": "account"})
|
|
);
|
|
server.await.expect("local server task must complete");
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn websocket_unknown_remote_subscription_notification_is_ignored_without_affecting_registered_subscription() {
|
|
let (listener, url) = bind_local_listener().await;
|
|
let server = tokio::spawn(async move {
|
|
let (stream, _) = listener.accept().await.expect("local server must accept client");
|
|
let mut websocket = tokio_tungstenite::accept_async(stream).await.expect("local WebSocket handshake must succeed");
|
|
let subscribe = read_request(&mut websocket).await;
|
|
send_result(&mut websocket, &subscribe, serde_json::json!(3)).await;
|
|
send_notification(&mut websocket, "rootNotification", 999, serde_json::json!(100)).await;
|
|
send_notification(&mut websocket, "rootNotification", 3, serde_json::json!(101)).await;
|
|
let request = read_request(&mut websocket).await;
|
|
send_result(&mut websocket, &request, serde_json::json!(true)).await;
|
|
});
|
|
let session = crate::WsSession::connect(local_endpoint(url.as_str())).await.expect("client handshake must succeed");
|
|
let mut subscription = session
|
|
.subscribe_typed(crate::WsSubscriptionKind::Root, std::vec::Vec::new(), |value| {
|
|
return std::result::Result::Ok(value);
|
|
})
|
|
.await
|
|
.expect("root subscribe must succeed");
|
|
let notification = subscription.recv().await.expect("valid notification must exist").expect("valid notification must decode");
|
|
assert_eq!(notification, serde_json::json!(101));
|
|
assert_eq!(subscription.state(), crate::WsSubscriptionState::Active);
|
|
assert_eq!(session.state(), crate::WsSessionState::Active);
|
|
let result = session.execute_json_rpc("afterUnknownRemote", std::vec::Vec::new()).await.expect("physical session must remain usable");
|
|
assert_eq!(result, serde_json::json!(true));
|
|
server.await.expect("local server task must complete");
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn websocket_notification_method_mismatch_fails_only_the_logical_subscription() {
|
|
let (listener, url) = bind_local_listener().await;
|
|
let server = tokio::spawn(async move {
|
|
let (stream, _) = listener.accept().await.expect("local server must accept client");
|
|
let mut websocket = tokio_tungstenite::accept_async(stream).await.expect("local WebSocket handshake must succeed");
|
|
let subscribe = read_request(&mut websocket).await;
|
|
send_result(&mut websocket, &subscribe, serde_json::json!(5)).await;
|
|
send_notification(&mut websocket, "rootNotification", 5, serde_json::json!(1)).await;
|
|
let request = read_request(&mut websocket).await;
|
|
send_result(&mut websocket, &request, serde_json::json!(true)).await;
|
|
});
|
|
let session = crate::WsSession::connect(local_endpoint(url.as_str())).await.expect("client handshake must succeed");
|
|
let mut subscription = session
|
|
.subscribe_typed(crate::WsSubscriptionKind::Slot, std::vec::Vec::new(), |value| {
|
|
return std::result::Result::Ok(value);
|
|
})
|
|
.await
|
|
.expect("slot subscribe must succeed");
|
|
wait_for_subscription_state(&subscription, crate::WsSubscriptionState::Failed).await;
|
|
assert!(subscription.recv().await.is_none());
|
|
assert_eq!(session.state(), crate::WsSessionState::Active);
|
|
let result = session.execute_json_rpc("afterMismatch", std::vec::Vec::new()).await.expect("physical session must remain usable");
|
|
assert_eq!(result, serde_json::json!(true));
|
|
server.await.expect("local server task must complete");
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn websocket_typed_notification_decode_failure_fails_only_one_subscription_and_surfaces_error() {
|
|
let (listener, url) = bind_local_listener().await;
|
|
let server = tokio::spawn(async move {
|
|
let (stream, _) = listener.accept().await.expect("local server must accept client");
|
|
let mut websocket = tokio_tungstenite::accept_async(stream).await.expect("local WebSocket handshake must succeed");
|
|
let subscribe = read_request(&mut websocket).await;
|
|
send_result(&mut websocket, &subscribe, serde_json::json!(8)).await;
|
|
send_notification(&mut websocket, "rootNotification", 8, serde_json::json!("not-a-slot")).await;
|
|
let request = read_request(&mut websocket).await;
|
|
send_result(&mut websocket, &request, serde_json::json!(true)).await;
|
|
});
|
|
let session = crate::WsSession::connect(local_endpoint(url.as_str())).await.expect("client handshake must succeed");
|
|
let mut subscription = session
|
|
.subscribe_typed(crate::WsSubscriptionKind::Root, std::vec::Vec::new(), |value| {
|
|
return match value.as_u64() {
|
|
std::option::Option::Some(slot) => std::result::Result::Ok(slot),
|
|
std::option::Option::None => {
|
|
std::result::Result::Err(ksp_core_lib::Error::new(crate::ERROR_CODE_INVALID_RESPONSE, "fixture root notification must be numeric"))
|
|
},
|
|
};
|
|
})
|
|
.await
|
|
.expect("root subscribe must succeed");
|
|
let error = subscription.recv().await.expect("decode error must be delivered").expect_err("fixture payload must fail typed decoder");
|
|
assert_eq!(error.code(), crate::ERROR_CODE_INVALID_RESPONSE);
|
|
wait_for_subscription_state(&subscription, crate::WsSubscriptionState::Failed).await;
|
|
assert_eq!(session.state(), crate::WsSessionState::Active);
|
|
let result = session.execute_json_rpc("afterDecodeFailure", std::vec::Vec::new()).await.expect("physical session must remain usable");
|
|
assert_eq!(result, serde_json::json!(true));
|
|
server.await.expect("local server task must complete");
|
|
}
|