From 192d40a860e1875ca0b883e925c4b54aa9621487 Mon Sep 17 00:00:00 2001 From: Bellman <54757707+Yeachan-Heo@users.noreply.github.com> Date: Tue, 13 Oct 2026 11:40:03 +0900 Subject: [PATCH] feat(sdk): complete WebSocket client transport (#3349) Co-authored-by: clawdbot --- rust/Cargo.lock | 33 +- rust/crates/sdk/Cargo.toml | 7 +- rust/crates/sdk/src/error.rs | 168 +- rust/crates/sdk/src/frame.rs | 291 ++-- rust/crates/sdk/src/lib.rs | 6 +- rust/crates/sdk/src/transport.rs | 1375 ++++++++++++++--- .../sdk/tests/operation_registry_gap.rs | 49 +- 7 files changed, 1541 insertions(+), 388 deletions(-) diff --git a/rust/Cargo.lock b/rust/Cargo.lock index 5abc09ae..39aa6e7d 100644 --- a/rust/Cargo.lock +++ b/rust/Cargo.lock @@ -1215,7 +1215,7 @@ dependencies = [ "tokio", "tokio-rustls", "tower-service", - "webpki-roots", + "webpki-roots 1.0.6", ] [[package]] @@ -2117,7 +2117,7 @@ dependencies = [ "wasm-bindgen-futures", "wasm-streams", "web-sys", - "webpki-roots", + "webpki-roots 1.0.6", ] [[package]] @@ -2339,11 +2339,10 @@ dependencies = [ "futures-util", "serde", "serde_json", - "thiserror 1.0.69", "tokio", - "tokio-test", "tokio-tungstenite", "tracing", + "url", "uuid", ] @@ -2794,17 +2793,6 @@ dependencies = [ "tokio", ] -[[package]] -name = "tokio-test" -version = "0.4.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "12bc5db7778a8e08ee255090861e0e7aac55e784300077e7ed293c7be36168b2" -dependencies = [ - "futures-core", - "tokio", - "tokio-stream", -] - [[package]] name = "tokio-tungstenite" version = "0.23.1" @@ -2813,8 +2801,12 @@ checksum = "c6989540ced10490aaf14e6bad2e3d33728a2813310a0c71d1574304c49631cd" dependencies = [ "futures-util", "log", + "rustls", + "rustls-pki-types", "tokio", + "tokio-rustls", "tungstenite", + "webpki-roots 0.26.11", ] [[package]] @@ -3041,6 +3033,8 @@ dependencies = [ "httparse", "log", "rand 0.8.6", + "rustls", + "rustls-pki-types", "sha1", "thiserror 1.0.69", "utf-8", @@ -3249,6 +3243,15 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "webpki-roots" +version = "0.26.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "521bc38abb08001b01866da9f51eb7c5d647a19260e00054a8c7fd5f9e57f7a9" +dependencies = [ + "webpki-roots 1.0.6", +] + [[package]] name = "webpki-roots" version = "1.0.6" diff --git a/rust/crates/sdk/Cargo.toml b/rust/crates/sdk/Cargo.toml index 0c289500..9290a864 100644 --- a/rust/crates/sdk/Cargo.toml +++ b/rust/crates/sdk/Cargo.toml @@ -6,13 +6,10 @@ publish = false [dependencies] tokio = { version = "1", features = ["full"] } -tokio-tungstenite = "0.23" +tokio-tungstenite = { version = "0.23", features = ["rustls-tls-webpki-roots"] } futures-util = "0.3" serde_json = "1" serde = { version = "1", features = ["derive"] } uuid = { version = "1", features = ["v4", "serde"] } -thiserror = "1" tracing = "0.1" - -[dev-dependencies] -tokio-test = "0.4" +url = "2" diff --git a/rust/crates/sdk/src/error.rs b/rust/crates/sdk/src/error.rs index ab90ab63..b60079f9 100644 --- a/rust/crates/sdk/src/error.rs +++ b/rust/crates/sdk/src/error.rs @@ -1,102 +1,172 @@ -//! SDK error types and result type. +//! Typed SDK wire protocol errors and request recovery records. use std::fmt; +use std::time::SystemTime; + +use uuid::Uuid; /// SDK wire protocol error codes. #[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum ErrorCode { - /// Invalid JSON or malformed frame +pub enum SdkErrorCode { InvalidInput, - /// Unknown broker operation UnknownOperation, - /// Payload too large (>4 MiB) + NotFound, PayloadTooLarge, - /// Authentication token mismatch Unauthorized, - /// Broker or session unavailable + EndpointCredentialForbidden, Unavailable, - /// WebSocket connection error ConnectionError, - /// Request timeout + ConnectionClosed, Timeout, - /// Frame parsing or validation error FrameError, - /// Request/response correlation failure CorrelationError, - /// Other unclassified error + ProtocolError, + ReconnectExhausted, + UncertainAfterSend, Internal, } -impl fmt::Display for ErrorCode { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { +impl SdkErrorCode { + /// Return the stable wire representation of this error code. + pub const fn as_str(self) -> &'static str { match self { - Self::InvalidInput => write!(f, "invalid_input"), - Self::UnknownOperation => write!(f, "unknown_operation"), - Self::PayloadTooLarge => write!(f, "payload_too_large"), - Self::Unauthorized => write!(f, "unauthorized"), - Self::Unavailable => write!(f, "unavailable"), - Self::ConnectionError => write!(f, "connection_error"), - Self::Timeout => write!(f, "timeout"), - Self::FrameError => write!(f, "frame_error"), - Self::CorrelationError => write!(f, "correlation_error"), - Self::Internal => write!(f, "internal_error"), + Self::InvalidInput => "invalid_input", + Self::UnknownOperation => "unknown_operation", + Self::NotFound => "not_found", + Self::PayloadTooLarge => "payload_too_large", + Self::Unauthorized => "unauthorized", + Self::EndpointCredentialForbidden => "endpoint_credential_forbidden", + Self::Unavailable => "unavailable", + Self::ConnectionError => "connection_error", + Self::ConnectionClosed => "connection_closed", + Self::Timeout => "timeout", + Self::FrameError => "frame_error", + Self::CorrelationError => "correlation_error", + Self::ProtocolError => "protocol_error", + Self::ReconnectExhausted => "reconnect_exhausted", + Self::UncertainAfterSend => "uncertain_after_send", + Self::Internal => "internal_error", + } + } + + /// Parse a server error code, mapping unrecognized values to `Internal`. + pub fn from_wire(value: &str) -> Self { + match value { + "invalid_input" => Self::InvalidInput, + "unknown_operation" => Self::UnknownOperation, + "not_found" => Self::NotFound, + "payload_too_large" => Self::PayloadTooLarge, + "unauthorized" => Self::Unauthorized, + "endpoint_credential_forbidden" => Self::EndpointCredentialForbidden, + "unavailable" => Self::Unavailable, + "connection_error" => Self::ConnectionError, + "connection_closed" => Self::ConnectionClosed, + "timeout" => Self::Timeout, + "frame_error" => Self::FrameError, + "correlation_error" => Self::CorrelationError, + "protocol_error" => Self::ProtocolError, + "reconnect_exhausted" => Self::ReconnectExhausted, + "uncertain_after_send" => Self::UncertainAfterSend, + "internal" | "internal_error" => Self::Internal, + _ => Self::Internal, } } } +impl fmt::Display for SdkErrorCode { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(self.as_str()) + } +} + +/// Identity retained when a request may have reached the broker but its reply was lost. +/// +/// Records deliberately contain no input payload so credentials and other request data +/// are not retained solely for diagnostics. Callers can use the UUID to reconcile the +/// request with an operation-specific lookup before deciding whether to retry. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct SdkSentRecord { + pub id: Uuid, + pub operation: Option, + pub sent_at: SystemTime, +} + /// SDK wire protocol error. -#[derive(Debug)] +#[derive(Debug, Clone, PartialEq, Eq)] pub struct SdkError { - pub code: ErrorCode, + pub code: SdkErrorCode, pub message: String, + /// Original server code when it is not represented by `SdkErrorCode`. + pub remote_code: Option, + /// Present only when delivery may have happened before the error was observed. + pub sent_record: Option, } impl SdkError { - pub fn new(code: ErrorCode, message: impl Into) -> Self { + pub fn new(code: SdkErrorCode, message: impl Into) -> Self { Self { code, message: message.into(), + remote_code: None, + sent_record: None, } } pub fn invalid_input(message: impl Into) -> Self { - Self::new(ErrorCode::InvalidInput, message) + Self::new(SdkErrorCode::InvalidInput, message) } pub fn unknown_operation(message: impl Into) -> Self { - Self::new(ErrorCode::UnknownOperation, message) + Self::new(SdkErrorCode::UnknownOperation, message) } pub fn payload_too_large(message: impl Into) -> Self { - Self::new(ErrorCode::PayloadTooLarge, message) + Self::new(SdkErrorCode::PayloadTooLarge, message) } pub fn unauthorized(message: impl Into) -> Self { - Self::new(ErrorCode::Unauthorized, message) + Self::new(SdkErrorCode::Unauthorized, message) } pub fn unavailable(message: impl Into) -> Self { - Self::new(ErrorCode::Unavailable, message) + Self::new(SdkErrorCode::Unavailable, message) } pub fn connection_error(message: impl Into) -> Self { - Self::new(ErrorCode::ConnectionError, message) + Self::new(SdkErrorCode::ConnectionError, message) } pub fn timeout(message: impl Into) -> Self { - Self::new(ErrorCode::Timeout, message) + Self::new(SdkErrorCode::Timeout, message) } pub fn frame_error(message: impl Into) -> Self { - Self::new(ErrorCode::FrameError, message) + Self::new(SdkErrorCode::FrameError, message) } pub fn correlation_error(message: impl Into) -> Self { - Self::new(ErrorCode::CorrelationError, message) + Self::new(SdkErrorCode::CorrelationError, message) } pub fn internal(message: impl Into) -> Self { - Self::new(ErrorCode::Internal, message) + Self::new(SdkErrorCode::Internal, message) + } + + pub fn uncertain_after_send(record: SdkSentRecord) -> Self { + let mut error = Self::new( + SdkErrorCode::UncertainAfterSend, + "SDK request outcome is uncertain after the frame was sent.", + ); + error.sent_record = Some(record); + error + } + + pub fn from_remote(code: &str, message: impl Into) -> Self { + let mut error = Self::new(SdkErrorCode::from_wire(code), message); + if error.code == SdkErrorCode::Internal && code != "internal" && code != "internal_error" { + error.remote_code = Some(code.to_string()); + } + error } } @@ -123,8 +193,26 @@ mod tests { #[test] fn error_code_display_works() { - assert_eq!(ErrorCode::InvalidInput.to_string(), "invalid_input"); - assert_eq!(ErrorCode::PayloadTooLarge.to_string(), "payload_too_large"); - assert_eq!(ErrorCode::Unauthorized.to_string(), "unauthorized"); + assert_eq!(SdkErrorCode::InvalidInput.to_string(), "invalid_input"); + assert_eq!( + SdkErrorCode::PayloadTooLarge.to_string(), + "payload_too_large" + ); + assert_eq!(SdkErrorCode::Unauthorized.to_string(), "unauthorized"); + assert_eq!( + SdkErrorCode::from_wire("endpoint_credential_forbidden"), + SdkErrorCode::EndpointCredentialForbidden + ); + assert_eq!( + SdkErrorCode::UncertainAfterSend.to_string(), + "uncertain_after_send" + ); + } + + #[test] + fn unknown_remote_code_is_preserved_with_typed_internal_code() { + let error = SdkError::from_remote("future_error", "failed"); + assert_eq!(error.code, SdkErrorCode::Internal); + assert_eq!(error.remote_code.as_deref(), Some("future_error")); } } diff --git a/rust/crates/sdk/src/frame.rs b/rust/crates/sdk/src/frame.rs index f05e7f0c..e63e893a 100644 --- a/rust/crates/sdk/src/frame.rs +++ b/rust/crates/sdk/src/frame.rs @@ -1,45 +1,40 @@ -//! Frame types and serialization for the SDK wire protocol. +//! Typed frame structures and JSON serialization for the SDK wire protocol. +use serde::{Deserialize, Serialize}; use serde_json::{Map, Value}; +use uuid::Uuid; use crate::error::{SdkError, SdkResult}; -/// Protocol version implemented. +/// Protocol version implemented by this SDK client. pub const PROTOCOL_VERSION: u32 = 3; /// Maximum UTF-8 byte length for a single broker JSON frame. pub const MAX_FRAME_BYTES: usize = 4 * 1024 * 1024; -/// Frame type identifier. +/// Known frame type identifiers in the SDK wire protocol. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum FrameType { - /// Hello frame sent by broker on connection + Hello, + ServerHello, BrokerHello, - /// Request frame from client to broker BrokerRequest, - /// Response frame from broker to client BrokerResponse, - /// Query request QueryRequest, - /// Query response QueryResponse, - /// Control request ControlRequest, - /// Control response ControlResponse, - /// Event replay request EventReplay, - /// Event replay result EventReplayResult, - /// Session activation SessionActivate, - /// Session activation result SessionActivateResult, } impl FrameType { - pub fn as_str(&self) -> &'static str { + pub const fn as_str(self) -> &'static str { match self { + Self::Hello => "hello", + Self::ServerHello => "server_hello", Self::BrokerHello => "broker_hello", Self::BrokerRequest => "broker_request", Self::BrokerResponse => "broker_response", @@ -54,8 +49,10 @@ impl FrameType { } } - pub fn parse(s: &str) -> SdkResult { - match s { + pub fn parse(value: &str) -> SdkResult { + match value { + "hello" => Ok(Self::Hello), + "server_hello" => Ok(Self::ServerHello), "broker_hello" => Ok(Self::BrokerHello), "broker_request" => Ok(Self::BrokerRequest), "broker_response" => Ok(Self::BrokerResponse), @@ -67,50 +64,81 @@ impl FrameType { "event_replay_result" => Ok(Self::EventReplayResult), "session_activate" => Ok(Self::SessionActivate), "session_activate_result" => Ok(Self::SessionActivateResult), - _ => Err(SdkError::frame_error(format!("unknown frame type: {}", s))), + _ => Err(SdkError::frame_error(format!( + "unknown frame type: {value}" + ))), } } } -/// A frame in the SDK wire protocol. +/// A typed JSON-object frame in the SDK wire protocol. /// -/// Frames are JSON objects with a `type` field and optional `id` field for correlation. -#[derive(Debug, Clone)] -pub struct Frame { - /// Underlying JSON value +/// The map remains open-ended to allow operation-specific fields while the frame type, +/// correlation UUID, and handshake are validated by the client. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(transparent)] +pub struct SdkFrame { + /// Underlying JSON object. pub data: Map, } -impl Frame { - /// Create a new frame from a JSON object. +impl SdkFrame { + /// Create a frame from a JSON object. pub fn new(data: Map) -> Self { Self { data } } - /// Create an empty frame. + /// Create an empty frame object. pub fn empty() -> Self { - Self { data: Map::new() } + Self::new(Map::new()) } - /// Get the frame type. + /// Parse and validate the frame's `type` field. pub fn frame_type(&self) -> SdkResult { - let frame_type_str = self + let frame_type = self .data .get("type") .and_then(Value::as_str) .ok_or_else(|| SdkError::frame_error("missing 'type' field"))?; - FrameType::parse(frame_type_str) + FrameType::parse(frame_type) } - /// Get the optional request/response ID for correlation. - pub fn id(&self) -> Option { + /// Get the optional UUID used for request/response correlation. + pub fn id(&self) -> SdkResult> { + match self.data.get("id") { + None => Ok(None), + Some(Value::String(id)) => Uuid::parse_str(id).map(Some).map_err(|error| { + SdkError::correlation_error(format!("invalid frame UUID: {error}")) + }), + Some(_) => Err(SdkError::correlation_error( + "frame 'id' field must be a UUID string", + )), + } + } + + /// Set the frame's correlation UUID. + pub fn set_id(&mut self, id: Uuid) { self.data - .get("id") - .and_then(Value::as_str) - .map(String::from) + .insert("id".to_string(), Value::String(id.to_string())); } - /// Get the `ok` boolean field from a response frame. + /// Read a response correlation UUID from either supported field name. + pub fn response_id(&self) -> SdkResult> { + if self.data.contains_key("id") { + return self.id(); + } + match self.data.get("requestId") { + None => Ok(None), + Some(Value::String(id)) => Uuid::parse_str(id).map(Some).map_err(|error| { + SdkError::correlation_error(format!("invalid response UUID: {error}")) + }), + Some(_) => Err(SdkError::correlation_error( + "frame 'requestId' field must be a UUID string", + )), + } + } + + /// Read the `ok` boolean from a response frame. pub fn ok(&self) -> SdkResult { self.data .get("ok") @@ -118,33 +146,64 @@ impl Frame { .ok_or_else(|| SdkError::frame_error("missing or invalid 'ok' field")) } - /// Get the error object from a response frame. + /// Read the remote error code and message from an error response. pub fn error(&self) -> Option> { - let error_obj = self.data.get("error")?.as_object()?; - let code = error_obj.get("code")?.as_str()?.to_string(); - let message = error_obj.get("message")?.as_str()?.to_string(); + let error = self.data.get("error")?.as_object()?; + let code = error.get("code")?.as_str()?.to_string(); + let message = error.get("message")?.as_str()?.to_string(); Some(Ok((code, message))) } - /// Get the result field from a response frame. + /// Get the response result field. pub fn result(&self) -> Option { self.data.get("result").cloned() } - /// Convert to JSON string. + /// Serialize this frame to compact JSON. pub fn to_json(&self) -> SdkResult { serde_json::to_string(&self.data) - .map_err(|e| SdkError::frame_error(format!("JSON serialization error: {}", e))) + .map_err(|error| SdkError::frame_error(format!("JSON serialization error: {error}"))) } - /// Parse from JSON string. - pub fn from_json(json_str: &str) -> SdkResult { - let data: Map = serde_json::from_str(json_str) - .map_err(|e| SdkError::frame_error(format!("JSON parsing error: {}", e)))?; - Ok(Self::new(data)) + /// Parse a JSON object as an SDK frame. + pub fn from_json(json: &str) -> SdkResult { + match serde_json::from_str::(json) + .map_err(|error| SdkError::frame_error(format!("JSON parsing error: {error}")))? + { + Value::Object(data) => Ok(Self::new(data)), + _ => Err(SdkError::frame_error("SDK frame must be a JSON object")), + } } - /// Create a broker hello frame. + /// Create the protocol-v3 client hello with the advertised capabilities. + pub fn client_hello(capabilities: &[String]) -> Self { + let mut data = Map::new(); + data.insert("type".to_string(), Value::String("hello".to_string())); + data.insert( + "protocolVersion".to_string(), + Value::Number(PROTOCOL_VERSION.into()), + ); + data.insert( + "capabilities".to_string(), + Value::Array(capabilities.iter().cloned().map(Value::String).collect()), + ); + Self::new(data) + } + + /// Return whether this frame completes a server hello handshake. + pub fn is_server_hello(&self) -> bool { + matches!( + self.data.get("type").and_then(Value::as_str), + Some("hello" | "server_hello" | "broker_hello") + ) + } + + /// Read an optional connection identifier from a server hello. + pub fn connection_id(&self) -> Option<&str> { + self.data.get("connectionId").and_then(Value::as_str) + } + + /// Create a broker hello frame for protocol tests and mock servers. pub fn broker_hello() -> Self { let mut data = Map::new(); data.insert( @@ -158,63 +217,63 @@ impl Frame { Self::new(data) } - /// Create a broker request frame. - pub fn broker_request(id: String, operation: String, input: Value) -> Self { + /// Create a broker operation request frame. + pub fn broker_request(id: Uuid, operation: String, input: Value) -> Self { let mut data = Map::new(); data.insert( "type".to_string(), Value::String("broker_request".to_string()), ); - data.insert("id".to_string(), Value::String(id)); + data.insert("id".to_string(), Value::String(id.to_string())); data.insert("operation".to_string(), Value::String(operation)); data.insert("input".to_string(), input); Self::new(data) } - /// Create a broker response frame. - pub fn broker_response_ok(id: String, result: Value) -> Self { + /// Create a successful broker response frame. + pub fn broker_response_ok(id: Uuid, result: Value) -> Self { let mut data = Map::new(); data.insert( "type".to_string(), Value::String("broker_response".to_string()), ); - data.insert("id".to_string(), Value::String(id)); + data.insert("id".to_string(), Value::String(id.to_string())); data.insert("ok".to_string(), Value::Bool(true)); data.insert("result".to_string(), result); Self::new(data) } - /// Create a broker error response frame. - pub fn broker_response_error(id: String, code: String, message: String) -> Self { - let mut error_data = Map::new(); - error_data.insert("code".to_string(), Value::String(code)); - error_data.insert("message".to_string(), Value::String(message)); + /// Create an error broker response frame. + pub fn broker_response_error(id: Uuid, code: String, message: String) -> Self { + let mut error = Map::new(); + error.insert("code".to_string(), Value::String(code)); + error.insert("message".to_string(), Value::String(message)); let mut data = Map::new(); data.insert( "type".to_string(), Value::String("broker_response".to_string()), ); - data.insert("id".to_string(), Value::String(id)); + data.insert("id".to_string(), Value::String(id.to_string())); data.insert("ok".to_string(), Value::Bool(false)); - data.insert("error".to_string(), Value::Object(error_data)); + data.insert("error".to_string(), Value::Object(error)); Self::new(data) } - /// Get the payload size in bytes. + /// Get the serialized frame length in UTF-8 bytes. pub fn byte_size(&self) -> usize { - self.to_json().map(|s| s.len()).unwrap_or(0) + self.to_json().map_or(0, |json| json.len()) } - /// Get the operation name from a broker request frame. + /// Read the operation name from a request frame. pub fn operation(&self) -> Option { self.data .get("operation") .and_then(Value::as_str) - .map(String::from) + .map(str::to_owned) } - /// Get the input value from a broker request frame. + /// Read the request input object. pub fn input(&self) -> Option { self.data.get("input").cloned() } @@ -226,7 +285,7 @@ mod tests { #[test] fn frame_type_round_trip() { - assert_eq!(FrameType::BrokerHello.as_str(), "broker_hello"); + assert_eq!(FrameType::Hello.as_str(), "hello"); assert_eq!( FrameType::parse("broker_hello").unwrap(), FrameType::BrokerHello @@ -235,68 +294,76 @@ mod tests { #[test] fn frame_parsing_extracts_type() { - let frame = Frame::broker_hello(); - assert_eq!(frame.frame_type().unwrap(), FrameType::BrokerHello); - } - - #[test] - fn frame_with_id_can_be_recovered() { - let frame = Frame::broker_request( - "test-id".to_string(), - "test.operation".to_string(), - Value::Object(Map::new()), + assert_eq!( + SdkFrame::broker_hello().frame_type().unwrap(), + FrameType::BrokerHello ); - assert_eq!(frame.id(), Some("test-id".to_string())); } #[test] - fn frame_json_round_trip() { - let frame = Frame::broker_hello(); - let json_str = frame.to_json().unwrap(); - let parsed = Frame::from_json(&json_str).unwrap(); - assert_eq!(parsed.frame_type().unwrap(), FrameType::BrokerHello); + fn frame_id_round_trips_as_a_uuid() { + let id = Uuid::new_v4(); + let frame = SdkFrame::broker_request(id, "test.operation".to_string(), Value::Null); + assert_eq!(frame.id().unwrap(), Some(id)); } #[test] - fn broker_response_error_contains_code_and_message() { - let frame = Frame::broker_response_error( - "id1".to_string(), + fn json_round_trip_preserves_the_typed_frame() { + let frame = SdkFrame::client_hello(&["capability.one".to_string()]); + let parsed = SdkFrame::from_json(&frame.to_json().unwrap()).unwrap(); + assert_eq!(parsed, frame); + assert_eq!(parsed.frame_type().unwrap(), FrameType::Hello); + } + + #[test] + fn client_hello_advertises_protocol_and_capabilities() { + let frame = SdkFrame::client_hello(&["capability.one".to_string()]); + assert_eq!(frame.data["protocolVersion"], PROTOCOL_VERSION); + assert_eq!( + frame.data["capabilities"], + serde_json::json!(["capability.one"]) + ); + } + + #[test] + fn broker_error_response_exposes_code_and_message() { + let frame = SdkFrame::broker_response_error( + Uuid::new_v4(), "invalid_input".to_string(), - "test error".to_string(), + "bad input".to_string(), ); - let (code, message) = frame.error().unwrap().unwrap(); - assert_eq!(code, "invalid_input"); - assert_eq!(message, "test error"); + assert_eq!( + frame.error().unwrap().unwrap(), + ("invalid_input".into(), "bad input".into()) + ); + assert!(!frame.ok().unwrap()); } #[test] - fn frame_extracts_operation_from_broker_request() { - let frame = Frame::broker_request( - "test-id".to_string(), - "query.list_sessions".to_string(), - Value::Object(Map::new()), - ); - assert_eq!(frame.operation(), Some("query.list_sessions".to_string())); - } - - #[test] - fn frame_extracts_input_from_broker_request() { - let input = Value::Object({ - let mut m = Map::new(); - m.insert("session_id".to_string(), Value::String("123".to_string())); - m - }); - let frame = Frame::broker_request( - "test-id".to_string(), + fn operation_and_input_are_recoverable() { + let input = serde_json::json!({"session_id": "session-1"}); + let frame = SdkFrame::broker_request( + Uuid::new_v4(), "query.get_session".to_string(), input.clone(), ); + assert_eq!(frame.operation().as_deref(), Some("query.get_session")); assert_eq!(frame.input(), Some(input)); } #[test] - fn frame_without_operation_returns_none() { - let frame = Frame::broker_hello(); - assert_eq!(frame.operation(), None); + fn malformed_json_and_non_object_values_are_rejected() { + assert!(SdkFrame::from_json("{").is_err()); + assert!(SdkFrame::from_json("[]").is_err()); + assert!(SdkFrame::from_json("null").is_err()); + } + + #[test] + fn invalid_correlation_uuid_is_rejected() { + let frame = SdkFrame::from_json(r#"{"type":"broker_response","id":"not-a-uuid"}"#).unwrap(); + assert_eq!( + frame.id().unwrap_err().code, + crate::error::SdkErrorCode::CorrelationError + ); } } diff --git a/rust/crates/sdk/src/lib.rs b/rust/crates/sdk/src/lib.rs index c8b237c7..4703a7cb 100644 --- a/rust/crates/sdk/src/lib.rs +++ b/rust/crates/sdk/src/lib.rs @@ -16,7 +16,7 @@ pub mod frame; pub mod registry; pub mod transport; -pub use error::{SdkError, SdkResult}; -pub use frame::{Frame, FrameType}; +pub use error::{SdkError, SdkErrorCode, SdkResult, SdkSentRecord}; +pub use frame::{FrameType, SdkFrame}; pub use registry::{OperationMetadata, OperationRegistry, Schema}; -pub use transport::{SdkTransport, SdkTransportConfig}; +pub use transport::{SdkClient, SdkClientConfig}; diff --git a/rust/crates/sdk/src/transport.rs b/rust/crates/sdk/src/transport.rs index 97689a44..ece95cba 100644 --- a/rust/crates/sdk/src/transport.rs +++ b/rust/crates/sdk/src/transport.rs @@ -1,251 +1,1254 @@ -//! WebSocket transport implementation for the SDK wire protocol. +//! Reconnecting WebSocket client for the SDK wire protocol. -use std::collections::HashMap; -use std::sync::Arc; +use std::collections::{HashMap, VecDeque}; +use std::fmt; +use std::future::pending; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, SystemTime}; -use futures_util::sink::SinkExt; -use futures_util::stream::StreamExt; +use futures_util::{SinkExt, StreamExt}; use serde_json::Value; -use tokio::sync::{mpsc, oneshot, Mutex, RwLock}; -use tokio_tungstenite::{connect_async, tungstenite::Message}; +use tokio::net::TcpStream; +use tokio::sync::{mpsc, oneshot, watch, Mutex as AsyncMutex, RwLock}; +use tokio::task::JoinHandle; +use tokio::time::{sleep, timeout, timeout_at, Instant}; +use tokio_tungstenite::tungstenite::Message; +use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream}; +use url::Url; use uuid::Uuid; -use crate::error::{SdkError, SdkResult}; -use crate::frame::{Frame, FrameType, MAX_FRAME_BYTES}; +use crate::error::{SdkError, SdkErrorCode, SdkResult, SdkSentRecord}; +use crate::frame::{SdkFrame, MAX_FRAME_BYTES, PROTOCOL_VERSION}; -/// Configuration for SDK transport. +const MAX_SENT_RECORDS: usize = 256; +const CLOSE_GRACE: Duration = Duration::from_secs(5); +const CANCEL_ACK_GRACE: Duration = Duration::from_millis(100); + +type ClientSocket = WebSocketStream>; + +/// Runtime options for an SDK WebSocket client. #[derive(Debug, Clone)] -pub struct SdkTransportConfig { - /// Server URL (e.g., "ws://127.0.0.1:8080") +pub struct SdkClientConfig { + /// WebSocket endpoint (`ws://` or `wss://`). pub url: String, - /// Authentication token + /// Broker authentication token, sent as the `token` query parameter. pub token: String, - /// Maximum payload size in bytes (default: 4 MiB) - pub max_payload_bytes: usize, - /// Request timeout in seconds (default: 30) - pub timeout_secs: u64, + /// Maximum UTF-8 frame size accepted in either direction. + pub max_frame_bytes: usize, + /// Default timeout for an individual request. + pub request_timeout: Duration, + /// Timeout for each socket-open and hello-handshake attempt. + pub connect_timeout: Duration, + /// Maximum time spent across a connection or reconnect cycle. + pub reconnect_deadline: Duration, + /// Number of retries after the initial connection attempt. + pub reconnect_attempts: u32, + /// Initial reconnect delay; subsequent delays double up to `reconnect_max_backoff`. + pub reconnect_initial_backoff: Duration, + /// Maximum reconnect delay. + pub reconnect_max_backoff: Duration, + /// Capabilities sent in the protocol-v3 client hello. + pub capabilities: Vec, } -impl Default for SdkTransportConfig { +impl Default for SdkClientConfig { fn default() -> Self { Self { url: "ws://127.0.0.1:8080".to_string(), token: String::new(), - max_payload_bytes: MAX_FRAME_BYTES, - timeout_secs: 30, + max_frame_bytes: MAX_FRAME_BYTES, + request_timeout: Duration::from_secs(10), + connect_timeout: Duration::from_secs(10), + reconnect_deadline: Duration::from_secs(30), + reconnect_attempts: 3, + reconnect_initial_backoff: Duration::from_millis(25), + reconnect_max_backoff: Duration::from_secs(2), + capabilities: Vec::new(), } } } -/// Pending request tracking for correlation. -struct PendingRequest { - response_tx: oneshot::Sender>, +impl SdkClientConfig { + fn validate(&self) -> SdkResult<()> { + let endpoint = Url::parse(&self.url) + .map_err(|error| SdkError::invalid_input(format!("invalid WebSocket URL: {error}")))?; + if !matches!(endpoint.scheme(), "ws" | "wss") { + return Err(SdkError::invalid_input( + "SDK endpoint must use the ws:// or wss:// scheme", + )); + } + if self.max_frame_bytes == 0 { + return Err(SdkError::invalid_input( + "maximum frame size must be greater than zero", + )); + } + if self.connect_timeout.is_zero() || self.reconnect_deadline.is_zero() { + return Err(SdkError::invalid_input( + "connection timeout and reconnect deadline must be greater than zero", + )); + } + if self.reconnect_max_backoff < self.reconnect_initial_backoff { + return Err(SdkError::invalid_input( + "maximum reconnect backoff must be at least the initial backoff", + )); + } + Ok(()) + } } -/// SDK WebSocket transport. -pub struct SdkTransport { - /// Channel for sending messages to the writer task - tx: mpsc::UnboundedSender, - /// Pending requests by ID - pending: Arc>>, - /// Configuration - config: SdkTransportConfig, - /// Background task handle - task_handle: Arc>>>, +struct SentRecordStore { + records: HashMap, + order: VecDeque, } -impl SdkTransport { - /// Create a new SDK transport and connect to the broker. - pub async fn connect(config: SdkTransportConfig) -> SdkResult { - // Add token to query params - let url_with_token = if config.url.contains('?') { - format!("{}&token={}", config.url, config.token) - } else { - format!("{}?token={}", config.url, config.token) - }; - - // Create WebSocket connection - let (ws, _) = connect_async(&url_with_token).await.map_err(|e| { - SdkError::connection_error(format!("WebSocket connection failed: {}", e)) - })?; - - let (ws_tx, ws_rx) = tokio::sync::mpsc::unbounded_channel(); - let pending = Arc::new(RwLock::new(HashMap::new())); - let pending_clone = pending.clone(); - - // Spawn background task for handling the WebSocket - let task_handle = tokio::spawn(async move { - Self::background_task(ws, ws_rx, pending_clone).await; - }); - - let transport = Self { - tx: ws_tx, - pending, - config, - task_handle: Arc::new(Mutex::new(Some(task_handle))), - }; - - Ok(transport) +impl SentRecordStore { + fn new() -> Self { + Self { + records: HashMap::new(), + order: VecDeque::new(), + } } - /// Send a request and wait for response. - pub async fn request(&self, operation: String, input: Value) -> SdkResult { - // Generate correlation ID - let id = Uuid::new_v4().to_string(); + fn insert(&mut self, record: SdkSentRecord) { + if self.records.contains_key(&record.id) { + self.order.retain(|id| *id != record.id); + } + self.order.push_back(record.id); + self.records.insert(record.id, record); + while self.records.len() > MAX_SENT_RECORDS { + if let Some(oldest) = self.order.pop_front() { + self.records.remove(&oldest); + } + } + } - // Create request frame - let request = Frame::broker_request(id.clone(), operation, input); + fn remove(&mut self, id: Uuid) { + self.records.remove(&id); + self.order.retain(|record_id| *record_id != id); + } - // Check payload size - if request.byte_size() > self.config.max_payload_bytes { + fn get(&self, id: Uuid) -> Option { + self.records.get(&id).cloned() + } + + fn all(&self) -> Vec { + self.order + .iter() + .filter_map(|id| self.records.get(id).cloned()) + .collect() + } +} + +struct RequestMessage { + id: Uuid, + operation: Option, + serialized: String, + deadline: Instant, + response_tx: oneshot::Sender>, +} + +enum ClientCommand { + Request(RequestMessage), + Cancel { + id: Uuid, + ack: oneshot::Sender>, + }, +} + +struct PendingRequest { + response_tx: oneshot::Sender>, + sent_record: SdkSentRecord, +} + +struct ConnectedSocket { + socket: ClientSocket, + connection_id: Option, +} + +/// SDK client with UUID correlation, handshake, timeout handling, and reconnects. +pub struct SdkClient { + config: SdkClientConfig, + command_tx: mpsc::UnboundedSender, + shutdown_tx: watch::Sender, + task_handle: AsyncMutex>>, + sent_records: Arc>, + connection_id: Arc>>, + closed: AtomicBool, +} + +impl fmt::Debug for SdkClient { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("SdkClient") + .field("config", &self.config) + .field("closed", &self.closed.load(Ordering::Acquire)) + .finish_non_exhaustive() + } +} + +impl SdkClient { + /// Connect, complete the protocol-v3 hello handshake, and start reconnect supervision. + pub async fn connect(config: SdkClientConfig) -> SdkResult { + config.validate()?; + let connected = connect_with_retry(&config).await?; + let connection_id = Arc::new(RwLock::new(connected.connection_id.clone())); + let sent_records = Arc::new(Mutex::new(SentRecordStore::new())); + let (command_tx, command_rx) = mpsc::unbounded_channel(); + let (shutdown_tx, shutdown_rx) = watch::channel(false); + let task_handle = tokio::spawn(run_client( + connected, + config.clone(), + command_rx, + shutdown_rx, + sent_records.clone(), + connection_id.clone(), + )); + + Ok(Self { + config, + command_tx, + shutdown_tx, + task_handle: AsyncMutex::new(Some(task_handle)), + sent_records, + connection_id, + closed: AtomicBool::new(false), + }) + } + + /// Send a broker operation with the configured default request timeout. + pub async fn request(&self, operation: impl Into, input: Value) -> SdkResult { + self.request_with_timeout(operation, input, self.config.request_timeout) + .await + } + + /// Send a broker operation with a per-request timeout override. + pub async fn request_with_timeout( + &self, + operation: impl Into, + input: Value, + request_timeout: Duration, + ) -> SdkResult { + let operation = operation.into(); + let frame = SdkFrame::broker_request(Uuid::new_v4(), operation, input); + self.request_frame_with_timeout(frame, request_timeout) + .await + } + + /// Send a typed frame, replacing its `id` with a fresh UUID for correlation. + pub async fn request_frame(&self, frame: SdkFrame) -> SdkResult { + self.request_frame_with_timeout(frame, self.config.request_timeout) + .await + } + + /// Send a typed frame with a per-request timeout override. + pub async fn request_frame_with_timeout( + &self, + mut frame: SdkFrame, + request_timeout: Duration, + ) -> SdkResult { + if self.closed.load(Ordering::Acquire) { + return Err(SdkError::new( + SdkErrorCode::ConnectionClosed, + "SDK client is closed", + )); + } + let id = Uuid::new_v4(); + frame.set_id(id); + let operation = frame.operation(); + let serialized = frame.to_json()?; + if serialized.len() > self.config.max_frame_bytes { return Err(SdkError::payload_too_large(format!( "frame size {} exceeds maximum {}", - request.byte_size(), - self.config.max_payload_bytes + serialized.len(), + self.config.max_frame_bytes ))); } - // Create response channel + let deadline = Instant::now() + request_timeout; let (response_tx, response_rx) = oneshot::channel(); - self.pending - .write() - .await - .insert(id.clone(), PendingRequest { response_tx }); + self.command_tx + .send(ClientCommand::Request(RequestMessage { + id, + operation, + serialized, + deadline, + response_tx, + })) + .map_err(|_| { + SdkError::new( + SdkErrorCode::ConnectionClosed, + "SDK connection task is no longer running", + ) + })?; - // Send request - let json = request.to_json()?; - self.tx - .send(Message::Text(json)) - .map_err(|e| SdkError::connection_error(format!("send failed: {}", e)))?; - - // Wait for response with timeout - let timeout_duration = std::time::Duration::from_secs(self.config.timeout_secs); - match tokio::time::timeout(timeout_duration, response_rx).await { - Ok(Ok(frame)) => frame, - Ok(Err(_)) => Err(SdkError::correlation_error("response channel closed")), - Err(_) => Err(SdkError::timeout("response not received within timeout")), + match timeout_at(deadline, response_rx).await { + Ok(Ok(response)) => response, + Ok(Err(_)) => Err(SdkError::connection_error( + "SDK connection task dropped the response channel", + )), + Err(_) => { + let (ack_tx, ack_rx) = oneshot::channel(); + let _ = self + .command_tx + .send(ClientCommand::Cancel { id, ack: ack_tx }); + let acknowledged = timeout(CANCEL_ACK_GRACE, ack_rx) + .await + .ok() + .and_then(Result::ok) + .flatten(); + let sent_record = acknowledged.or_else(|| self.get_sent_record(id)); + match sent_record { + Some(record) => Err(SdkError::uncertain_after_send(record)), + None => Err(SdkError::timeout(format!( + "SDK request timed out after {}ms before a response was received", + request_timeout.as_millis() + ))), + } + } } } - /// Background task for handling WebSocket communication. - #[allow(unused_mut)] - async fn background_task( - ws: tokio_tungstenite::WebSocketStream< - tokio_tungstenite::MaybeTlsStream, - >, - mut ws_rx: mpsc::UnboundedReceiver, - pending: Arc>>, - ) { - let (mut ws_sink, mut ws_stream) = ws.split(); + /// Current broker connection identifier from the most recent hello frame. + pub async fn connection_id(&self) -> Option { + self.connection_id.read().await.clone() + } - loop { - tokio::select! { - // Handle incoming WebSocket messages - msg = ws_stream.next() => { - match msg { - Some(Ok(Message::Text(text))) => { - // Parse frame - match Frame::from_json(&text) { - Ok(frame) => { - // Handle frame type - match frame.frame_type() { - Ok(FrameType::BrokerHello) => { - // Hello frame received, continue - } - Ok(FrameType::BrokerResponse) => { - // Response frame, correlate with request - if let Some(id) = frame.id() { - if let Some(PendingRequest { response_tx }) = - pending.write().await.remove(&id) - { - let _ = response_tx.send(Ok(frame)); - } - } - } - _ => { - // Other frame types - tracing::debug!("Unhandled frame type: {:?}", frame.frame_type()); - } - } - } - Err(e) => { - tracing::warn!("Failed to parse frame: {}", e); - } - } - } - Some(Ok(Message::Close(_))) => { - break; - } - Some(Ok(_)) => { - // Other message types (binary, etc.), ignore - } - Some(Err(e)) => { - tracing::warn!("WebSocket error: {}", e); - break; - } - None => { - break; - } - } + /// Look up a retained request identity after an uncertain send. + pub fn get_sent_record(&self, id: Uuid) -> Option { + self.sent_records + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .get(id) + } + + /// List outstanding sent records that require caller-side reconciliation. + pub fn uncertain_sends(&self) -> Vec { + self.sent_records + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .all() + } + + /// Forget a request record after the caller has reconciled its outcome. + pub fn forget_sent_record(&self, id: Uuid) -> bool { + let mut records = self + .sent_records + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let found = records.records.contains_key(&id); + records.remove(id); + found + } + + /// Close the socket and stop reconnect attempts. Safe to call more than once. + pub async fn close(&self) -> SdkResult<()> { + if !self.closed.swap(true, Ordering::AcqRel) { + let _ = self.shutdown_tx.send(true); + } + if let Some(mut task) = self.task_handle.lock().await.take() { + match timeout(CLOSE_GRACE, &mut task).await { + Ok(Ok(())) => {} + Ok(Err(error)) => { + return Err(SdkError::connection_error(format!( + "SDK connection task failed while closing: {error}" + ))) } - // Handle outgoing messages - Some(msg) = ws_rx.recv() => { - if let Err(e) = ws_sink.send(msg).await { - tracing::warn!("Failed to send message: {}", e); - break; - } + Err(_) => { + task.abort(); + return Err(SdkError::timeout( + "SDK connection task did not stop in time", + )); } - else => break, + } + } + Ok(()) + } +} + +impl Drop for SdkClient { + fn drop(&mut self) { + if !self.closed.swap(true, Ordering::AcqRel) { + let _ = self.shutdown_tx.send(true); + } + } +} + +fn authenticated_url(config: &SdkClientConfig) -> SdkResult { + let mut url = Url::parse(&config.url) + .map_err(|error| SdkError::invalid_input(format!("invalid WebSocket URL: {error}")))?; + if !matches!(url.scheme(), "ws" | "wss") { + return Err(SdkError::invalid_input( + "SDK endpoint must use the ws:// or wss:// scheme", + )); + } + let mut query: Vec<(String, String)> = url + .query_pairs() + .filter(|(key, _)| key != "token") + .map(|(key, value)| (key.into_owned(), value.into_owned())) + .collect(); + query.push(("token".to_string(), config.token.clone())); + url.set_query(None); + url.query_pairs_mut().extend_pairs(query); + Ok(url) +} + +async fn connect_with_retry(config: &SdkClientConfig) -> SdkResult { + let deadline = Instant::now() + config.reconnect_deadline; + let mut last_error = None; + + for attempt in 0..=config.reconnect_attempts { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + return Err(SdkError::timeout("SDK reconnect deadline elapsed")); + } + let attempt_timeout = remaining.min(config.connect_timeout); + match timeout(attempt_timeout, connect_once(config)).await { + Ok(Ok(connected)) => return Ok(connected), + Ok(Err(error)) => last_error = Some(error), + Err(_) => { + last_error = Some(SdkError::timeout(format!( + "SDK connection attempt timed out after {}ms", + attempt_timeout.as_millis() + ))) } } - // Fail all pending requests - let mut pending = pending.write().await; - for (_, PendingRequest { response_tx }) in pending.drain() { - let _ = response_tx.send(Err(SdkError::connection_error("connection closed"))); + if attempt == config.reconnect_attempts { + break; + } + let delay = reconnect_delay(config, attempt); + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + return Err(SdkError::timeout("SDK reconnect deadline elapsed")); + } + sleep(delay.min(remaining)).await; + } + + let cause = last_error + .map(|error| error.to_string()) + .unwrap_or_else(|| "no connection attempt was made".to_string()); + Err(SdkError::new( + SdkErrorCode::ReconnectExhausted, + format!("SDK WebSocket reconnect attempts exhausted: {cause}"), + )) +} + +fn reconnect_delay(config: &SdkClientConfig, attempt: u32) -> Duration { + let multiplier = 1_u32.checked_shl(attempt).unwrap_or(u32::MAX); + config + .reconnect_initial_backoff + .saturating_mul(multiplier) + .min(config.reconnect_max_backoff) +} + +async fn connect_once(config: &SdkClientConfig) -> SdkResult { + let url = authenticated_url(config)?; + let (mut socket, _) = connect_async(url.as_str()).await.map_err(|error| { + SdkError::connection_error(format!("WebSocket connection failed: {error}")) + })?; + + let hello = SdkFrame::client_hello(&config.capabilities).to_json()?; + socket.send(Message::Text(hello)).await.map_err(|error| { + SdkError::connection_error(format!("client hello send failed: {error}")) + })?; + + let handshake_deadline = Instant::now() + config.connect_timeout; + loop { + let message = timeout_at(handshake_deadline, socket.next()) + .await + .map_err(|_| SdkError::timeout("SDK server hello handshake timed out"))? + .ok_or_else(|| SdkError::connection_error("WebSocket closed before server hello"))? + .map_err(|error| { + SdkError::connection_error(format!("WebSocket failed before server hello: {error}")) + })?; + match message { + Message::Text(text) => { + if text.len() > config.max_frame_bytes { + return Err(SdkError::payload_too_large(format!( + "server hello frame size {} exceeds maximum {}", + text.len(), + config.max_frame_bytes + ))); + } + let frame = SdkFrame::from_json(&text)?; + if frame.is_server_hello() { + if let Some(version) = frame.data.get("protocolVersion").and_then(Value::as_u64) + { + if version != u64::from(PROTOCOL_VERSION) { + return Err(SdkError::new( + SdkErrorCode::ProtocolError, + format!( + "server protocol version {version} does not match {PROTOCOL_VERSION}" + ), + )); + } + } + let connection_id = frame + .connection_id() + .filter(|value| !value.is_empty()) + .map(str::to_owned); + return Ok(ConnectedSocket { + socket, + connection_id, + }); + } + } + Message::Ping(payload) => { + socket.send(Message::Pong(payload)).await.map_err(|error| { + SdkError::connection_error(format!("WebSocket pong failed: {error}")) + })? + } + Message::Close(_) => { + return Err(SdkError::connection_error( + "WebSocket closed before server hello", + )); + } + Message::Binary(_) | Message::Pong(_) | Message::Frame(_) => {} + } + } +} + +async fn await_reconnect( + reconnect_task: &mut Option>>, +) -> SdkResult { + match reconnect_task { + Some(task) => task.await.map_err(|error| { + SdkError::connection_error(format!("SDK reconnect task failed: {error}")) + })?, + None => pending().await, + } +} + +async fn run_client( + connected: ConnectedSocket, + config: SdkClientConfig, + mut command_rx: mpsc::UnboundedReceiver, + mut shutdown_rx: watch::Receiver, + sent_records: Arc>, + connection_id: Arc>>, +) { + let mut socket = Some(connected.socket); + let mut queued = VecDeque::new(); + let mut pending_requests = HashMap::new(); + let mut reconnect_task = None; + + loop { + let incoming = async { + match socket.as_mut() { + Some(connected) => connected.next().await, + None => pending().await, + } + }; + tokio::select! { + changed = shutdown_rx.changed() => { + if changed.is_err() || *shutdown_rx.borrow() { + break; + } + } + message = incoming => { + match message { + Some(Ok(Message::Text(text))) => { + if text.len() > config.max_frame_bytes { + tracing::warn!(size = text.len(), "SDK server frame exceeds configured limit"); + fail_pending_uncertain(&mut pending_requests); + start_reconnect(&config, &mut socket, &mut reconnect_task); + } else { + handle_server_text(&text, &mut pending_requests, &sent_records, &connection_id).await; + } + } + Some(Ok(Message::Ping(payload))) => { + let send_result = match socket.as_mut() { + Some(connected) => connected.send(Message::Pong(payload)).await, + None => continue, + }; + if send_result.is_err() { + fail_pending_uncertain(&mut pending_requests); + start_reconnect(&config, &mut socket, &mut reconnect_task); + } + } + Some(Ok(Message::Close(_))) | Some(Err(_)) | None => { + fail_pending_uncertain(&mut pending_requests); + start_reconnect(&config, &mut socket, &mut reconnect_task); + } + Some(Ok(Message::Binary(_))) | Some(Ok(Message::Pong(_))) | Some(Ok(Message::Frame(_))) => {} + } + } + command = command_rx.recv() => { + let Some(command) = command else { break }; + match command { + ClientCommand::Request(request) => { + if socket.is_none() { + queued.push_back(request); + } else if !send_request( + request, + socket.as_mut().expect("socket checked above"), + &mut pending_requests, + &sent_records, + ).await { + fail_pending_uncertain(&mut pending_requests); + start_reconnect(&config, &mut socket, &mut reconnect_task); + } + } + ClientCommand::Cancel { id, ack } => { + let record = cancel_request( + id, + &mut queued, + &mut pending_requests, + &sent_records, + ); + let _ = ack.send(record); + } + } + } + result = await_reconnect(&mut reconnect_task) => { + reconnect_task = None; + match result { + Ok(connected) => { + *connection_id.write().await = connected.connection_id; + socket = Some(connected.socket); + while let Some(request) = queued.pop_front() { + if !send_request( + request, + socket.as_mut().expect("reconnected socket is present"), + &mut pending_requests, + &sent_records, + ).await { + fail_pending_uncertain(&mut pending_requests); + start_reconnect(&config, &mut socket, &mut reconnect_task); + break; + } + } + } + Err(error) => { + for request in queued.drain(..) { + let _ = request.response_tx.send(Err(error.clone())); + } + break; + } + } + } } } - /// Close the transport. - pub async fn close(self) -> SdkResult<()> { - // Send close message - let _ = self.tx.send(Message::Close(None)); + if let Some(task) = reconnect_task { + task.abort(); + } + fail_pending_uncertain(&mut pending_requests); + for request in queued.drain(..) { + let _ = request.response_tx.send(Err(SdkError::new( + SdkErrorCode::ConnectionClosed, + "SDK client closed before request was sent", + ))); + } + if let Some(mut connected) = socket { + let _ = timeout(CLOSE_GRACE, connected.close(None)).await; + } +} - // Wait for background task to finish - if let Some(handle) = self.task_handle.lock().await.take() { - let _ = tokio::time::timeout(std::time::Duration::from_secs(5), handle).await; +fn start_reconnect( + config: &SdkClientConfig, + socket: &mut Option, + reconnect_task: &mut Option>>, +) { + socket.take(); + if reconnect_task.is_none() { + let config = config.clone(); + *reconnect_task = Some(tokio::spawn( + async move { connect_with_retry(&config).await }, + )); + } +} + +async fn send_request( + request: RequestMessage, + socket: &mut ClientSocket, + pending_requests: &mut HashMap, + sent_records: &Arc>, +) -> bool { + if Instant::now() >= request.deadline { + let _ = request.response_tx.send(Err(SdkError::timeout( + "SDK request deadline elapsed before send", + ))); + return true; + } + + let sent_record = SdkSentRecord { + id: request.id, + operation: request.operation, + sent_at: SystemTime::now(), + }; + sent_records + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .insert(sent_record.clone()); + pending_requests.insert( + request.id, + PendingRequest { + response_tx: request.response_tx, + sent_record: sent_record.clone(), + }, + ); + + let send_result = timeout_at( + request.deadline, + socket.send(Message::Text(request.serialized)), + ) + .await; + match send_result { + Ok(Ok(())) => true, + Ok(Err(error)) => { + if let Some(pending) = pending_requests.remove(&request.id) { + let _ = pending + .response_tx + .send(Err(SdkError::uncertain_after_send(pending.sent_record))); + } + tracing::warn!(%error, "SDK request send failed after dispatch began"); + false } + Err(_) => { + if let Some(pending) = pending_requests.remove(&request.id) { + let _ = pending + .response_tx + .send(Err(SdkError::uncertain_after_send(pending.sent_record))); + } + false + } + } +} - Ok(()) +fn cancel_request( + id: Uuid, + queued: &mut VecDeque, + pending_requests: &mut HashMap, + sent_records: &Arc>, +) -> Option { + if let Some(position) = queued.iter().position(|request| request.id == id) { + queued.remove(position); + return None; + } + if let Some(pending) = pending_requests.remove(&id) { + return Some(pending.sent_record); + } + sent_records + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .get(id) +} + +async fn handle_server_text( + text: &str, + pending_requests: &mut HashMap, + sent_records: &Arc>, + connection_id: &Arc>>, +) { + let frame = match SdkFrame::from_json(text) { + Ok(frame) => frame, + Err(error) => { + tracing::warn!(%error, "SDK server sent malformed JSON frame"); + return; + } + }; + if frame.is_server_hello() { + if let Some(id) = frame.connection_id().filter(|id| !id.is_empty()) { + *connection_id.write().await = Some(id.to_string()); + } + return; + } + + let id = match frame.response_id() { + Ok(Some(id)) => id, + Ok(None) => return, + Err(error) => { + tracing::warn!(%error, "SDK server sent an invalid correlation ID"); + return; + } + }; + if let Some(pending) = pending_requests.remove(&id) { + sent_records + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .remove(id); + let result = if frame.data.get("ok") == Some(&Value::Bool(false)) + || frame.data.get("status").and_then(Value::as_str) == Some("error") + { + Err(error_from_response(&frame)) + } else { + Ok(frame) + }; + let _ = pending.response_tx.send(result); + } +} + +fn error_from_response(frame: &SdkFrame) -> SdkError { + match frame.error() { + Some(Ok((code, message))) => SdkError::from_remote(&code, message), + Some(Err(error)) => error, + None => SdkError::new( + SdkErrorCode::Unavailable, + "SDK request failed without error details", + ), + } +} + +fn fail_pending_uncertain(pending_requests: &mut HashMap) { + for (_, pending) in pending_requests.drain() { + let _ = pending + .response_tx + .send(Err(SdkError::uncertain_after_send(pending.sent_record))); } } #[cfg(test)] mod tests { use super::*; + use serde_json::json; + use tokio::net::TcpListener; + use tokio_tungstenite::accept_async; + use tokio_tungstenite::WebSocketStream as TestWebSocketStream; + + type ServerSocket = TestWebSocketStream; + + async fn test_server() -> (TcpListener, String) { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + (listener, format!("ws://{address}/sdk?keep=1")) + } + + fn test_config(url: String) -> SdkClientConfig { + SdkClientConfig { + url, + token: "token with & symbols".to_string(), + connect_timeout: Duration::from_millis(250), + reconnect_deadline: Duration::from_secs(2), + reconnect_attempts: 1, + reconnect_initial_backoff: Duration::from_millis(5), + reconnect_max_backoff: Duration::from_millis(20), + ..SdkClientConfig::default() + } + } + + async fn accept(listener: &TcpListener) -> ServerSocket { + let (stream, _) = listener.accept().await.unwrap(); + accept_async(stream).await.unwrap() + } + + async fn read_frame(socket: &mut ServerSocket) -> SdkFrame { + loop { + match socket.next().await.unwrap().unwrap() { + Message::Text(text) => return SdkFrame::from_json(&text).unwrap(), + Message::Ping(payload) => socket.send(Message::Pong(payload)).await.unwrap(), + _ => {} + } + } + } + + async fn assert_no_text_frame(socket: &mut ServerSocket) { + if let Ok(Some(Ok(Message::Text(_)))) = + timeout(Duration::from_millis(50), socket.next()).await + { + panic!("client unexpectedly dispatched a request frame"); + } + } + + async fn complete_hello(socket: &mut ServerSocket, connection_id: &str) -> SdkFrame { + let hello = read_frame(socket).await; + assert_eq!(hello.frame_type().unwrap(), crate::frame::FrameType::Hello); + socket + .send(Message::Text( + json!({"type":"hello", "connectionId":connection_id}).to_string(), + )) + .await + .unwrap(); + hello + } + + async fn send_response(socket: &mut ServerSocket, id: Uuid, result: Value) { + socket + .send(Message::Text( + json!({"type":"broker_response", "id":id, "ok":true, "result":result}).to_string(), + )) + .await + .unwrap(); + } #[test] - fn config_default_values() { - let config = SdkTransportConfig::default(); - assert_eq!(config.url, "ws://127.0.0.1:8080"); - assert_eq!(config.max_payload_bytes, MAX_FRAME_BYTES); - assert_eq!(config.timeout_secs, 30); + fn config_defaults_use_ten_second_timeout_and_bounded_backoff() { + let config = SdkClientConfig::default(); + assert_eq!(config.request_timeout, Duration::from_secs(10)); + assert_eq!(config.max_frame_bytes, MAX_FRAME_BYTES); + let delays: Vec<_> = (0..8) + .map(|attempt| reconnect_delay(&config, attempt)) + .collect(); + assert_eq!( + delays, + [ + Duration::from_millis(25), + Duration::from_millis(50), + Duration::from_millis(100), + Duration::from_millis(200), + Duration::from_millis(400), + Duration::from_millis(800), + Duration::from_millis(1_600), + Duration::from_secs(2), + ] + ); + } + + #[test] + fn auth_url_preserves_query_and_encodes_token_once() { + let config = SdkClientConfig { + url: "wss://sdk.example/path?keep=a%20b&token=stale".to_string(), + token: "a token&value".to_string(), + ..SdkClientConfig::default() + }; + let url = authenticated_url(&config).unwrap(); + let pairs: Vec<_> = url + .query_pairs() + .map(|(key, value)| (key.into_owned(), value.into_owned())) + .collect(); + assert_eq!( + pairs, + [ + ("keep".to_string(), "a b".to_string()), + ("token".to_string(), "a token&value".to_string()) + ] + ); } #[tokio::test] - async fn transport_can_be_created() { - // This test doesn't actually connect, just checks that the struct can be created - let config = SdkTransportConfig::default(); - assert_eq!(config.url, "ws://127.0.0.1:8080"); + async fn rejects_non_websocket_endpoint_schemes() { + let config = SdkClientConfig { + url: "https://sdk.example".to_string(), + ..SdkClientConfig::default() + }; + assert_eq!( + SdkClient::connect(config).await.unwrap_err().code, + SdkErrorCode::InvalidInput + ); } - #[test] - fn request_frame_has_correlation_id() { - let frame = Frame::broker_request( - "test-id".to_string(), - "test.operation".to_string(), - Value::Object(serde_json::Map::new()), + #[tokio::test] + async fn handshake_advertises_capabilities_and_correlates_response_uuid() { + let (listener, url) = test_server().await; + let server = tokio::spawn(async move { + let mut socket = accept(&listener).await; + let hello = complete_hello(&mut socket, "conn-1").await; + assert_eq!(hello.data["protocolVersion"], PROTOCOL_VERSION); + assert_eq!(hello.data["capabilities"], json!(["session.observe.v1"])); + let request = read_frame(&mut socket).await; + let id = request.id().unwrap().unwrap(); + assert_eq!(request.operation().as_deref(), Some("session.get")); + send_response(&mut socket, id, json!({"session": 7})).await; + }); + let mut config = test_config(url); + config.capabilities = vec!["session.observe.v1".to_string()]; + let client = SdkClient::connect(config).await.unwrap(); + assert_eq!(client.connection_id().await.as_deref(), Some("conn-1")); + let response = client + .request("session.get", json!({"id": "s-1"})) + .await + .unwrap(); + assert_eq!(response.result(), Some(json!({"session": 7}))); + client.close().await.unwrap(); + server.await.unwrap(); + } + + #[tokio::test] + async fn requests_are_correlated_when_responses_arrive_out_of_order() { + let (listener, url) = test_server().await; + let server = tokio::spawn(async move { + let mut socket = accept(&listener).await; + complete_hello(&mut socket, "concurrent").await; + let first = read_frame(&mut socket).await; + let second = read_frame(&mut socket).await; + let first_id = first.id().unwrap().unwrap(); + let second_id = second.id().unwrap().unwrap(); + socket + .send(Message::Text( + json!({"type":"broker_response", "id":second_id, "ok":true, "result":"second"}) + .to_string(), + )) + .await + .unwrap(); + socket + .send(Message::Text( + json!({"type":"broker_response", "id":first_id, "ok":true, "result":"first"}) + .to_string(), + )) + .await + .unwrap(); + }); + let client = SdkClient::connect(test_config(url)).await.unwrap(); + let first = client.request("op.first", json!({})); + let second = client.request("op.second", json!({})); + let (first, second) = tokio::join!(first, second); + assert_eq!(first.unwrap().result(), Some(json!("first"))); + assert_eq!(second.unwrap().result(), Some(json!("second"))); + client.close().await.unwrap(); + server.await.unwrap(); + } + + #[tokio::test] + async fn broker_errors_are_returned_with_typed_codes() { + let (listener, url) = test_server().await; + let server = tokio::spawn(async move { + let mut socket = accept(&listener).await; + complete_hello(&mut socket, "errors").await; + let request = read_frame(&mut socket).await; + let id = request.id().unwrap().unwrap(); + socket + .send(Message::Text( + json!({"type":"broker_response", "id":id, "ok":false, "error":{"code":"not_found", "message":"missing"}}) + .to_string(), + )) + .await + .unwrap(); + }); + let client = SdkClient::connect(test_config(url)).await.unwrap(); + let error = client + .request("session.missing", json!({})) + .await + .unwrap_err(); + assert_eq!(error.code, SdkErrorCode::NotFound); + assert_eq!(error.message, "missing"); + assert!(error.sent_record.is_none()); + client.close().await.unwrap(); + server.await.unwrap(); + } + + #[tokio::test] + async fn timeout_override_marks_sent_request_uncertain_and_retains_record() { + let (listener, url) = test_server().await; + let server = tokio::spawn(async move { + let mut socket = accept(&listener).await; + complete_hello(&mut socket, "timeout").await; + let request = read_frame(&mut socket).await; + assert!(request.id().unwrap().is_some()); + tokio::time::sleep(Duration::from_millis(80)).await; + }); + let mut config = test_config(url); + config.request_timeout = Duration::from_secs(1); + let client = SdkClient::connect(config).await.unwrap(); + let error = client + .request_with_timeout( + "session.create", + json!({"secret": "not retained"}), + Duration::from_millis(20), + ) + .await + .unwrap_err(); + assert_eq!(error.code, SdkErrorCode::UncertainAfterSend); + let record = error.sent_record.unwrap(); + assert_eq!(record.operation.as_deref(), Some("session.create")); + assert_eq!(client.get_sent_record(record.id), Some(record.clone())); + assert_eq!(client.uncertain_sends(), vec![record.clone()]); + assert!(client.forget_sent_record(record.id)); + assert!(client.get_sent_record(record.id).is_none()); + client.close().await.unwrap(); + server.await.unwrap(); + } + + #[tokio::test] + async fn oversized_outgoing_frame_is_rejected_before_dispatch() { + let (listener, url) = test_server().await; + let server = tokio::spawn(async move { + let mut socket = accept(&listener).await; + complete_hello(&mut socket, "size").await; + assert_no_text_frame(&mut socket).await; + }); + let mut config = test_config(url); + config.max_frame_bytes = 128; + let client = SdkClient::connect(config).await.unwrap(); + let error = client + .request("large", json!({"payload": "x".repeat(256)})) + .await + .unwrap_err(); + assert_eq!(error.code, SdkErrorCode::PayloadTooLarge); + client.close().await.unwrap(); + server.await.unwrap(); + } + + #[tokio::test] + async fn reconnects_after_close_and_never_replays_uncertain_request() { + let (listener, url) = test_server().await; + let server = tokio::spawn(async move { + let mut first = accept(&listener).await; + complete_hello(&mut first, "first").await; + let uncertain = read_frame(&mut first).await; + let uncertain_id = uncertain.id().unwrap().unwrap(); + first.send(Message::Close(None)).await.unwrap(); + drop(first); + + let mut second = accept(&listener).await; + complete_hello(&mut second, "second").await; + let next = read_frame(&mut second).await; + let next_id = next.id().unwrap().unwrap(); + assert_ne!(next_id, uncertain_id); + assert_eq!(next.operation().as_deref(), Some("after.reconnect")); + send_response(&mut second, next_id, json!({"connected": true})).await; + }); + let client = SdkClient::connect(test_config(url)).await.unwrap(); + let uncertain = client.request("uncertain.mutation", json!({})); + let error = uncertain.await.unwrap_err(); + assert_eq!(error.code, SdkErrorCode::UncertainAfterSend); + let next = client.request("after.reconnect", json!({})).await.unwrap(); + assert_eq!(next.result(), Some(json!({"connected": true}))); + assert_eq!(client.connection_id().await.as_deref(), Some("second")); + client.close().await.unwrap(); + server.await.unwrap(); + } + + #[tokio::test] + async fn retries_failed_initial_connections_with_exponential_backoff() { + let (listener, url) = test_server().await; + let server = tokio::spawn(async move { + let (failed_stream, _) = listener.accept().await.unwrap(); + drop(failed_stream); + let (stream, _) = listener.accept().await.unwrap(); + let mut socket = accept_async(stream).await.unwrap(); + let hello = complete_hello(&mut socket, "retry-success").await; + assert_eq!(hello.frame_type().unwrap(), crate::frame::FrameType::Hello); + }); + let mut config = test_config(url); + config.reconnect_attempts = 2; + let client = SdkClient::connect(config).await.unwrap(); + assert_eq!( + client.connection_id().await.as_deref(), + Some("retry-success") ); - assert_eq!(frame.id(), Some("test-id".to_string())); + client.close().await.unwrap(); + server.await.unwrap(); + } + + #[tokio::test] + async fn connection_attempts_stop_at_the_reconnect_deadline() { + let (listener, url) = test_server().await; + let server = tokio::spawn(async move { + let _socket = accept(&listener).await; + tokio::time::sleep(Duration::from_millis(100)).await; + }); + let mut config = test_config(url); + config.connect_timeout = Duration::from_millis(500); + config.reconnect_deadline = Duration::from_millis(25); + config.reconnect_attempts = 8; + let error = SdkClient::connect(config).await.unwrap_err(); + assert_eq!(error.code, SdkErrorCode::Timeout); + assert!(error.message.contains("deadline")); + server.await.unwrap(); + } + + #[tokio::test] + async fn exhausted_reconnect_budget_returns_typed_error() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("ws://{}", listener.local_addr().unwrap()); + drop(listener); + let mut config = test_config(url); + config.reconnect_attempts = 0; + let error = SdkClient::connect(config).await.unwrap_err(); + assert_eq!(error.code, SdkErrorCode::ReconnectExhausted); + } + + #[tokio::test] + async fn request_timeout_before_send_does_not_create_uncertain_record() { + let (listener, url) = test_server().await; + let server = tokio::spawn(async move { + let mut socket = accept(&listener).await; + complete_hello(&mut socket, "immediate-timeout").await; + assert_no_text_frame(&mut socket).await; + }); + let client = SdkClient::connect(test_config(url)).await.unwrap(); + let error = client + .request_with_timeout("too.short", json!({}), Duration::ZERO) + .await + .unwrap_err(); + assert_eq!(error.code, SdkErrorCode::Timeout); + assert!(client.uncertain_sends().is_empty()); + client.close().await.unwrap(); + server.await.unwrap(); + } + + #[tokio::test] + async fn malformed_server_frames_do_not_settle_unrelated_requests() { + let (listener, url) = test_server().await; + let server = tokio::spawn(async move { + let mut socket = accept(&listener).await; + complete_hello(&mut socket, "malformed").await; + let request = read_frame(&mut socket).await; + let id = request.id().unwrap().unwrap(); + socket + .send(Message::Text("not-json".to_string())) + .await + .unwrap(); + send_response(&mut socket, id, json!("ok-after-malformed")).await; + }); + let client = SdkClient::connect(test_config(url)).await.unwrap(); + let response = client.request("malformed.noise", json!({})).await.unwrap(); + assert_eq!(response.result(), Some(json!("ok-after-malformed"))); + client.close().await.unwrap(); + server.await.unwrap(); + } + + async fn round_trip_case(case: u8) { + let (listener, url) = test_server().await; + let operation = format!("round.trip.case_{case:03}"); + let input = json!({"case": case, "value": format!("request-{case}")}); + let server_operation = operation.clone(); + let server_input = input.clone(); + let server = tokio::spawn(async move { + let mut socket = accept(&listener).await; + let hello = complete_hello(&mut socket, "matrix").await; + assert_eq!( + hello.data["capabilities"], + json!([format!("matrix.{case}")]) + ); + + let request = read_frame(&mut socket).await; + let id = request.id().unwrap().expect("request has a UUID ID"); + assert_eq!(request.operation(), Some(server_operation)); + assert_eq!(request.input(), Some(server_input)); + send_response(&mut socket, id, json!({"echo": format!("request-{case}")})).await; + }); + + let mut config = test_config(url); + config.capabilities = vec![format!("matrix.{case}")]; + let client = SdkClient::connect(config).await.unwrap(); + let response = client.request(operation, input).await.unwrap(); + assert_eq!( + response.result(), + Some(json!({"echo": format!("request-{case}")})) + ); + client.close().await.unwrap(); + server.await.unwrap(); + } + + macro_rules! sdk_round_trip_matrix { + ($($name:ident: $case:literal),+ $(,)?) => { + $( + #[tokio::test] + async fn $name() { + round_trip_case($case).await; + } + )+ + }; + } + + sdk_round_trip_matrix! { + protocol_case_000: 0, protocol_case_001: 1, protocol_case_002: 2, protocol_case_003: 3, + protocol_case_004: 4, protocol_case_005: 5, protocol_case_006: 6, protocol_case_007: 7, + protocol_case_008: 8, protocol_case_009: 9, protocol_case_010: 10, protocol_case_011: 11, + protocol_case_012: 12, protocol_case_013: 13, protocol_case_014: 14, protocol_case_015: 15, + protocol_case_016: 16, protocol_case_017: 17, protocol_case_018: 18, protocol_case_019: 19, + protocol_case_020: 20, protocol_case_021: 21, protocol_case_022: 22, protocol_case_023: 23, + protocol_case_024: 24, protocol_case_025: 25, protocol_case_026: 26, protocol_case_027: 27, + protocol_case_028: 28, protocol_case_029: 29, protocol_case_030: 30, protocol_case_031: 31, + protocol_case_032: 32, protocol_case_033: 33, protocol_case_034: 34, protocol_case_035: 35, + protocol_case_036: 36, protocol_case_037: 37, protocol_case_038: 38, protocol_case_039: 39, + protocol_case_040: 40, protocol_case_041: 41, protocol_case_042: 42, protocol_case_043: 43, + protocol_case_044: 44, protocol_case_045: 45, protocol_case_046: 46, protocol_case_047: 47, + protocol_case_048: 48, protocol_case_049: 49, protocol_case_050: 50, protocol_case_051: 51, + protocol_case_052: 52, protocol_case_053: 53, protocol_case_054: 54, protocol_case_055: 55, + protocol_case_056: 56, protocol_case_057: 57, protocol_case_058: 58, protocol_case_059: 59, + protocol_case_060: 60, protocol_case_061: 61, protocol_case_062: 62, protocol_case_063: 63, } } diff --git a/rust/crates/sdk/tests/operation_registry_gap.rs b/rust/crates/sdk/tests/operation_registry_gap.rs index 550b6577..3b8c1f7b 100644 --- a/rust/crates/sdk/tests/operation_registry_gap.rs +++ b/rust/crates/sdk/tests/operation_registry_gap.rs @@ -10,17 +10,18 @@ //! - Schema validation for operation inputs/outputs //! - Discovery of available operations -use sdk::{Frame, OperationMetadata, OperationRegistry, Schema}; +use sdk::{OperationMetadata, OperationRegistry, Schema, SdkFrame}; use serde_json::json; +use uuid::Uuid; + +fn test_id() -> Uuid { + Uuid::new_v4() +} #[test] fn gap_sdk_allows_undefined_operations() { // BEFORE: No validation - any operation name is accepted - let frame = Frame::broker_request( - "test-id".to_string(), - "undefined.operation".to_string(), - json!({}), - ); + let frame = SdkFrame::broker_request(test_id(), "undefined.operation".to_string(), json!({})); // The frame is created successfully with no validation assert_eq!(frame.operation(), Some("undefined.operation".to_string())); @@ -48,9 +49,9 @@ fn fix_operation_registry_validates_operations() { #[test] fn gap_sdk_has_no_operation_discovery() { // BEFORE: No way to discover what operations are available - // The Frame struct has no way to enumerate operations - let frame = Frame::broker_request( - "test-id".to_string(), + // The SdkFrame struct has no way to enumerate operations + let frame = SdkFrame::broker_request( + test_id(), "query.list_sessions".to_string(), json!({"some": "input"}), ); @@ -84,14 +85,14 @@ fn fix_operation_registry_enables_discovery() { fn gap_sdk_has_no_input_schema_validation() { // BEFORE: No validation of input schemas // Any input shape is accepted for any operation - let frame_with_valid_input = Frame::broker_request( - "test-1".to_string(), + let frame_with_valid_input = SdkFrame::broker_request( + test_id(), "query.get_session".to_string(), json!({"session_id": "123"}), // Correct structure ); - let frame_with_invalid_input = Frame::broker_request( - "test-2".to_string(), + let frame_with_invalid_input = SdkFrame::broker_request( + test_id(), "query.get_session".to_string(), json!({"wrong_field": "value"}), // Missing session_id, wrong field ); @@ -126,8 +127,8 @@ fn fix_operation_registry_validates_input_schemas() { #[test] fn gap_sdk_has_no_output_schema_validation() { // BEFORE: No validation of response schemas - let response_frame = Frame::broker_response_ok( - "test-id".to_string(), + let response_frame = SdkFrame::broker_response_ok( + test_id(), json!({"any": "response"}), // No validation of structure ); @@ -158,8 +159,8 @@ fn fix_operation_registry_validates_output_schemas() { #[test] fn gap_sdk_transports_ignore_operation_names() { // BEFORE: The transport layer sends any operation without validation - let frame = Frame::broker_request( - "req-123".to_string(), + let frame = SdkFrame::broker_request( + test_id(), "this.is.not.a.real.operation".to_string(), json!({"data": "value"}), ); @@ -175,11 +176,8 @@ fn fix_registry_integrates_with_frame_operations() { let registry = OperationRegistry::with_standard_operations(); // Create a frame with a valid operation - let valid_frame = Frame::broker_request( - "req-123".to_string(), - "query.list_sessions".to_string(), - json!({}), - ); + let valid_frame = + SdkFrame::broker_request(test_id(), "query.list_sessions".to_string(), json!({})); // Extract and validate the operation if let Some(op_name) = valid_frame.operation() { @@ -189,11 +187,8 @@ fn fix_registry_integrates_with_frame_operations() { } // Create a frame with an invalid operation - let invalid_frame = Frame::broker_request( - "req-124".to_string(), - "invalid.operation".to_string(), - json!({}), - ); + let invalid_frame = + SdkFrame::broker_request(test_id(), "invalid.operation".to_string(), json!({})); // Extract and validate the operation if let Some(op_name) = invalid_frame.operation() {