From eb92878e3c696c768f1d2294c3718ecd04cf6a87 Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 16 Oct 2026 05:26:11 +0900 Subject: [PATCH] feat: support typed in-process WIP clients without HTTP --- MIGRATION.md | 41 ++ crates/wip-client/Cargo.toml | 14 +- crates/wip-client/README.md | 20 +- crates/wip-client/src/http_adapter.rs | 346 ++++++++++++++ crates/wip-client/src/lib.rs | 347 ++++++++------ crates/wip-client/src/validation.rs | 80 ++++ crates/wip-client/tests/direct.rs | 614 +++++++++++++++++++++++++ crates/wip-client/tests/parity.rs | 289 ++++++++++++ crates/wip-client/tests/wire_limits.rs | 128 ++++++ 9 files changed, 1725 insertions(+), 154 deletions(-) create mode 100644 crates/wip-client/src/http_adapter.rs create mode 100644 crates/wip-client/src/validation.rs create mode 100644 crates/wip-client/tests/direct.rs create mode 100644 crates/wip-client/tests/parity.rs create mode 100644 crates/wip-client/tests/wire_limits.rs diff --git a/MIGRATION.md b/MIGRATION.md index 1c25ebe..a8a30bb 100644 --- a/MIGRATION.md +++ b/MIGRATION.md @@ -1,5 +1,46 @@ # Structured Interface reference and scope publication migration +## Transport-independent Client API + +The default `wip-client` feature set still exposes the HTTP-compatible +`wip_client::Client` and `PreparedRequest`. Its constructor takes state and wire +limits, sessions canonicalize HTTP endpoints, and `complete` consumes an HTTP +response. Existing TUI and native HTTP consumers need no source migration; the +wire contract is unchanged. + +For an in-process Host (including Client and Host in one WASM module), disable +default features and use `wip_client::direct::{Client, Request, Response}`: + +```toml +wip-client = { version = "0.2.0", default-features = false } +``` + +- `direct::Client::new(ClientLimits)` takes no wire limits. +- `open_session(worldspace_identity, SecurityContext)` accepts an opaque identity, + not a URL. Issue a separate identity per Worldspace view/Host lifetime and a + separate security identity whenever authorization context changes. Direct and + canonical HTTP session identities cannot alias each other's cache. +- Preparation methods retain their names and arguments. Their result is + `direct::PreparedRequest { id, request: Request }`, with `Observe`, + `FetchInterface`, and `CallOperation` variants containing Protocol request DTOs. +- Route that exact DTO to the session's Host/security context; mark calls + dispatched before invoking the Host. Complete with the original `id` and the + matching `Response` variant containing `Result`. + No HTTP request, response, JSON codec, or artificial endpoint is needed. +- Direct responses undergo the same semantic validation and state transitions + as HTTP responses. Wrong interactions and invalid results are rejected; + invalid call responses retain an unknown execution outcome. Retrievals may + be explicitly refreshed, but operations are never automatically retried. +- `ClientLimits` still bounds cache, sessions, in-flight requests, and history. + HTTP status/envelope checks and byte/JSON nesting limits belong to the optional + HTTP adapter. An embedding application owns DTO allocation/execution budgets. + +The `direct` module API documentation contains a runnable typed example. Run +`cargo test -p wip-client --no-default-features` to exercise the core without an +HTTP dependency; ordinary `cargo test -p wip-client` also runs unchanged HTTP +regressions and direct/HTTP state parity tests. + + The 0.2.0 API/wire change follows `wip-reference` canonical Core Model, Protocol and WIP over HTTP at `6086f3c5ef10aa1464ce9c824750d7f667217bfd`. It replaces the earlier opaque Interface reference contract; HTTP routes and diff --git a/crates/wip-client/Cargo.toml b/crates/wip-client/Cargo.toml index 740b367..f504fe6 100644 --- a/crates/wip-client/Cargo.toml +++ b/crates/wip-client/Cargo.toml @@ -9,6 +9,18 @@ repository.workspace = true readme = "README.md" documentation = "https://docs.rs/wip-client" +[features] +default = ["http"] +http = ["dep:wip-http"] + [dependencies] -wip-http.workspace = true +wip-http = { workspace = true, optional = true } wip-protocol.workspace = true + +[[test]] +name = "adapter" +required-features = ["http"] + +[[test]] +name = "runtime" +required-features = ["http"] diff --git a/crates/wip-client/README.md b/crates/wip-client/README.md index 84c96ba..de7b82b 100644 --- a/crates/wip-client/README.md +++ b/crates/wip-client/README.md @@ -2,14 +2,30 @@ Presentation-independent stateful runtime for Worldspace Interface Protocol (WIP) clients. -`wip-client` manages endpoint- and security-context-isolated Known Space, object and Interface Descriptor observations, bounded request ownership, supersession, out-of-order response suppression, validator recovery, and operation call contexts. It produces and consumes framework-neutral `wip-http` messages but performs no network I/O and contains no UI types. +`wip-client` manages Worldspace-view- and security-context-isolated Known Space, object and Interface Descriptor observations, bounded request ownership, supersession, out-of-order response suppression, validator recovery, and operation call contexts. Its typed logical core requires neither HTTP nor JSON. The default HTTP facade produces and consumes framework-neutral `wip-http` messages using the same logical runtime. Neither path performs network I/O or contains UI types. ```toml [dependencies] wip-client = "0.2.0" ``` -Applications provide the HTTP adapter and presentation layer, send each prepared request exactly as owned by the runtime, then return the corresponding response for completion. +The default `http` feature preserves `wip_client::Client::new(state_limits, wire_limits)`, canonical HTTP endpoint sessions, `PreparedRequest`, and HTTP `complete`. Applications send each prepared request and return its corresponding response. + +## Direct typed Client (no HTTP dependency) + +```toml +[dependencies] +wip-client = { version = "0.2.0", default-features = false } +wip-protocol = "0.2.0" +``` + +Use `wip_client::direct::{Client, Request, Response}`. Construct the client with `Client::new(state_limits)` and open a session using an opaque, application-owned Worldspace view identity plus `SecurityContext`; no URL is parsed or required. Allocate different identities for different Hosts/views and when replacing a Host. Direct identities and HTTP endpoint identities occupy distinct namespaces, even when their strings match. + +The existing observation, interface acquisition, and call preparation methods return `direct::PreparedRequest { id, request }`. Dispatch the `Request::Observe`, `Request::FetchInterface`, or `Request::CallOperation` DTO to the Host, then call `complete(id, Response::(result))`, where `result` is `Result`. The prepared request must be dispatched unchanged; routing it to the session's Host/security context remains the application's responsibility. Mark calls dispatched immediately before invoking the Host. + +Typed results are **not trusted**: the runtime checks interaction correspondence, observation depth/names/paths, scoped Interface correspondence, descriptor structure and format, descriptor-bound arguments/results, and per-interaction Protocol error codes. Invalid call responses leave the outcome unknown, just as with HTTP. HTTP status/envelope classification, JSON encoding, and wire size/depth limits remain exclusive to the optional adapter; `ClientLimits` bounds retained logical state on both paths. Direct DTO allocation and execution budgets belong to the embedding application. + +See [migration guidance](../../MIGRATION.md#transport-independent-client-api) and the `direct` module's executable API example. ## Interface identity and scope lifetime diff --git a/crates/wip-client/src/http_adapter.rs b/crates/wip-client/src/http_adapter.rs new file mode 100644 index 0000000..ac26299 --- /dev/null +++ b/crates/wip-client/src/http_adapter.rs @@ -0,0 +1,346 @@ +//! Optional HTTP facade. Encoding and wire classification live only here. +use std::ops::{Deref, DerefMut}; + +use wip_http::http::{Request, Response}; +use wip_http::{ + ClientResponseError, DecodedResponse, Endpoint, Limits, StatusMismatch, + decode_call_operation_response, decode_fetch_interface_response, decode_observe_response, + encode_call_operation_request, encode_fetch_interface_request, encode_observe_request, +}; + +use super::*; + +/// Encoded HTTP request paired with its runtime identity. +#[derive(Debug)] +pub struct PreparedRequest { + /// Identity required when completing or failing this request. + pub id: RequestId, + /// Standard WIP over HTTP request. + pub request: Request>, +} + +/// HTTP-compatible facade over the typed logical client. +/// +/// State inspection and transport failure reporting delegate to the same core. +/// Request encoding is performed before changing cache or in-flight state. +#[derive(Debug)] +pub struct Client { + core: DirectClient, + wire_limits: Limits, +} + +impl Deref for Client { + type Target = DirectClient; + fn deref(&self) -> &Self::Target { + &self.core + } +} + +impl DerefMut for Client { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.core + } +} + +impl Client { + /// Creates an empty runtime with explicit state and wire bounds. + #[must_use] + pub fn new(limits: ClientLimits, wire_limits: Limits) -> Self { + Self { + core: DirectClient::new(limits), + wire_limits, + } + } + + /// Returns one cached object observation. + #[must_use] + pub fn object(&self, session: &SessionId, path: &str) -> Option { + self.core.object(session, path) + } + + /// Returns bounded Known Space in canonical path order. + #[must_use] + pub fn known_space(&self, session: &SessionId) -> Option> { + self.core.known_space(session) + } + + /// Returns one materialized lazy-tree node. + #[must_use] + pub fn tree(&self, session: &SessionId, path: &str) -> Option<&TreeObservation> { + self.core.tree(session, path) + } + + /// Returns the descriptor observation for a complete scoped reference. + #[must_use] + pub fn interface( + &self, + session: &SessionId, + reference: &InterfaceReference, + ) -> Option<&InterfaceObservation> { + self.core.interface(session, reference) + } + + /// Returns bounded operation history from oldest to newest. + #[must_use] + pub fn call_history(&self, session: &SessionId) -> Option<&VecDeque> { + self.core.call_history(session) + } + + /// Returns bounded diagnostic history from oldest to newest. + #[must_use] + pub fn diagnostics(&self, session: &SessionId) -> Option<&VecDeque> { + self.core.diagnostics(session) + } + + /// Returns the number of requests currently in flight. + #[must_use] + pub fn in_flight_len(&self) -> usize { + self.core.in_flight_len() + } + + /// Returns the immutable context of an in-flight operation. + #[must_use] + pub fn pending_call_context(&self, id: RequestId) -> Option<&CallContext> { + self.core.pending_call_context(id) + } + + /// Marks an operation as handed to the transport. + pub fn mark_dispatched(&mut self, id: RequestId) -> Result<(), ClientError> { + self.core.mark_dispatched(id) + } + + /// Records a failure with no response, preserving the dispatch boundary. + pub fn fail_transport( + &mut self, + id: RequestId, + detail: impl Into, + ) -> Result { + self.core.fail_transport(id, detail) + } + + /// Records timeout or disconnect after dispatch without retrying. + pub fn fail_dispatched_call( + &mut self, + id: RequestId, + failure: DispatchedTransportFailure, + detail: impl Into, + ) -> Result { + self.core.fail_dispatched_call(id, failure, detail) + } + + /// Opens a session isolated by canonical endpoint and security context. + pub fn open_session( + &mut self, + endpoint: &str, + security_context: SecurityContext, + ) -> Result { + let endpoint = Endpoint::parse(endpoint).map_err(invalid_request)?; + self.core.open_session_id(SessionId { + endpoint: endpoint.to_string(), + security_context, + binding: SessionBinding::Http, + }) + } + + /// Ensures requested coverage without duplicating fresh or loading work. + pub fn ensure_observed( + &mut self, + session: &SessionId, + path: impl Into, + depth: u32, + ) -> Result, ClientError> { + let path = path.into(); + wip_protocol::validate_path(&path).map_err(invalid_request)?; + let coverage = observation_coverage(self.core.session(session)?, &path, depth); + if coverage != ObservationCoverage::Missing + || self.core.pending.len() >= self.core.limits.in_flight + { + return Ok(None); + } + self.refresh_observed(session, path, depth).map(Some) + } + + /// Ensures direct children without duplicating fresh or loading work. + pub fn ensure_children_observed( + &mut self, + session: &SessionId, + path: impl Into, + ) -> Result, ClientError> { + self.ensure_observed(session, path, 1) + } + + /// Refreshes one object without replacing previously observed children. + pub fn refresh_object( + &mut self, + session: &SessionId, + path: impl Into, + ) -> Result { + self.refresh_observed(session, path, 0) + } + + /// Refreshes one object and its direct children. + pub fn refresh_children( + &mut self, + session: &SessionId, + path: impl Into, + ) -> Result { + self.refresh_observed(session, path, 1) + } + + /// Explicitly refreshes observation coverage, superseding older requests. + pub fn refresh_observed( + &mut self, + session: &SessionId, + path: impl Into, + depth: u32, + ) -> Result { + let endpoint = endpoint(session)?; + let limits = self.wire_limits; + let (id, request) = self + .core + .refresh_observed_with(session, path, depth, |request| { + encode_observe_request(&endpoint, request, limits).map_err(invalid_request) + })?; + Ok(PreparedRequest { id, request }) + } + + /// Ensures an interface without duplicating work or evicting another descriptor. + pub fn ensure_interface( + &mut self, + session: &SessionId, + reference: InterfaceReference, + ) -> Result, ClientError> { + let state = self.core.session(session)?; + if state.interfaces.contains_key(&reference) + || state.interfaces.len() >= self.core.limits.interfaces_per_session + || self.core.pending.len() >= self.core.limits.in_flight + { + return Ok(None); + } + self.prepare_interface(session, reference).map(Some) + } + + /// Explicitly starts or refreshes a descriptor observation. + pub fn prepare_interface( + &mut self, + session: &SessionId, + reference: InterfaceReference, + ) -> Result { + let endpoint = endpoint(session)?; + let limits = self.wire_limits; + let (id, request) = self + .core + .prepare_interface_with(session, reference, |request| { + encode_fetch_interface_request(&endpoint, request, limits).map_err(invalid_request) + })?; + Ok(PreparedRequest { id, request }) + } + + /// Constructs an operation using fresh observations and an immutable CallContext. + pub fn prepare_call( + &mut self, + session: &SessionId, + path: &str, + interface: &InterfaceReference, + operation: impl Into, + arguments: BTreeMap, + ) -> Result { + let endpoint = endpoint(session)?; + let limits = self.wire_limits; + let (id, request) = self.core.prepare_call_with( + session, + path, + interface, + operation, + arguments, + |request, descriptor| { + encode_call_operation_request(&endpoint, request, descriptor, limits) + .map_err(invalid_request) + }, + )?; + Ok(PreparedRequest { id, request }) + } + + /// Decodes HTTP metadata and body, then validates and applies the logical result. + pub fn complete( + &mut self, + id: RequestId, + response: Response>, + ) -> Result { + let pending = self.core.pending.get(&id).ok_or_else(unknown_request)?; + let session = pending.session().clone(); + let mut mismatch = None; + let decoded = match pending { + Pending::Observe { request, .. } => { + decode_observe_response(request, &response, self.wire_limits) + .map(|value| LogicalResponse::Observe(unpack(value, &mut mismatch))) + } + Pending::Interface { request, .. } => { + decode_fetch_interface_response(request, &response, self.wire_limits) + .map(|value| LogicalResponse::FetchInterface(unpack(value, &mut mismatch))) + } + Pending::Call { context, .. } => decode_call_operation_response( + &context.request, + &context.descriptor, + &response, + self.wire_limits, + ) + .map(|value| LogicalResponse::CallOperation(unpack(value, &mut mismatch))), + } + .map_err(classify_response_error); + let completion = self.core.complete_result(id, decoded)?; + if !matches!(completion, Completion::StaleResponseRejected(_)) + && let Some(mismatch) = mismatch + { + let maximum = self.core.limits.history; + self.core.session_mut(&session)?.push_diagnostic( + Diagnostic { + request_id: id, + kind: DiagnosticKind::HttpStatusMismatch, + detail: format!( + "received HTTP {}; canonical status is {}", + mismatch.actual, mismatch.expected + ), + }, + maximum, + ); + } + Ok(completion) + } +} + +fn endpoint(session: &SessionId) -> Result { + if session.binding != SessionBinding::Http { + return Err(invalid_request( + "direct Worldspace session cannot be used as an HTTP endpoint", + )); + } + Endpoint::parse(&session.endpoint).map_err(invalid_request) +} + +fn unpack( + value: DecodedResponse, + mismatch: &mut Option, +) -> Result { + match value { + DecodedResponse::Success(value) => Ok(value), + DecodedResponse::ProtocolFailure { + error, + status_mismatch, + } => { + *mismatch = status_mismatch; + Err(error) + } + } +} + +fn classify_response_error(error: ClientResponseError) -> ClientError { + let kind = match error { + ClientResponseError::InvalidResponse { .. } => ClientErrorKind::InvalidResponse, + ClientResponseError::UnsupportedDescriptorFormat { .. } => { + ClientErrorKind::UnsupportedDescriptorFormat + } + ClientResponseError::TransportBinding(_) => ClientErrorKind::Transport, + }; + ClientError::new(kind, error.to_string()) +} diff --git a/crates/wip-client/src/lib.rs b/crates/wip-client/src/lib.rs index d107bfc..f67f21e 100644 --- a/crates/wip-client/src/lib.rs +++ b/crates/wip-client/src/lib.rs @@ -1,28 +1,65 @@ -//! Presentation-independent stateful runtime for WIP clients. +//! Stateful WIP client with a transport-independent logical core. //! -//! [`Client`] separates observations by canonical WIP over HTTP endpoint and caller- -//! supplied security context. It reconciles requested observation coverage without -//! duplicating fresh or loading work, creates requests with the standard -//! [`wip_http`] adapter, correlates responses by [`RequestId`], and retains -//! bounded object, tree, interface, diagnostic, and operation state. The crate -//! contains no terminal, widget, or UI event types. +//! Use [`direct::Client`] for typed in-process interactions without HTTP or JSON. +//! The default `http` feature retains the original `Client` HTTP facade. +//! Both paths share bounded state, validation, correlation, and Scope lifetime handling. #![deny(missing_docs)] +mod validation; +use validation::{invalid_request, validate_call, validate_interface, validate_observe}; + use std::collections::{BTreeMap, VecDeque}; use std::error::Error; use std::fmt::{self, Display, Formatter}; -use wip_http::http::{Request, Response}; -use wip_http::{ - ClientResponseError, DecodedResponse, Endpoint, Limits, decode_call_operation_response, - decode_fetch_interface_response, decode_observe_response, encode_call_operation_request, - encode_fetch_interface_request, encode_observe_request, -}; +#[cfg(feature = "http")] +mod http_adapter; +#[cfg(feature = "http")] +pub use http_adapter::{Client, PreparedRequest}; + +/// Typed, transport-independent client API. Available without default features. +/// +/// ``` +/// use wip_client::direct::{Client, Request, Response}; +/// use wip_client::{ClientLimits, Completion, SecurityContext}; +/// use wip_protocol::{Object, ObjectObservation}; +/// +/// let mut client = Client::new(ClientLimits::new(2, 32, 16, 8, 16)?); +/// let session = client.open_session("embedded-world-v1", SecurityContext::new("reader"))?; +/// let prepared = client.refresh_object(&session, "/")?; +/// let Request::Observe(request) = prepared.request else { unreachable!() }; +/// assert_eq!(request.path, "/"); +/// // An in-process Host returns this DTO directly, without HTTP or JSON. +/// let result = Ok(ObjectObservation { +/// object: Object { +/// name: String::new(), description: None, interfaces: vec![], +/// r#ref: None, validator: None, +/// }, +/// children: None, +/// }); +/// let completion = client.complete(prepared.id, Response::Observe(result))?; +/// assert!(matches!(completion, Completion::Object(_))); +/// # Ok::<(), Box>(()) +/// ``` +/// +/// `FetchInterface` and `CallOperation` use the corresponding request/response +/// variants. Dispatch each prepared request unchanged to its session's Host and +/// security context. Call [`DirectClient::mark_dispatched`] immediately before invoking +/// an Operation. Typed responses are untrusted and validated against the pending +/// request and immutable CallContext; the runtime never automatically retries. +pub mod direct { + pub use super::{ + DirectClient as Client, DirectPreparedRequest as PreparedRequest, + LogicalRequest as Request, LogicalResponse as Response, + }; +} + use wip_protocol::{ - CallOperationRequest, CallOperationResponse, FetchInterfaceRequest, InterfaceDescriptor, - InterfaceReference, InterfaceTarget, Object, ObjectObservation as ProtocolObjectObservation, - ObserveRequest, OperationDeclaration, ProtocolError, ProtocolErrorCode, Target, Value, + CallOperationRequest, CallOperationResponse, FetchInterfaceRequest, FetchInterfaceResponse, + INTERFACE_FORMAT_V1, InterfaceDescriptor, InterfaceReference, InterfaceTarget, Object, + ObjectObservation as ProtocolObjectObservation, ObserveRequest, OperationDeclaration, + ProtocolError, ProtocolErrorCode, ProtocolInteraction, Target, Value, }; /// Monotonic identity assigned to one logical request. @@ -58,15 +95,23 @@ impl SecurityContext { } } +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] +enum SessionBinding { + Direct, + #[cfg(feature = "http")] + Http, +} + /// Identity of an isolated client session. #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] pub struct SessionId { endpoint: String, security_context: SecurityContext, + binding: SessionBinding, } impl SessionId { - /// Returns the canonical endpoint URL, including its trailing slash. + /// Returns the canonical HTTP endpoint or application-defined direct Worldspace identity. #[must_use] pub fn endpoint(&self) -> &str { &self.endpoint @@ -368,13 +413,35 @@ pub struct Diagnostic { pub detail: String, } -/// Encoded HTTP request paired with its runtime identity. +/// Typed request paired with its runtime identity. #[derive(Debug)] -pub struct PreparedRequest { +pub struct DirectPreparedRequest { /// Identity required when completing or failing this request. pub id: RequestId, - /// Standard WIP over HTTP request. - pub request: Request>, + /// Logical interaction to dispatch to the selected Worldspace. + pub request: LogicalRequest, +} + +/// A typed request with no transport encoding. +#[derive(Debug, Clone, PartialEq)] +pub enum LogicalRequest { + /// Observe an Object and requested descendants. + Observe(ObserveRequest), + /// Fetch a scoped Interface descriptor. + FetchInterface(FetchInterfaceRequest), + /// Invoke an Operation using immutable prepared inputs. + CallOperation(CallOperationRequest), +} + +/// An untrusted typed result. Every completion is validated against its request. +#[derive(Debug, Clone, PartialEq)] +pub enum LogicalResponse { + /// Observe success or Protocol failure. + Observe(Result), + /// Interface success or Protocol failure. + FetchInterface(Result), + /// Operation success or Protocol failure. + CallOperation(Result), } /// Result of applying a correlated response to the runtime. @@ -553,40 +620,43 @@ impl Pending { /// Stateful, presentation-independent WIP client runtime. #[derive(Debug)] -pub struct Client { +pub struct DirectClient { limits: ClientLimits, - wire_limits: Limits, sessions: BTreeMap, pending: BTreeMap, next_request_id: u64, } -impl Client { - /// Creates an empty runtime with explicit state and wire bounds. +impl DirectClient { + /// Creates an empty logical runtime with explicit state bounds. #[must_use] - pub fn new(limits: ClientLimits, wire_limits: Limits) -> Self { + pub fn new(limits: ClientLimits) -> Self { Self { limits, - wire_limits, sessions: BTreeMap::new(), pending: BTreeMap::new(), next_request_id: 1, } } - /// Opens or returns a session isolated by canonical endpoint and security context. + /// Opens or returns a session isolated by Worldspace view and security context. + /// + /// The identity is opaque, not a URL. Use different identities for different + /// Worldspace views and replace it when an in-process Host is replaced. + /// No endpoint parsing, HTTP binding, or JSON codec is involved. pub fn open_session( &mut self, endpoint: &str, security_context: SecurityContext, ) -> Result { - let endpoint = Endpoint::parse(endpoint).map_err(|error| { - ClientError::new(ClientErrorKind::InvalidRequest, error.to_string()) - })?; - let id = SessionId { - endpoint: endpoint.to_string(), + self.open_session_id(SessionId { + endpoint: endpoint.to_owned(), security_context, - }; + binding: SessionBinding::Direct, + }) + } + + fn open_session_id(&mut self, id: SessionId) -> Result { if !self.sessions.contains_key(&id) { if self.sessions.len() >= self.limits.sessions { return Err(ClientError::new( @@ -679,7 +749,7 @@ impl Client { session: &SessionId, path: impl Into, depth: u32, - ) -> Result, ClientError> { + ) -> Result, ClientError> { let path = path.into(); wip_protocol::validate_path(&path).map_err(|error| { ClientError::new(ClientErrorKind::InvalidRequest, error.to_string()) @@ -697,7 +767,7 @@ impl Client { &mut self, session: &SessionId, path: impl Into, - ) -> Result, ClientError> { + ) -> Result, ClientError> { self.ensure_observed(session, path, 1) } @@ -707,7 +777,7 @@ impl Client { &mut self, session: &SessionId, path: impl Into, - ) -> Result { + ) -> Result { self.refresh_observed(session, path, 0) } @@ -716,30 +786,40 @@ impl Client { &mut self, session: &SessionId, path: impl Into, - ) -> Result { + ) -> Result { self.refresh_observed(session, path, 1) } /// Explicitly starts or refreshes an object observation through `depth`. /// - /// Unlike [`Client::ensure_observed`], this always creates a new request and + /// Unlike [`DirectClient::ensure_observed`], this always creates a new request and /// supersedes older requests for the same root path. pub fn refresh_observed( &mut self, session: &SessionId, path: impl Into, depth: u32, - ) -> Result { + ) -> Result { + let (id, request) = self.refresh_observed_with(session, path, depth, |request| { + Ok(LogicalRequest::Observe(request.clone())) + })?; + Ok(DirectPreparedRequest { id, request }) + } + + fn refresh_observed_with( + &mut self, + session: &SessionId, + path: impl Into, + depth: u32, + encode: impl FnOnce(&ObserveRequest) -> Result, + ) -> Result<(RequestId, T), ClientError> { self.ensure_capacity()?; let request = ObserveRequest { path: path.into(), depth, }; - let endpoint = session_endpoint(session)?; - let encoded = - encode_observe_request(&endpoint, &request, self.wire_limits).map_err(|error| { - ClientError::new(ClientErrorKind::InvalidRequest, error.to_string()) - })?; + request.validate().map_err(invalid_request)?; + let encoded = encode(&request)?; let id = self.allocate_request_id(); let maximum = self.limits.objects_per_session; let state = self.session_mut(session)?; @@ -782,10 +862,7 @@ impl Client { request, }, ); - Ok(PreparedRequest { - id, - request: encoded, - }) + Ok((id, encoded)) } /// Ensures that an interface descriptor has been observed without issuing @@ -800,7 +877,7 @@ impl Client { &mut self, session: &SessionId, reference: InterfaceReference, - ) -> Result, ClientError> { + ) -> Result, ClientError> { let state = self.session(session)?; if state.interfaces.contains_key(&reference) || state.interfaces.len() >= self.limits.interfaces_per_session @@ -813,22 +890,31 @@ impl Client { /// Explicitly starts or refreshes an interface descriptor observation. /// - /// Unlike [`Client::ensure_interface`], this always creates a new request and + /// Unlike [`DirectClient::ensure_interface`], this always creates a new request and /// may evict an older non-loading descriptor when the interface cache is full. pub fn prepare_interface( &mut self, session: &SessionId, reference: InterfaceReference, - ) -> Result { + ) -> Result { + let (id, request) = self.prepare_interface_with(session, reference, |request| { + Ok(LogicalRequest::FetchInterface(request.clone())) + })?; + Ok(DirectPreparedRequest { id, request }) + } + + fn prepare_interface_with( + &mut self, + session: &SessionId, + reference: InterfaceReference, + encode: impl FnOnce(&FetchInterfaceRequest) -> Result, + ) -> Result<(RequestId, T), ClientError> { self.ensure_capacity()?; let request = FetchInterfaceRequest { interface: reference, }; - let endpoint = session_endpoint(session)?; - let encoded = encode_fetch_interface_request(&endpoint, &request, self.wire_limits) - .map_err(|error| { - ClientError::new(ClientErrorKind::InvalidRequest, error.to_string()) - })?; + request.validate().map_err(invalid_request)?; + let encoded = encode(&request)?; let id = self.allocate_request_id(); let maximum = self.limits.interfaces_per_session; let state = self.session_mut(session)?; @@ -863,10 +949,7 @@ impl Client { request, }, ); - Ok(PreparedRequest { - id, - request: encoded, - }) + Ok((id, encoded)) } /// Constructs an operation from fresh object and interface observations. @@ -883,7 +966,27 @@ impl Client { interface: &InterfaceReference, operation: impl Into, arguments: BTreeMap, - ) -> Result { + ) -> Result { + let (id, request) = self.prepare_call_with( + session, + path, + interface, + operation, + arguments, + |request, _| Ok(LogicalRequest::CallOperation(request.clone())), + )?; + Ok(DirectPreparedRequest { id, request }) + } + + fn prepare_call_with( + &mut self, + session: &SessionId, + path: &str, + interface: &InterfaceReference, + operation: impl Into, + arguments: BTreeMap, + encode: impl FnOnce(&CallOperationRequest, &InterfaceDescriptor) -> Result, + ) -> Result<(RequestId, T), ClientError> { self.ensure_capacity()?; let operation = operation.into(); let context = { @@ -969,14 +1072,11 @@ impl Client { request, } }; - let endpoint = session_endpoint(session)?; - let encoded = encode_call_operation_request( - &endpoint, - &context.request, - &context.descriptor, - self.wire_limits, - ) - .map_err(|error| ClientError::new(ClientErrorKind::InvalidRequest, error.to_string()))?; + context + .request + .validate_with(&context.descriptor) + .map_err(invalid_request)?; + let encoded = encode(&context.request, &context.descriptor)?; let id = self.allocate_request_id(); self.session_mut(&context.session)? .pin_call_object(context.object_key); @@ -987,10 +1087,7 @@ impl Client { dispatched: false, }, ); - Ok(PreparedRequest { - id, - request: encoded, - }) + Ok((id, encoded)) } /// Marks an operation request as handed to the transport. @@ -1008,11 +1105,20 @@ impl Client { } } - /// Applies one HTTP response to the exact request identity that produced it. + /// Validates and applies a typed response to the exact request that produced it. + /// A mismatched interaction is invalid; invalid call responses have unknown outcomes. pub fn complete( &mut self, id: RequestId, - response: Response>, + response: LogicalResponse, + ) -> Result { + self.complete_result(id, Ok(response)) + } + + fn complete_result( + &mut self, + id: RequestId, + response: Result, ) -> Result { let pending = self.pending.remove(&id).ok_or_else(unknown_request)?; match pending { @@ -1141,13 +1247,13 @@ impl Client { id: RequestId, session: &SessionId, request: ObserveRequest, - response: Response>, + response: Result, ) -> Result { if !self.entry_request_is_current(session, &request.path, id) { return self.reject_stale(session, id); } - match decode_observe_response(&request, &response, self.wire_limits) { - Ok(DecodedResponse::Success(value)) => { + match validate_observe(&request, response) { + Ok(Ok(value)) => { let maximum = self.limits.objects_per_session; let mut candidate = self.session(session)?.clone(); if let Err(error) = @@ -1168,11 +1274,7 @@ impl Client { self.sessions.insert(session.clone(), candidate); Ok(Completion::Object(observation)) } - Ok(DecodedResponse::ProtocolFailure { - error, - status_mismatch, - }) => { - self.record_status_mismatch(session, id, status_mismatch)?; + Ok(Err(error)) => { let observation_state = if error.code == ProtocolErrorCode::NotFound { ObservationState::Stale } else { @@ -1205,13 +1307,13 @@ impl Client { id: RequestId, session: &SessionId, request: FetchInterfaceRequest, - response: Response>, + response: Result, ) -> Result { if !self.interface_request_is_current(session, &request.interface, id) { return self.reject_stale(session, id); } - match decode_fetch_interface_response(&request, &response, self.wire_limits) { - Ok(DecodedResponse::Success(value)) => { + match validate_interface(&request, response) { + Ok(Ok(value)) => { let state = self.session_mut(session)?; // Newer scope evidence (Observe or another Interface name) can // supersede even the first fetch for this pair. @@ -1290,11 +1392,7 @@ impl Client { record.touched = touched; Ok(Completion::Interface(observation)) } - Ok(DecodedResponse::ProtocolFailure { - error, - status_mismatch, - }) => { - self.record_status_mismatch(session, id, status_mismatch)?; + Ok(Err(error)) => { self.session_mut(session)? .interfaces .get_mut(&request.interface) @@ -1319,15 +1417,10 @@ impl Client { &mut self, id: RequestId, context: CallContext, - response: Response>, + response: Result, ) -> Result { - match decode_call_operation_response( - &context.request, - &context.descriptor, - &response, - self.wire_limits, - ) { - Ok(DecodedResponse::Success(value)) => { + match validate_call(&context, response) { + Ok(Ok(value)) => { let session_id = context.session.clone(); let maximum = self.limits.history; let state = self.session_mut(&session_id)?; @@ -1347,12 +1440,8 @@ impl Client { state.push_call(record.clone(), maximum); Ok(Completion::Call(Box::new(record))) } - Ok(DecodedResponse::ProtocolFailure { - error, - status_mismatch, - }) => { + Ok(Err(error)) => { let session_id = context.session.clone(); - self.record_status_mismatch(&session_id, id, status_mismatch)?; if error.code == ProtocolErrorCode::OperationOutcomeUnknown { let maximum = self.limits.history; let state = self.session_mut(&session_id)?; @@ -1410,7 +1499,7 @@ impl Client { Ok(Completion::Call(Box::new(record))) } Err(error) => { - let failure = classify_response_error(error); + let failure = error; self.finish_unknown_call(id, context, OutcomeUnknownReason::ResponseDecode, failure) } } @@ -1452,9 +1541,8 @@ impl Client { id: RequestId, session: &SessionId, subject: RetrievalSubject, - error: ClientResponseError, + error: ClientError, ) -> Result { - let error = classify_response_error(error); match subject { RetrievalSubject::Observe(request) => { set_observe_subject_state( @@ -1505,29 +1593,6 @@ impl Client { Ok(()) } - fn record_status_mismatch( - &mut self, - session: &SessionId, - id: RequestId, - mismatch: Option, - ) -> Result<(), ClientError> { - if let Some(mismatch) = mismatch { - let maximum = self.limits.history; - self.session_mut(session)?.push_diagnostic( - Diagnostic { - request_id: id, - kind: DiagnosticKind::HttpStatusMismatch, - detail: format!( - "received HTTP {}; canonical status is {}", - mismatch.actual, mismatch.expected - ), - }, - maximum, - ); - } - Ok(()) - } - fn push_failure_diagnostic( &mut self, session: &SessionId, @@ -1682,30 +1747,10 @@ fn observation_coverage(state: &Session, path: &str, depth: u32) -> ObservationC } } -fn session_endpoint(session: &SessionId) -> Result { - Endpoint::parse(&session.endpoint) - .map_err(|error| ClientError::new(ClientErrorKind::InvalidRequest, error.to_string())) -} - fn unknown_request() -> ClientError { ClientError::new(ClientErrorKind::InvalidRequest, "unknown request identity") } -fn classify_response_error(error: ClientResponseError) -> ClientError { - match error { - ClientResponseError::InvalidResponse { .. } => { - ClientError::new(ClientErrorKind::InvalidResponse, error.to_string()) - } - ClientResponseError::UnsupportedDescriptorFormat { .. } => ClientError::new( - ClientErrorKind::UnsupportedDescriptorFormat, - error.to_string(), - ), - ClientResponseError::TransportBinding(_) => { - ClientError::new(ClientErrorKind::Transport, error.to_string()) - } - } -} - fn set_observe_subject_state( state: &mut Session, request: &ObserveRequest, diff --git a/crates/wip-client/src/validation.rs b/crates/wip-client/src/validation.rs new file mode 100644 index 0000000..522dd15 --- /dev/null +++ b/crates/wip-client/src/validation.rs @@ -0,0 +1,80 @@ +use super::*; + +fn invalid_response(error: impl Display) -> ClientError { + ClientError::new(ClientErrorKind::InvalidResponse, error.to_string()) +} + +pub(super) fn invalid_request(error: impl Display) -> ClientError { + ClientError::new(ClientErrorKind::InvalidRequest, error.to_string()) +} + +fn validate_failure( + result: Result, + interaction: ProtocolInteraction, + validate: impl FnOnce(&T) -> Result<(), ClientError>, +) -> Result, ClientError> { + match &result { + Ok(value) => validate(value)?, + Err(error) if !error.code.is_allowed_for(interaction) => { + return Err(invalid_response( + "Protocol error is forbidden for this interaction", + )); + } + Err(_) => {} + } + Ok(result) +} + +pub(super) fn validate_observe( + request: &ObserveRequest, + response: Result, +) -> Result, ClientError> { + let LogicalResponse::Observe(result) = response? else { + return Err(invalid_response( + "response interaction does not match Observe request", + )); + }; + validate_failure(result, ProtocolInteraction::Observe, |value| { + request.validate_response(value).map_err(invalid_response) + }) +} + +pub(super) fn validate_interface( + request: &FetchInterfaceRequest, + response: Result, +) -> Result, ClientError> { + let LogicalResponse::FetchInterface(result) = response? else { + return Err(invalid_response( + "response interaction does not match FetchInterface request", + )); + }; + validate_failure(result, ProtocolInteraction::FetchInterface, |value| { + value.validate_for(request).map_err(invalid_response)?; + if value.descriptor.format != INTERFACE_FORMAT_V1 { + return Err(ClientError::new( + ClientErrorKind::UnsupportedDescriptorFormat, + format!( + "unsupported interface descriptor format `{}`", + value.descriptor.format + ), + )); + } + Ok(()) + }) +} + +pub(super) fn validate_call( + context: &CallContext, + response: Result, +) -> Result, ClientError> { + let LogicalResponse::CallOperation(result) = response? else { + return Err(invalid_response( + "response interaction does not match CallOperation request", + )); + }; + validate_failure(result, ProtocolInteraction::CallOperation, |value| { + value + .validate_for(&context.descriptor, &context.request.operation) + .map_err(invalid_response) + }) +} diff --git a/crates/wip-client/tests/direct.rs b/crates/wip-client/tests/direct.rs new file mode 100644 index 0000000..0a7b3b6 --- /dev/null +++ b/crates/wip-client/tests/direct.rs @@ -0,0 +1,614 @@ +//! Public API tests: no HTTP types, codec, endpoint parser, or JSON dependency. +use std::collections::BTreeMap; + +use wip_client::direct::{Client, PreparedRequest, Request, Response}; +use wip_client::{ + CallOutcome, ClientErrorKind, ClientLimits, Completion, DispatchedTransportFailure, + ObservationState, OutcomeUnknownReason, SecurityContext, SessionId, +}; +use wip_protocol::{ + CallOperationResponse, FetchInterfaceResponse, INTERFACE_FORMAT_V1, InterfaceDescriptor, + InterfaceReference, Object, ObjectObservation, OperationDeclaration, ParameterDeclaration, + ProtocolError, ProtocolErrorCode, ReturnDeclaration, TypeExpr, Value, +}; + +fn setup() -> (Client, SessionId) { + let mut client = Client::new(ClientLimits::new(4, 16, 8, 8, 4).unwrap()); + let session = client + .open_session("in-process world", SecurityContext::new("alice")) + .unwrap(); + (client, session) +} + +fn reference() -> InterfaceReference { + InterfaceReference { + scope: "/scope".into(), + name: "math".into(), + } +} + +fn object(name: &str, identity: Option<&str>) -> ObjectObservation { + ObjectObservation { + object: Object { + name: name.into(), + description: None, + interfaces: vec![reference()], + r#ref: identity.map(str::to_owned), + validator: None, + }, + children: None, + } +} + +fn descriptor() -> InterfaceDescriptor { + InterfaceDescriptor { + format: INTERFACE_FORMAT_V1.into(), + documentation: None, + types: vec![], + operations: vec![OperationDeclaration { + name: "double".into(), + documentation: None, + parameters: vec![ParameterDeclaration { + name: "n".into(), + documentation: None, + required: true, + r#type: TypeExpr::Integer, + }], + returns: ReturnDeclaration { + documentation: None, + r#type: TypeExpr::Integer, + }, + }], + } +} + +fn interface(scope_ref: Option<&str>) -> FetchInterfaceResponse { + FetchInterfaceResponse { + interface: reference(), + scope_ref: scope_ref.map(str::to_owned), + validator: None, + descriptor: descriptor(), + } +} + +fn failure(code: ProtocolErrorCode) -> ProtocolError { + ProtocolError { + code, + message: "test failure".into(), + } +} + +fn observe(client: &mut Client, session: &SessionId, path: &str, value: ObjectObservation) { + let request = client.refresh_object(session, path).unwrap(); + assert!( + matches!(&request.request, Request::Observe(value) if value.path == path && value.depth == 0) + ); + client + .complete(request.id, Response::Observe(Ok(value))) + .unwrap(); +} + +fn ready(client: &mut Client, session: &SessionId) { + observe(client, session, "/scope/item", object("item", None)); + let request = client.prepare_interface(session, reference()).unwrap(); + assert!( + matches!(request.request, Request::FetchInterface(value) if value.interface == reference()) + ); + client + .complete(request.id, Response::FetchInterface(Ok(interface(None)))) + .unwrap(); +} + +fn call(client: &mut Client, session: &SessionId) -> PreparedRequest { + client + .prepare_call( + session, + "/scope/item", + &reference(), + "double", + BTreeMap::from([("n".into(), Value::Integer(3))]), + ) + .unwrap() +} + +#[test] +fn typed_interactions_succeed_with_optional_identity_and_validators_absent() { + let (mut client, session) = setup(); + ready(&mut client, &session); + assert!( + client + .ensure_observed(&session, "/scope/item", 0) + .unwrap() + .is_none() + ); + assert!( + client + .ensure_interface(&session, reference()) + .unwrap() + .is_none() + ); + let request = call(&mut client, &session); + let Request::CallOperation(logical) = request.request else { + panic!("wrong request") + }; + assert_eq!(logical.target.path, "/scope/item"); + assert_eq!(logical.target.validator, None); + assert_eq!(logical.interface.scope_ref, None); + assert_eq!(logical.interface.validator, None); + assert_eq!(logical.arguments["n"], Value::Integer(3)); + client.mark_dispatched(request.id).unwrap(); + let result = CallOperationResponse { + result: Value::Integer(6), + validator: Some(vec![2]), + }; + let completion = client + .complete(request.id, Response::CallOperation(Ok(result.clone()))) + .unwrap(); + assert!( + matches!(completion, Completion::Call(record) if record.outcome == CallOutcome::Success(result)) + ); + assert_eq!( + client.object(&session, "/scope/item").unwrap().validator, + Some(vec![2]) + ); + assert_eq!(client.in_flight_len(), 0); + assert_eq!( + client + .complete( + request.id, + Response::CallOperation(Ok(CallOperationResponse { + result: Value::Integer(6), + validator: None, + })) + ) + .unwrap_err() + .kind, + ClientErrorKind::InvalidRequest + ); +} + +#[test] +fn worldspace_and_security_identities_partition_observations_without_urls() { + let (mut client, session) = setup(); + ready(&mut client, &session); + let other_world = client + .open_session("another world", SecurityContext::new("alice")) + .unwrap(); + let other_security = client + .open_session("in-process world", SecurityContext::new("bob")) + .unwrap(); + for other in [&other_world, &other_security] { + assert!(client.object(other, "/scope/item").is_none()); + assert!(client.interface(other, &reference()).is_none()); + } + assert_eq!( + client + .open_session("in-process world", SecurityContext::new("alice")) + .unwrap(), + session + ); +} + +#[test] +fn typed_observations_validate_name_depth_duplicate_children_and_interface_scope() { + let (mut client, session) = setup(); + let mut wrong_scope = object("item", None); + wrong_scope.object.interfaces[0].scope = "/elsewhere".into(); + let mut unexpected_children = object("item", None); + unexpected_children.children = Some(vec![]); + let mut duplicate = object("item", None); + duplicate.children = Some(vec![object("child", None), object("child", None)]); + for (depth, response) in [ + (0, object("wrong-name", None)), + (1, object("item", None)), + (0, unexpected_children), + (0, wrong_scope), + (1, duplicate), + ] { + let request = client + .refresh_observed(&session, "/scope/item", depth) + .unwrap(); + let error = client + .complete(request.id, Response::Observe(Ok(response))) + .unwrap_err(); + assert_eq!(error.kind, ClientErrorKind::InvalidResponse); + assert!(matches!( + client.object(&session, "/scope/item").unwrap().state, + ObservationState::Error(_) + )); + } +} + +#[test] +fn typed_descriptor_correspondence_structure_and_format_are_checked() { + let (mut client, session) = setup(); + let mut mismatched = interface(None); + mismatched.interface.name = "other".into(); + let mut malformed = interface(None); + malformed + .descriptor + .operations + .push(malformed.descriptor.operations[0].clone()); + let mut unsupported = interface(None); + unsupported.descriptor.format = "future".into(); + for (value, kind) in [ + (mismatched, ClientErrorKind::InvalidResponse), + (malformed, ClientErrorKind::InvalidResponse), + (unsupported, ClientErrorKind::UnsupportedDescriptorFormat), + ] { + let request = client.prepare_interface(&session, reference()).unwrap(); + assert_eq!( + client + .complete(request.id, Response::FetchInterface(Ok(value))) + .unwrap_err() + .kind, + kind + ); + } +} + +#[test] +fn typed_arguments_and_results_use_the_fixed_descriptor() { + let (mut client, session) = setup(); + ready(&mut client, &session); + let invalid = client + .prepare_call( + &session, + "/scope/item", + &reference(), + "double", + BTreeMap::from([("n".into(), Value::String("not integer".into()))]), + ) + .unwrap_err(); + assert_eq!(invalid.kind, ClientErrorKind::InvalidRequest); + assert_eq!(client.in_flight_len(), 0); + let request = call(&mut client, &session); + client.mark_dispatched(request.id).unwrap(); + let completion = client + .complete( + request.id, + Response::CallOperation(Ok(CallOperationResponse { + result: Value::String("not integer".into()), + validator: None, + })), + ) + .unwrap(); + assert!( + matches!(completion, Completion::Call(record) if matches!(record.outcome, + CallOutcome::Unknown { reason: OutcomeUnknownReason::ResponseDecode, .. })) + ); + assert_eq!( + client.object(&session, "/scope/item").unwrap().state, + ObservationState::Stale + ); + assert!( + client + .ensure_observed(&session, "/scope/item", 0) + .unwrap() + .is_none() + ); + assert_eq!(client.in_flight_len(), 0); +} + +#[test] +fn typed_protocol_failures_validate_interaction_and_apply_recovery() { + let (mut client, session) = setup(); + let request = client.refresh_object(&session, "/scope/item").unwrap(); + assert!(matches!( + client + .complete( + request.id, + Response::Observe(Err(failure(ProtocolErrorCode::NotFound))) + ) + .unwrap(), + Completion::ProtocolFailure(_) + )); + let request = client.prepare_interface(&session, reference()).unwrap(); + assert!(matches!( + client + .complete( + request.id, + Response::FetchInterface(Err(failure(ProtocolErrorCode::InterfaceNotFound))) + ) + .unwrap(), + Completion::ProtocolFailure(_) + )); + let request = client.refresh_object(&session, "/scope/item").unwrap(); + assert_eq!( + client + .complete( + request.id, + Response::Observe(Err(failure(ProtocolErrorCode::OperationOutcomeUnknown))) + ) + .unwrap_err() + .kind, + ClientErrorKind::InvalidResponse + ); + let request = client.prepare_interface(&session, reference()).unwrap(); + assert_eq!( + client + .complete( + request.id, + Response::FetchInterface(Err(failure(ProtocolErrorCode::ValidatorMismatch))) + ) + .unwrap_err() + .kind, + ClientErrorKind::InvalidResponse + ); + ready(&mut client, &session); + let request = call(&mut client, &session); + let completion = client + .complete( + request.id, + Response::CallOperation(Err(failure(ProtocolErrorCode::InterfaceMismatch))), + ) + .unwrap(); + assert!( + matches!(completion, Completion::Call(record) if matches!(record.outcome, CallOutcome::ProtocolFailure(_))) + ); + assert_eq!( + client.object(&session, "/scope/item").unwrap().state, + ObservationState::Stale + ); + assert_eq!( + client.interface(&session, &reference()).unwrap().state, + ObservationState::Stale + ); +} + +#[test] +fn mismatched_interactions_are_not_trusted_and_calls_remain_unknown() { + let (mut client, session) = setup(); + let request = client.refresh_object(&session, "/scope/item").unwrap(); + assert_eq!( + client + .complete(request.id, Response::FetchInterface(Ok(interface(None)))) + .unwrap_err() + .kind, + ClientErrorKind::InvalidResponse + ); + let request = client.prepare_interface(&session, reference()).unwrap(); + assert_eq!( + client + .complete(request.id, Response::Observe(Ok(object("item", None)))) + .unwrap_err() + .kind, + ClientErrorKind::InvalidResponse + ); + ready(&mut client, &session); + let request = call(&mut client, &session); + let completion = client + .complete( + request.id, + Response::Observe(Err(failure(ProtocolErrorCode::NotFound))), + ) + .unwrap(); + assert!( + matches!(completion, Completion::Call(record) if matches!(record.outcome, CallOutcome::Unknown { .. })) + ); +} + +#[test] +fn typed_stale_responses_cannot_replace_new_observations() { + let (mut client, session) = setup(); + let old = client.refresh_object(&session, "/scope/item").unwrap(); + assert!( + client + .ensure_observed(&session, "/scope/item", 0) + .unwrap() + .is_none() + ); + let new = client.refresh_object(&session, "/scope/item").unwrap(); + client + .complete(new.id, Response::Observe(Ok(object("item", Some("new"))))) + .unwrap(); + assert_eq!( + client + .complete(old.id, Response::Observe(Ok(object("item", Some("old"))))) + .unwrap(), + Completion::StaleResponseRejected(old.id) + ); + assert_eq!( + client + .object(&session, "/scope/item") + .unwrap() + .object + .unwrap() + .r#ref + .as_deref(), + Some("new") + ); +} + +#[test] +fn scope_replacement_retires_descriptors_and_pending_fetch_but_not_call_snapshot() { + let (mut client, session) = setup(); + observe( + &mut client, + &session, + "/scope", + object("scope", Some("old")), + ); + ready(&mut client, &session); + let call = call(&mut client, &session); + let fetch = client.prepare_interface(&session, reference()).unwrap(); + observe( + &mut client, + &session, + "/scope", + object("scope", Some("new")), + ); + assert!( + client + .interface(&session, &reference()) + .unwrap() + .descriptor + .is_none() + ); + assert_eq!( + client + .complete( + fetch.id, + Response::FetchInterface(Ok(interface(Some("old")))) + ) + .unwrap(), + Completion::StaleResponseRejected(fetch.id) + ); + assert_eq!( + client.pending_call_context(call.id).unwrap().descriptor(), + &descriptor() + ); + assert_eq!( + client + .pending_call_context(call.id) + .unwrap() + .request() + .interface + .scope_ref, + None + ); + // A newer descriptor has a different return type; the prepared call still uses Integer. + let fetch = client.prepare_interface(&session, reference()).unwrap(); + let mut next = interface(Some("new")); + next.descriptor.operations[0].returns.r#type = TypeExpr::String; + client + .complete(fetch.id, Response::FetchInterface(Ok(next))) + .unwrap(); + let completion = client + .complete( + call.id, + Response::CallOperation(Ok(CallOperationResponse { + result: Value::Integer(6), + validator: None, + })), + ) + .unwrap(); + assert!( + matches!(completion, Completion::Call(record) if matches!(record.outcome, CallOutcome::Success(_))) + ); +} + +#[test] +fn scope_not_found_retires_interfaces_and_late_fetch_cannot_restore_them() { + let (mut client, session) = setup(); + ready(&mut client, &session); + let old = client.prepare_interface(&session, reference()).unwrap(); + let missing = client.refresh_object(&session, "/scope").unwrap(); + client + .complete( + missing.id, + Response::Observe(Err(failure(ProtocolErrorCode::NotFound))), + ) + .unwrap(); + assert!(client.interface(&session, &reference()).is_none()); + assert_eq!( + client + .complete(old.id, Response::FetchInterface(Ok(interface(None)))) + .unwrap(), + Completion::StaleResponseRejected(old.id) + ); +} + +#[test] +fn failures_before_and_after_dispatch_never_retry_operations() { + let (mut client, session) = setup(); + ready(&mut client, &session); + let before = call(&mut client, &session); + assert!( + matches!(client.fail_transport(before.id, "not handed to Host").unwrap(), + Completion::Call(record) if matches!(record.outcome, CallOutcome::NotDispatched(_))) + ); + assert_eq!( + client.object(&session, "/scope/item").unwrap().state, + ObservationState::Fresh + ); + let after = call(&mut client, &session); + client.mark_dispatched(after.id).unwrap(); + assert!( + matches!(client.fail_dispatched_call(after.id, DispatchedTransportFailure::Timeout, "deadline").unwrap(), + Completion::Call(record) if matches!(record.outcome, CallOutcome::Unknown { reason: OutcomeUnknownReason::Timeout, .. })) + ); + assert_eq!(client.in_flight_len(), 0); + assert!( + client + .ensure_observed(&session, "/scope/item", 0) + .unwrap() + .is_none() + ); + ready(&mut client, &session); + let unknown = call(&mut client, &session); + assert!( + matches!(client.complete(unknown.id, Response::CallOperation(Err(failure(ProtocolErrorCode::OperationOutcomeUnknown)))).unwrap(), + Completion::Call(record) if matches!(record.outcome, CallOutcome::Unknown { reason: OutcomeUnknownReason::Protocol, .. })) + ); + assert_eq!(client.in_flight_len(), 0); + assert_eq!(client.call_history(&session).unwrap().len(), 3); +} + +#[test] +fn direct_limits_bound_sessions_inflight_cache_and_history() { + let mut client = Client::new(ClientLimits::new(1, 1, 1, 1, 1).unwrap()); + let session = client + .open_session("world", SecurityContext::new("alice")) + .unwrap(); + assert_eq!( + client + .open_session("other", SecurityContext::new("alice")) + .unwrap_err() + .kind, + ClientErrorKind::Capacity + ); + let first = client.refresh_object(&session, "/scope/item").unwrap(); + assert!( + client + .ensure_observed(&session, "/scope/other", 0) + .unwrap() + .is_none() + ); + assert_eq!( + client + .refresh_object(&session, "/scope/other") + .unwrap_err() + .kind, + ClientErrorKind::Capacity + ); + client + .complete(first.id, Response::Observe(Ok(object("item", None)))) + .unwrap(); + observe(&mut client, &session, "/scope/other", object("other", None)); + assert_eq!(client.known_space(&session).unwrap().len(), 1); + for _ in 0..3 { + let request = client.refresh_object(&session, "/scope/other").unwrap(); + client + .complete(request.id, Response::Observe(Ok(object("wrong", None)))) + .unwrap_err(); + } + assert_eq!(client.diagnostics(&session).unwrap().len(), 1); +} + +#[test] +fn invalid_direct_requests_do_not_acquire_pending_state() { + let (mut client, session) = setup(); + assert_eq!( + client + .refresh_object(&session, "not/absolute") + .unwrap_err() + .kind, + ClientErrorKind::InvalidRequest + ); + assert_eq!( + client + .prepare_interface( + &session, + InterfaceReference { + scope: "/".into(), + name: "".into() + } + ) + .unwrap_err() + .kind, + ClientErrorKind::InvalidRequest + ); + assert_eq!(client.in_flight_len(), 0); + assert!(client.known_space(&session).unwrap().is_empty()); +} diff --git a/crates/wip-client/tests/parity.rs b/crates/wip-client/tests/parity.rs new file mode 100644 index 0000000..f397310 --- /dev/null +++ b/crates/wip-client/tests/parity.rs @@ -0,0 +1,289 @@ +#![cfg(feature = "http")] +//! The same stateful scenario runs through direct DTOs and the HTTP facade. +use std::collections::BTreeMap; +use wip_client::direct::{Client as DirectClient, Request, Response}; +use wip_client::{ + CallOutcome, Client, ClientLimits, Completion, InterfaceObservation, ObjectObservation, + RequestId, SecurityContext, SessionId, +}; +use wip_http::{ + Limits, decode_call_operation_request, decode_fetch_interface_request, decode_observe_request, + encode_call_operation_response, encode_fetch_interface_response, encode_observe_response, + encode_protocol_error_response, +}; +use wip_protocol::{ + CallOperationResponse, FetchInterfaceResponse, INTERFACE_FORMAT_V1, InterfaceDescriptor, + InterfaceReference, Object, ObjectObservation as ProtocolObservation, OperationDeclaration, + ProtocolError, ProtocolErrorCode, ProtocolInteraction, ReturnDeclaration, TypeExpr, Value, +}; + +fn limits() -> Limits { + Limits::new(64 * 1024, 64 * 1024, 64).unwrap() +} +fn reference() -> InterfaceReference { + InterfaceReference { + scope: "/scope".into(), + name: "math".into(), + } +} +fn descriptor() -> InterfaceDescriptor { + InterfaceDescriptor { + format: INTERFACE_FORMAT_V1.into(), + documentation: None, + types: vec![], + operations: vec![OperationDeclaration { + name: "run".into(), + documentation: None, + parameters: vec![], + returns: ReturnDeclaration { + documentation: None, + r#type: TypeExpr::Integer, + }, + }], + } +} +fn observation(name: &str, identity: &str) -> ProtocolObservation { + ProtocolObservation { + object: Object { + name: name.into(), + description: None, + interfaces: vec![reference()], + r#ref: Some(identity.into()), + validator: None, + }, + children: None, + } +} +fn interface(identity: &str) -> FetchInterfaceResponse { + FetchInterfaceResponse { + interface: reference(), + scope_ref: Some(identity.into()), + validator: None, + descriptor: descriptor(), + } +} + +enum Channel { + Direct(DirectClient), + Http(Client), +} +impl Channel { + fn core(&self) -> &DirectClient { + match self { + Self::Direct(client) => client, + Self::Http(client) => client, + } + } + fn observe(&mut self, session: &SessionId, path: &str) -> (RequestId, Request) { + match self { + Self::Direct(client) => { + let p = client.refresh_object(session, path).unwrap(); + (p.id, p.request) + } + Self::Http(client) => { + let p = client.refresh_object(session, path).unwrap(); + ( + p.id, + Request::Observe(decode_observe_request(p.request.body(), limits()).unwrap()), + ) + } + } + } + fn fetch(&mut self, session: &SessionId) -> (RequestId, Request) { + match self { + Self::Direct(client) => { + let p = client.prepare_interface(session, reference()).unwrap(); + (p.id, p.request) + } + Self::Http(client) => { + let p = client.prepare_interface(session, reference()).unwrap(); + ( + p.id, + Request::FetchInterface( + decode_fetch_interface_request(p.request.body(), limits()).unwrap(), + ), + ) + } + } + } + fn call(&mut self, session: &SessionId) -> (RequestId, Request) { + match self { + Self::Direct(client) => { + let p = client + .prepare_call(session, "/scope/item", &reference(), "run", BTreeMap::new()) + .unwrap(); + client.mark_dispatched(p.id).unwrap(); + (p.id, p.request) + } + Self::Http(client) => { + let p = client + .prepare_call(session, "/scope/item", &reference(), "run", BTreeMap::new()) + .unwrap(); + client.mark_dispatched(p.id).unwrap(); + ( + p.id, + Request::CallOperation( + decode_call_operation_request(p.request.body(), &descriptor(), limits()) + .unwrap(), + ), + ) + } + } + } + fn complete(&mut self, pending: (RequestId, Request), result: Response) -> Completion { + let (id, request) = pending; + match self { + Self::Direct(client) => client.complete(id, result).unwrap(), + Self::Http(client) => { + let wire = match (request, result) { + (Request::Observe(request), Response::Observe(Ok(value))) => { + encode_observe_response(&request, &value, limits()) + } + (Request::FetchInterface(request), Response::FetchInterface(Ok(value))) => { + encode_fetch_interface_response(&request, &value, limits()) + } + (Request::CallOperation(request), Response::CallOperation(Ok(value))) => { + encode_call_operation_response(&request, &descriptor(), &value, limits()) + } + (_, Response::Observe(Err(error))) => encode_protocol_error_response( + ProtocolInteraction::Observe, + &error, + limits(), + ), + (_, Response::FetchInterface(Err(error))) => encode_protocol_error_response( + ProtocolInteraction::FetchInterface, + &error, + limits(), + ), + (_, Response::CallOperation(Err(error))) => encode_protocol_error_response( + ProtocolInteraction::CallOperation, + &error, + limits(), + ), + _ => panic!("test interaction mismatch"), + } + .unwrap(); + client.complete(id, wire).unwrap() + } + } + } +} + +#[derive(Debug, PartialEq)] +struct Snapshot { + objects: Vec, + interface: Option, + outcomes: Vec, + requests: Vec, +} +fn scenario(http: bool) -> Snapshot { + let state_limits = ClientLimits::new(4, 16, 8, 8, 8).unwrap(); + let (mut channel, session) = if http { + let mut client = Client::new(state_limits, limits()); + let session = client + .open_session("https://example.test/wip", SecurityContext::new("alice")) + .unwrap(); + (Channel::Http(client), session) + } else { + let mut client = DirectClient::new(state_limits); + let session = client + .open_session("local world", SecurityContext::new("alice")) + .unwrap(); + (Channel::Direct(client), session) + }; + let mut requests = vec![]; + for (path, name, identity) in [ + ("/scope", "scope", "scope-v1"), + ("/scope/item", "item", "item-v1"), + ] { + let p = channel.observe(&session, path); + requests.push(p.1.clone()); + channel.complete(p, Response::Observe(Ok(observation(name, identity)))); + } + let p = channel.fetch(&session); + requests.push(p.1.clone()); + channel.complete(p, Response::FetchInterface(Ok(interface("scope-v1")))); + let p = channel.call(&session); + requests.push(p.1.clone()); + channel.complete( + p, + Response::CallOperation(Ok(CallOperationResponse { + result: Value::Integer(42), + validator: Some(vec![2]), + })), + ); + let p = channel.call(&session); + requests.push(p.1.clone()); + channel.complete( + p, + Response::CallOperation(Err(ProtocolError { + code: ProtocolErrorCode::InterfaceMismatch, + message: "changed".into(), + })), + ); + let p = channel.observe(&session, "/scope/item"); + channel.complete(p, Response::Observe(Ok(observation("item", "item-v1")))); + let p = channel.fetch(&session); + channel.complete(p, Response::FetchInterface(Ok(interface("scope-v1")))); + let old = channel.fetch(&session); + let p = channel.observe(&session, "/scope"); + channel.complete(p, Response::Observe(Ok(observation("scope", "scope-v2")))); + let old_id = old.0; + assert_eq!( + channel.complete(old, Response::FetchInterface(Ok(interface("scope-v1")))), + Completion::StaleResponseRejected(old_id) + ); + let p = channel.fetch(&session); + channel.complete(p, Response::FetchInterface(Ok(interface("scope-v2")))); + let p = channel.call(&session); + requests.push(p.1.clone()); + channel.complete( + p, + Response::CallOperation(Err(ProtocolError { + code: ProtocolErrorCode::OperationOutcomeUnknown, + message: "unknown".into(), + })), + ); + assert_eq!(channel.core().in_flight_len(), 0); + Snapshot { + objects: channel.core().known_space(&session).unwrap(), + interface: channel.core().interface(&session, &reference()).cloned(), + outcomes: channel + .core() + .call_history(&session) + .unwrap() + .iter() + .map(|record| record.outcome.clone()) + .collect(), + requests, + } +} + +#[test] +fn direct_and_http_agree_on_requests_state_calls_scope_invalidation_and_uncertainty() { + assert_eq!(scenario(false), scenario(true)); +} + +#[test] +fn http_endpoint_and_identical_direct_identity_do_not_share_cache() { + let mut client = Client::new(ClientLimits::new(4, 16, 8, 8, 8).unwrap(), limits()); + let http = client + .open_session("https://example.test/wip", SecurityContext::new("alice")) + .unwrap(); + let direct = DirectClient::open_session( + &mut client, + "https://example.test/wip/", + SecurityContext::new("alice"), + ) + .unwrap(); + assert_ne!(http, direct); + let request = DirectClient::refresh_object(&mut client, &direct, "/scope").unwrap(); + DirectClient::complete( + &mut client, + request.id, + Response::Observe(Ok(observation("scope", "local"))), + ) + .unwrap(); + assert!(client.object(&http, "/scope").is_none()); + assert!(client.refresh_object(&direct, "/scope").is_err()); +} diff --git a/crates/wip-client/tests/wire_limits.rs b/crates/wip-client/tests/wire_limits.rs new file mode 100644 index 0000000..7762f2f --- /dev/null +++ b/crates/wip-client/tests/wire_limits.rs @@ -0,0 +1,128 @@ +#![cfg(feature = "http")] +//! Wire budgets are adapter-only and failed encoding must not acquire state. +use std::collections::BTreeMap; +use wip_client::{Client, ClientErrorKind, ClientLimits, ObservationState, SecurityContext}; +use wip_http::{Limits, encode_fetch_interface_response, encode_observe_response}; +use wip_protocol::{ + FetchInterfaceRequest, FetchInterfaceResponse, INTERFACE_FORMAT_V1, InterfaceDescriptor, + InterfaceReference, Object, ObjectObservation, ObserveRequest, OperationDeclaration, + ReturnDeclaration, TypeExpr, +}; + +fn state_limits() -> ClientLimits { + ClientLimits::new(2, 8, 8, 8, 8).unwrap() +} + +#[test] +fn request_encoding_failure_leaves_retrieval_cache_and_inflight_unchanged() { + let mut client = Client::new(state_limits(), Limits::new(1, 4096, 32).unwrap()); + let session = client + .open_session("https://example.test/wip", SecurityContext::new("alice")) + .unwrap(); + assert_eq!( + client.refresh_object(&session, "/item").unwrap_err().kind, + ClientErrorKind::InvalidRequest + ); + let reference = InterfaceReference { + scope: "/".into(), + name: "test".into(), + }; + assert_eq!( + client + .prepare_interface(&session, reference.clone()) + .unwrap_err() + .kind, + ClientErrorKind::InvalidRequest + ); + assert!(client.object(&session, "/item").is_none()); + assert!(client.interface(&session, &reference).is_none()); + assert_eq!(client.in_flight_len(), 0); +} + +#[test] +fn call_encoding_failure_keeps_fresh_observations_and_no_pending_call() { + let limits = Limits::new(80, 4096, 32).unwrap(); + let mut client = Client::new(state_limits(), limits); + let session = client + .open_session("https://example.test/wip", SecurityContext::new("alice")) + .unwrap(); + let reference = InterfaceReference { + scope: "/".into(), + name: "test".into(), + }; + let observe = client.refresh_object(&session, "/item").unwrap(); + client + .complete( + observe.id, + encode_observe_response( + &ObserveRequest { + path: "/item".into(), + depth: 0, + }, + &ObjectObservation { + object: Object { + name: "item".into(), + description: None, + interfaces: vec![reference.clone()], + r#ref: None, + validator: None, + }, + children: None, + }, + limits, + ) + .unwrap(), + ) + .unwrap(); + let fetch = client + .prepare_interface(&session, reference.clone()) + .unwrap(); + client + .complete( + fetch.id, + encode_fetch_interface_response( + &FetchInterfaceRequest { + interface: reference.clone(), + }, + &FetchInterfaceResponse { + interface: reference.clone(), + scope_ref: None, + validator: None, + descriptor: InterfaceDescriptor { + format: INTERFACE_FORMAT_V1.into(), + documentation: None, + types: vec![], + operations: vec![OperationDeclaration { + name: "run".into(), + documentation: None, + parameters: vec![], + returns: ReturnDeclaration { + documentation: None, + r#type: TypeExpr::Integer, + }, + }], + }, + }, + limits, + ) + .unwrap(), + ) + .unwrap(); + assert_eq!( + client + .prepare_call(&session, "/item", &reference, "run", BTreeMap::new()) + .unwrap_err() + .kind, + ClientErrorKind::InvalidRequest + ); + assert_eq!( + client.object(&session, "/item").unwrap().state, + ObservationState::Fresh + ); + assert_eq!( + client.interface(&session, &reference).unwrap().state, + ObservationState::Fresh + ); + assert!(client.call_history(&session).unwrap().is_empty()); + assert_eq!(client.in_flight_len(), 0); +}