diff --git a/Cargo.lock b/Cargo.lock index b9a0d18..fbac6ce 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -456,6 +456,14 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "23b97319f7b8343df12cc98938e5c3eb436064524c8d2b4e30a1d3a36eecdf81" +[[package]] +name = "wip-client" +version = "0.1.0" +dependencies = [ + "wip-https", + "wip-protocol", +] + [[package]] name = "wip-host" version = "0.1.0" diff --git a/README.md b/README.md index 8b83f4b..8710690 100644 --- a/README.md +++ b/README.md @@ -4,10 +4,10 @@ Rust building blocks for the Web Interface Protocol (WIP). The repository is a resolver-3 virtual Cargo workspace whose packages all live under `crates/*`. It provides a transport-independent protocol model, the -standard framework-independent WIP over HTTPS v1 binding, a Rust -interface-authoring facade, a request-scoped host runtime, and a private -public-API conformance consumer. The repository root is not itself a package. - +standard framework-independent WIP over HTTPS v1 binding, a stateful +presentation-independent client runtime, a Rust interface-authoring facade, a +request-scoped host runtime, and a private public-API conformance consumer. The +repository root is not itself a package. ## Crate graph and responsibilities ```text @@ -15,6 +15,9 @@ wip-public-api-example (workspace-only executable, integration tests, doctest) ├── wip-protocol logical declarations, values, calls, and validation ├── wip-https standard HTTPS v1 routes, JSON codecs, envelopes, and batching │ └── wip-protocol +├── wip-client stateful Known Space, lazy tree, observation, and call runtime +│ ├── wip-https +│ └── wip-protocol ├── wip-interface application DTO codecs and interface authoring facade │ ├── wip-protocol │ └── wip-interface-macros proc-macro implementation detail @@ -27,6 +30,7 @@ wip-public-api-example (workspace-only executable, integration tests, doctest) | --- | --- | --- | | `wip-protocol` | Transport-independent descriptors, object/call/result types, logical `Value`s, validation, and stable protocol error categories. | Application DTOs, implementations, persistence, or wire encoding. | | `wip-https` | Standard HTTPS v1 endpoint/route derivation, UTF-8 JSON codecs, direct success bodies, error/status classification, and retrieval batches. | HTTP I/O, TLS termination, authentication schemes, server frameworks, filesystem projection, or client presentation/cache policy. | +| `wip-client` | Endpoint/security-context-isolated Known Space, normalized object and interface observations, lazy entry trees, immutable operation call contexts, stale transitions, and bounded histories. | HTTP I/O, credentials, persistent cache, automatic operation retries, or presentation widgets/events. | | `wip-interface` | The public `WipType`/`interface` authoring API, Rust-to-protocol codecs, generated descriptors, and typed implementation dispatch. | Object publication, backend lifetime, or transport policy. | | `wip-interface-macros` | Procedural-macro expansion used by `wip-interface`. It is not an application dependency. | A supported direct user API. | | `wip-host` | Opaque interface registration, explicit object membership, validator checks, and interface-qualified dispatch against one host-owned instance. | Backend atomicity, global object lookup, or network serving. | diff --git a/crates/wip-client/Cargo.toml b/crates/wip-client/Cargo.toml new file mode 100644 index 0000000..a6cdd55 --- /dev/null +++ b/crates/wip-client/Cargo.toml @@ -0,0 +1,10 @@ +[package] +name = "wip-client" +version = "0.1.0" +edition = "2024" +description = "Presentation-independent stateful WIP client runtime" +license = "MIT OR Apache-2.0" + +[dependencies] +wip-https = { path = "../wip-https" } +wip-protocol = { path = "../wip-protocol" } diff --git a/crates/wip-client/src/lib.rs b/crates/wip-client/src/lib.rs new file mode 100644 index 0000000..3bd8ede --- /dev/null +++ b/crates/wip-client/src/lib.rs @@ -0,0 +1,1691 @@ +//! Presentation-independent stateful runtime for WIP clients. +//! +//! [`Client`] separates observations by canonical HTTPS endpoint and caller- +//! supplied security context. It creates requests with the standard +//! [`wip_https`] 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. + +#![deny(missing_docs)] + +use std::collections::{BTreeMap, VecDeque}; +use std::error::Error; +use std::fmt::{self, Display, Formatter}; + +use wip_https::http::{Request, Response}; +use wip_https::{ + ClientResponseError, DecodedResponse, Endpoint, Limits, decode_call_operation_response, + decode_fetch_interface_response, decode_fetch_response, decode_fetch_tree_response, + encode_call_operation_request, encode_fetch_interface_request, encode_fetch_request, + encode_fetch_tree_request, +}; +use wip_protocol::{ + CallOperationRequest, CallOperationResponse, FetchInterfaceRequest, FetchRequest, + FetchTreeRequest, FetchTreeResponse, InterfaceDescriptor, InterfaceTarget, Object, + ObjectTreeNode, OperationDeclaration, ProtocolError, ProtocolErrorCode, Target, Value, +}; + +/// Monotonic identity assigned to one logical request. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct RequestId(u64); + +impl RequestId { + /// Returns the process-local numeric identity. + #[must_use] + pub const fn get(self) -> u64 { + self.0 + } +} + +/// Opaque identity for authentication and authorization state. +/// +/// Applications must issue a different value whenever credentials, subject, +/// tenant, or other security-relevant context changes. +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct SecurityContext(String); + +impl SecurityContext { + /// Creates a security-context identity. + #[must_use] + pub fn new(identity: impl Into) -> Self { + Self(identity.into()) + } + + /// Returns the application-defined identity. + #[must_use] + pub fn as_str(&self) -> &str { + &self.0 + } +} + +/// Identity of an isolated client session. +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct SessionId { + endpoint: String, + security_context: SecurityContext, +} + +impl SessionId { + /// Returns the canonical endpoint URL, including its trailing slash. + #[must_use] + pub fn endpoint(&self) -> &str { + &self.endpoint + } + + /// Returns the security context that partitions this session. + #[must_use] + pub fn security_context(&self) -> &SecurityContext { + &self.security_context + } +} + +/// Bounded state limits for the client runtime. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ClientLimits { + /// Maximum simultaneously retained endpoint/security-context sessions. + pub sessions: usize, + /// Maximum normalized objects retained by each session. + pub objects_per_session: usize, + /// Maximum interface observations retained by each session. + pub interfaces_per_session: usize, + /// Maximum requests awaiting a response. + pub in_flight: usize, + /// Maximum call results and diagnostics retained in each history. + pub history: usize, +} + +impl ClientLimits { + /// Creates a limit set, rejecting every zero bound. + pub fn new( + sessions: usize, + objects_per_session: usize, + interfaces_per_session: usize, + in_flight: usize, + history: usize, + ) -> Result { + if sessions == 0 + || objects_per_session == 0 + || interfaces_per_session == 0 + || in_flight == 0 + || history == 0 + { + return Err(ConfigurationError); + } + Ok(Self { + sessions, + objects_per_session, + interfaces_per_session, + in_flight, + history, + }) + } +} + +/// A client bound must be nonzero. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ConfigurationError; + +impl Display for ConfigurationError { + fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result { + formatter.write_str("client limits must be nonzero") + } +} + +impl Error for ConfigurationError {} + +/// High-level category for a client-local failure. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ClientErrorKind { + /// A peer response violated its interaction contract. + InvalidResponse, + /// A valid descriptor uses a format unsupported by this runtime. + UnsupportedDescriptorFormat, + /// The exchange failed outside a valid WIP response envelope. + Transport, + /// A valid Host protocol failure was applied to the observation. + ProtocolFailure, + /// A local request or operation selection was invalid. + InvalidRequest, + /// Required fresh object or interface state was unavailable. + MissingObservation, + /// A configured state bound prevented the requested work. + Capacity, + /// A response was superseded by a newer request for the same subject. + StaleResponse, +} + +/// Cloneable client-local failure suitable for model state and diagnostics. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ClientError { + /// Stable local failure category. + pub kind: ClientErrorKind, + /// Human-readable diagnostic that callers must not branch on. + pub detail: String, +} + +impl ClientError { + fn new(kind: ClientErrorKind, detail: impl Into) -> Self { + Self { + kind, + detail: detail.into(), + } + } +} + +impl Display for ClientError { + fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result { + write!( + formatter, + "client failure ({:?}): {}", + self.kind, self.detail + ) + } +} + +impl Error for ClientError {} + +/// Lifecycle of one cached observation. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ObservationState { + /// A request is currently loading this subject. + Loading(RequestId), + /// The observation is the latest accepted value. + Fresh, + /// The value remains available but must be reobserved before use. + Stale, + /// The most recent observation attempt failed. + Error(ClientError), +} + +/// Object observation associated with a request path. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ObjectObservation { + /// Canonical path used to observe the object. + pub path: String, + /// Last accepted object, if one has been received. + pub object: Option, + /// Object validator copied from `object` for convenient inspection. + pub validator: Option>, + /// Current observation lifecycle. + pub state: ObservationState, +} + +/// Lazily materialized entry-tree node. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TreeObservation { + /// Canonical node path. + pub path: String, + /// Materialized direct child paths in Host order. + pub children: Vec, + /// Whether the latest expansion rooted here omitted indexable entries. + pub truncated: bool, + /// Current expansion lifecycle. + pub state: ObservationState, +} + +/// Cached interface descriptor observation. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct InterfaceObservation { + /// Opaque interface reference. + pub reference: String, + /// Last accepted descriptor, if one has been received. + pub descriptor: Option, + /// Optional interface validator. Absence makes the reference immutable. + pub validator: Option>, + /// Current observation lifecycle. + pub state: ObservationState, +} + +/// Immutable operation inputs fixed before request construction. +#[derive(Debug, Clone, PartialEq)] +pub struct CallContext { + session: SessionId, + object_key: u64, + object: ObjectObservation, + interface: InterfaceObservation, + descriptor: InterfaceDescriptor, + operation: OperationDeclaration, + request: CallOperationRequest, +} + +impl CallContext { + /// Returns the isolated session used by this call. + #[must_use] + pub fn session(&self) -> &SessionId { + &self.session + } + + /// Returns the object observation fixed at call construction. + #[must_use] + pub fn object(&self) -> &ObjectObservation { + &self.object + } + + /// Returns the interface observation fixed at call construction. + #[must_use] + pub fn interface(&self) -> &InterfaceObservation { + &self.interface + } + + /// Returns the descriptor used for both argument and result validation. + #[must_use] + pub fn descriptor(&self) -> &InterfaceDescriptor { + &self.descriptor + } + + /// Returns the exact operation declaration fixed for this call. + #[must_use] + pub fn operation(&self) -> &OperationDeclaration { + &self.operation + } + + /// Returns the fully constructed logical request. + #[must_use] + pub fn request(&self) -> &CallOperationRequest { + &self.request + } +} + +/// Why a dispatched call has an unknown outcome. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum OutcomeUnknownReason { + /// The Host explicitly returned `OperationOutcomeUnknown`. + Protocol, + /// The transport timed out after dispatch. + Timeout, + /// The connection was lost after dispatch. + Disconnect, + /// A response arrived but could not be decoded or validated. + ResponseDecode, +} + +/// Terminal state retained for an operation call. +#[derive(Debug, Clone, PartialEq)] +pub enum CallOutcome { + /// A result validated with the context's fixed descriptor. + Success(CallOperationResponse), + /// A canonical Host failure whose non-execution guarantee is known. + ProtocolFailure(ProtocolError), + /// Dispatch occurred, but execution or commit cannot be ruled out. + Unknown { + /// Stable reason for the uncertainty. + reason: OutcomeUnknownReason, + /// Optional local or Host diagnostic. + failure: Option, + }, + /// The request failed before dispatch and therefore has no unknown effect. + NotDispatched(ClientError), +} + +/// One bounded operation-history record. +#[derive(Debug, Clone, PartialEq)] +pub struct CallRecord { + /// Request identity. + pub request_id: RequestId, + /// Immutable construction and validation context. + pub context: CallContext, + /// Terminal call outcome. + pub outcome: CallOutcome, +} + +/// Transport failure occurring after operation dispatch. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum DispatchedTransportFailure { + /// The transport deadline elapsed. + Timeout, + /// The connection ended before a complete response arrived. + Disconnect, +} + +/// Kind of retained runtime diagnostic. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum DiagnosticKind { + /// HTTP status disagreed with a valid protocol error envelope. + HttpStatusMismatch, + /// A response was rejected because a newer request owns the subject. + StaleResponse, + /// A client-local response or transport failure occurred. + ClientFailure, +} + +/// One bounded diagnostic-history entry. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Diagnostic { + /// Request associated with the diagnostic. + pub request_id: RequestId, + /// Stable diagnostic category. + pub kind: DiagnosticKind, + /// Human-readable bounded detail. + pub detail: String, +} + +/// 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 HTTPS request. + pub request: Request>, +} + +/// Result of applying a correlated response to the runtime. +#[derive(Debug, Clone, PartialEq)] +pub enum Completion { + /// An object observation was updated. + Object(ObjectObservation), + /// A lazy tree expansion was updated. + Tree(TreeObservation), + /// An interface observation was updated. + Interface(InterfaceObservation), + /// An operation reached a terminal outcome. + Call(Box), + /// A valid protocol failure was applied to a retrieval observation. + ProtocolFailure(ProtocolError), + /// A superseded response was rejected without mutating current state. + StaleResponseRejected(RequestId), +} + +#[derive(Debug, Clone)] +struct EntryRecord { + object_key: Option, + state: ObservationState, +} + +#[derive(Debug, Clone)] +struct ObjectRecord { + object: Object, + touched: u64, +} + +#[derive(Debug, Clone)] +struct Session { + entries: BTreeMap, + objects: BTreeMap, + refs: BTreeMap, + trees: BTreeMap, + interfaces: BTreeMap, + calls: VecDeque, + diagnostics: VecDeque, + next_object_key: u64, + clock: u64, +} + +#[derive(Debug, Clone)] +struct InterfaceRecord { + observation: InterfaceObservation, + touched: u64, +} + +impl Session { + fn new() -> Self { + Self { + entries: BTreeMap::new(), + objects: BTreeMap::new(), + refs: BTreeMap::new(), + trees: BTreeMap::new(), + interfaces: BTreeMap::new(), + calls: VecDeque::new(), + diagnostics: VecDeque::new(), + next_object_key: 1, + clock: 0, + } + } + + fn tick(&mut self) -> u64 { + self.clock = self.clock.saturating_add(1); + self.clock + } + + fn object_observation(&self, path: &str) -> Option { + let entry = self.entries.get(path)?; + let object = entry + .object_key + .and_then(|key| self.objects.get(&key)) + .map(|record| record.object.clone()); + let validator = object.as_ref().and_then(|value| value.validator.clone()); + Some(ObjectObservation { + path: path.to_owned(), + object, + validator, + state: entry.state.clone(), + }) + } + + fn mark_object_stale(&mut self, key: u64) { + for entry in self.entries.values_mut() { + if entry.object_key == Some(key) { + entry.state = ObservationState::Stale; + } + } + } + + fn push_diagnostic(&mut self, value: Diagnostic, maximum: usize) { + push_bounded(&mut self.diagnostics, value, maximum); + } + + fn push_call(&mut self, value: CallRecord, maximum: usize) { + push_bounded(&mut self.calls, value, maximum); + } +} + +#[derive(Debug)] +enum Pending { + Fetch { + session: SessionId, + request: FetchRequest, + }, + Tree { + session: SessionId, + request: FetchTreeRequest, + }, + Interface { + session: SessionId, + request: FetchInterfaceRequest, + }, + Call { + context: Box, + dispatched: bool, + }, +} + +impl Pending { + fn session(&self) -> &SessionId { + match self { + Self::Fetch { session, .. } + | Self::Tree { session, .. } + | Self::Interface { session, .. } => session, + Self::Call { context, .. } => context.session(), + } + } +} + +/// Stateful, presentation-independent WIP client runtime. +#[derive(Debug)] +pub struct Client { + 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. + #[must_use] + pub fn new(limits: ClientLimits, wire_limits: Limits) -> 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. + 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(), + security_context, + }; + if !self.sessions.contains_key(&id) { + if self.sessions.len() >= self.limits.sessions { + return Err(ClientError::new( + ClientErrorKind::Capacity, + "session capacity reached", + )); + } + self.sessions.insert(id.clone(), Session::new()); + } + Ok(id) + } + + /// Returns an object observation independently of any UI selection. + #[must_use] + pub fn object(&self, session: &SessionId, path: &str) -> Option { + self.sessions.get(session)?.object_observation(path) + } + + /// Returns the bounded Known Space in canonical path order. + #[must_use] + pub fn known_space(&self, session: &SessionId) -> Option> { + let state = self.sessions.get(session)?; + Some( + state + .entries + .keys() + .filter_map(|path| state.object_observation(path)) + .filter(|observation| observation.object.is_some()) + .collect(), + ) + } + + /// Returns one materialized lazy-tree node. + #[must_use] + pub fn tree(&self, session: &SessionId, path: &str) -> Option<&TreeObservation> { + self.sessions.get(session)?.trees.get(path) + } + + /// Returns the current interface observation. + #[must_use] + pub fn interface(&self, session: &SessionId, reference: &str) -> Option<&InterfaceObservation> { + self.sessions + .get(session)? + .interfaces + .get(reference) + .map(|record| &record.observation) + } + + /// Returns operation history from oldest to newest. + #[must_use] + pub fn call_history(&self, session: &SessionId) -> Option<&VecDeque> { + Some(&self.sessions.get(session)?.calls) + } + + /// Returns diagnostic history from oldest to newest. + #[must_use] + pub fn diagnostics(&self, session: &SessionId) -> Option<&VecDeque> { + Some(&self.sessions.get(session)?.diagnostics) + } + + /// Returns the number of requests currently in flight. + #[must_use] + pub fn in_flight_len(&self) -> usize { + self.pending.len() + } + + /// Returns the immutable context of an operation that is still in flight. + #[must_use] + pub fn pending_call_context(&self, id: RequestId) -> Option<&CallContext> { + match self.pending.get(&id)? { + Pending::Call { context, .. } => Some(context), + _ => None, + } + } + + /// Starts or refreshes an object observation. + pub fn prepare_fetch( + &mut self, + session: &SessionId, + path: impl Into, + ) -> Result { + self.ensure_capacity()?; + let request = FetchRequest { path: path.into() }; + let endpoint = session_endpoint(session)?; + let encoded = + encode_fetch_request(&endpoint, &request, self.wire_limits).map_err(|error| { + ClientError::new(ClientErrorKind::InvalidRequest, error.to_string()) + })?; + let id = self.allocate_request_id(); + let state = self.session_mut(session)?; + let entry = state + .entries + .entry(request.path.clone()) + .or_insert(EntryRecord { + object_key: None, + state: ObservationState::Loading(id), + }); + entry.state = ObservationState::Loading(id); + self.pending.insert( + id, + Pending::Fetch { + session: session.clone(), + request, + }, + ); + Ok(PreparedRequest { + id, + request: encoded, + }) + } + + /// Starts or refreshes a bounded lazy tree expansion. + pub fn prepare_tree( + &mut self, + session: &SessionId, + path: impl Into, + depth: u32, + limit: u32, + ) -> Result { + self.ensure_capacity()?; + let request = FetchTreeRequest { + path: path.into(), + depth, + limit, + }; + let endpoint = session_endpoint(session)?; + let encoded = + encode_fetch_tree_request(&endpoint, &request, self.wire_limits).map_err(|error| { + ClientError::new(ClientErrorKind::InvalidRequest, error.to_string()) + })?; + let id = self.allocate_request_id(); + let state = self.session_mut(session)?; + let tree = state + .trees + .entry(request.path.clone()) + .or_insert(TreeObservation { + path: request.path.clone(), + children: Vec::new(), + truncated: false, + state: ObservationState::Loading(id), + }); + tree.state = ObservationState::Loading(id); + self.pending.insert( + id, + Pending::Tree { + session: session.clone(), + request, + }, + ); + Ok(PreparedRequest { + id, + request: encoded, + }) + } + + /// Starts or refreshes an interface descriptor observation. + pub fn prepare_interface( + &mut self, + session: &SessionId, + reference: impl Into, + ) -> Result { + self.ensure_capacity()?; + let request = FetchInterfaceRequest { + interface: reference.into(), + }; + 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()) + })?; + let id = self.allocate_request_id(); + let maximum = self.limits.interfaces_per_session; + let state = self.session_mut(session)?; + ensure_interface_slot(state, &request.interface, maximum)?; + let touched = state.tick(); + let record = state + .interfaces + .entry(request.interface.clone()) + .or_insert(InterfaceRecord { + observation: InterfaceObservation { + reference: request.interface.clone(), + descriptor: None, + validator: None, + state: ObservationState::Loading(id), + }, + touched, + }); + record.observation.state = ObservationState::Loading(id); + record.touched = touched; + self.pending.insert( + id, + Pending::Interface { + session: session.clone(), + request, + }, + ); + Ok(PreparedRequest { + id, + request: encoded, + }) + } + + /// Constructs an operation from fresh object and interface observations. + /// + /// Membership, both validators, the descriptor, and the operation declaration + /// are fixed in the returned request's immutable [`CallContext`]. + pub fn prepare_call( + &mut self, + session: &SessionId, + path: &str, + interface: &str, + operation: impl Into, + arguments: BTreeMap, + ) -> Result { + self.ensure_capacity()?; + let operation = operation.into(); + let context = { + let state = self.session(session)?; + let object_observation = state.object_observation(path).ok_or_else(|| { + ClientError::new( + ClientErrorKind::MissingObservation, + "object is not observed", + ) + })?; + if object_observation.state != ObservationState::Fresh { + return Err(ClientError::new( + ClientErrorKind::MissingObservation, + "object observation is not fresh", + )); + } + let object = object_observation + .object + .as_ref() + .expect("fresh object exists"); + if !object.interfaces.iter().any(|member| member == interface) { + return Err(ClientError::new( + ClientErrorKind::InvalidRequest, + "interface is not a member of the observed object", + )); + } + let interface_observation = state + .interfaces + .get(interface) + .map(|record| record.observation.clone()) + .ok_or_else(|| { + ClientError::new( + ClientErrorKind::MissingObservation, + "interface is not observed", + ) + })?; + if interface_observation.state != ObservationState::Fresh { + return Err(ClientError::new( + ClientErrorKind::MissingObservation, + "interface observation is not fresh", + )); + } + let descriptor = interface_observation + .descriptor + .clone() + .expect("fresh interface exists"); + descriptor + .validate_arguments(&operation, &arguments) + .map_err(|error| { + ClientError::new(ClientErrorKind::InvalidRequest, error.to_string()) + })?; + let operation_declaration = descriptor + .operations + .iter() + .find(|declaration| declaration.name == operation) + .expect("validated operation exists") + .clone(); + let request = CallOperationRequest { + target: Target { + path: path.to_owned(), + validator: object.validator.clone(), + }, + interface: InterfaceTarget { + reference: interface.to_owned(), + validator: interface_observation.validator.clone(), + }, + operation, + arguments, + }; + let object_key = state.entries[path] + .object_key + .expect("fresh entry has object key"); + CallContext { + session: session.clone(), + object_key, + object: object_observation, + interface: interface_observation, + descriptor, + operation: operation_declaration, + 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()))?; + let id = self.allocate_request_id(); + self.pending.insert( + id, + Pending::Call { + context: Box::new(context), + dispatched: false, + }, + ); + Ok(PreparedRequest { + id, + request: encoded, + }) + } + + /// Marks an operation request as handed to the transport. + pub fn mark_dispatched(&mut self, id: RequestId) -> Result<(), ClientError> { + match self.pending.get_mut(&id) { + Some(Pending::Call { dispatched, .. }) => { + *dispatched = true; + Ok(()) + } + Some(_) => Err(ClientError::new( + ClientErrorKind::InvalidRequest, + "only operation calls have a dispatch boundary", + )), + None => Err(unknown_request()), + } + } + + /// Applies one HTTP response to the exact request identity that produced it. + pub fn complete( + &mut self, + id: RequestId, + response: Response>, + ) -> Result { + let pending = self.pending.remove(&id).ok_or_else(unknown_request)?; + match pending { + Pending::Fetch { session, request } => { + self.complete_fetch(id, &session, request, response) + } + Pending::Tree { session, request } => { + self.complete_tree(id, &session, request, response) + } + Pending::Interface { session, request } => { + self.complete_interface(id, &session, request, response) + } + Pending::Call { context, .. } => self.complete_call(id, *context, response), + } + } + + /// Records a transport failure for a request that produced no response. + pub fn fail_transport( + &mut self, + id: RequestId, + detail: impl Into, + ) -> Result { + let detail = detail.into(); + let pending = self.pending.remove(&id).ok_or_else(unknown_request)?; + match pending { + Pending::Call { + context, + dispatched: true, + } => self.finish_unknown_call( + id, + *context, + OutcomeUnknownReason::Disconnect, + ClientError::new(ClientErrorKind::Transport, detail), + ), + Pending::Call { + context, + dispatched: false, + } => { + let record = CallRecord { + request_id: id, + context: *context, + outcome: CallOutcome::NotDispatched(ClientError::new( + ClientErrorKind::Transport, + detail, + )), + }; + let maximum = self.limits.history; + self.session_mut(record.context.session())? + .push_call(record.clone(), maximum); + Ok(Completion::Call(Box::new(record))) + } + other => { + let session = other.session().clone(); + let error = ClientError::new(ClientErrorKind::Transport, detail); + self.mark_retrieval_error(id, other, error.clone())?; + self.push_failure_diagnostic(&session, id, &error)?; + Err(error) + } + } + } + + /// Records a timeout or disconnect after an operation was dispatched. + pub fn fail_dispatched_call( + &mut self, + id: RequestId, + failure: DispatchedTransportFailure, + detail: impl Into, + ) -> Result { + match self.pending.get(&id) { + Some(Pending::Call { + dispatched: true, .. + }) => {} + Some(_) => { + return Err(ClientError::new( + ClientErrorKind::InvalidRequest, + "request was not a dispatched operation call", + )); + } + None => return Err(unknown_request()), + } + let Some(Pending::Call { context, .. }) = self.pending.remove(&id) else { + unreachable!("validated dispatched call remains pending") + }; + let reason = match failure { + DispatchedTransportFailure::Timeout => OutcomeUnknownReason::Timeout, + DispatchedTransportFailure::Disconnect => OutcomeUnknownReason::Disconnect, + }; + self.finish_unknown_call( + id, + *context, + reason, + ClientError::new(ClientErrorKind::Transport, detail), + ) + } + + fn complete_fetch( + &mut self, + id: RequestId, + session: &SessionId, + request: FetchRequest, + response: Response>, + ) -> Result { + if !self.entry_request_is_current(session, &request.path, id) { + return self.reject_stale(session, id); + } + match decode_fetch_response(&request, &response, self.wire_limits) { + Ok(DecodedResponse::Success(object)) => { + let maximum = self.limits.objects_per_session; + let mut candidate = self.session(session)?.clone(); + if let Err(error) = insert_object(&mut candidate, &request.path, object, maximum) { + self.session_mut(session)? + .entries + .get_mut(&request.path) + .expect("loading entry") + .state = ObservationState::Error(error.clone()); + self.push_failure_diagnostic(session, id, &error)?; + return Err(error); + } + let observation = candidate + .object_observation(&request.path) + .expect("inserted observation exists"); + self.sessions.insert(session.clone(), candidate); + Ok(Completion::Object(observation)) + } + Ok(DecodedResponse::ProtocolFailure { + error, + status_mismatch, + }) => { + self.record_status_mismatch(session, id, status_mismatch)?; + self.session_mut(session)? + .entries + .get_mut(&request.path) + .expect("loading entry") + .state = if error.code == ProtocolErrorCode::NotFound { + ObservationState::Stale + } else { + ObservationState::Error(ClientError::new( + ClientErrorKind::ProtocolFailure, + error.to_string(), + )) + }; + Ok(Completion::ProtocolFailure(error)) + } + Err(error) => self.retrieval_decode_error( + id, + session, + RetrievalSubject::Object(request.path), + error, + ), + } + } + + fn complete_tree( + &mut self, + id: RequestId, + session: &SessionId, + request: FetchTreeRequest, + response: Response>, + ) -> Result { + if !self.tree_request_is_current(session, &request.path, id) { + return self.reject_stale(session, id); + } + match decode_fetch_tree_response(&request, &response, self.wire_limits) { + Ok(DecodedResponse::Success(tree)) => { + let maximum = self.limits.objects_per_session; + let mut candidate = self.session(session)?.clone(); + if let Err(error) = apply_tree(&mut candidate, &request.path, tree, maximum) { + self.session_mut(session)? + .trees + .get_mut(&request.path) + .expect("loading tree") + .state = ObservationState::Error(error.clone()); + self.push_failure_diagnostic(session, id, &error)?; + return Err(error); + } + let observation = candidate.trees[&request.path].clone(); + self.sessions.insert(session.clone(), candidate); + Ok(Completion::Tree(observation)) + } + Ok(DecodedResponse::ProtocolFailure { + error, + status_mismatch, + }) => { + self.record_status_mismatch(session, id, status_mismatch)?; + self.session_mut(session)? + .trees + .get_mut(&request.path) + .expect("loading tree") + .state = if error.code == ProtocolErrorCode::NotFound { + ObservationState::Stale + } else { + ObservationState::Error(ClientError::new( + ClientErrorKind::ProtocolFailure, + error.to_string(), + )) + }; + Ok(Completion::ProtocolFailure(error)) + } + Err(error) => self.retrieval_decode_error( + id, + session, + RetrievalSubject::Tree(request.path), + error, + ), + } + } + + fn complete_interface( + &mut self, + id: RequestId, + session: &SessionId, + request: FetchInterfaceRequest, + response: Response>, + ) -> 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)) => { + let state = self.session_mut(session)?; + let existing = &state.interfaces[&request.interface].observation; + if let Some(descriptor) = &existing.descriptor + && ((existing.validator.is_none() && descriptor != &value.descriptor) + || (existing.validator.is_some() + && existing.validator == value.validator + && descriptor != &value.descriptor)) + { + let error = ClientError::new( + ClientErrorKind::InvalidResponse, + "interface representation changed for an immutable observation identity", + ); + state + .interfaces + .get_mut(&request.interface) + .expect("interface exists") + .observation + .state = ObservationState::Error(error.clone()); + self.push_failure_diagnostic(session, id, &error)?; + return Err(error); + } + let touched = state.tick(); + let observation = InterfaceObservation { + reference: value.interface, + descriptor: Some(value.descriptor), + validator: value.validator, + state: ObservationState::Fresh, + }; + let record = state + .interfaces + .get_mut(&request.interface) + .expect("interface exists"); + record.observation = observation.clone(); + record.touched = touched; + Ok(Completion::Interface(observation)) + } + Ok(DecodedResponse::ProtocolFailure { + error, + status_mismatch, + }) => { + self.record_status_mismatch(session, id, status_mismatch)?; + self.session_mut(session)? + .interfaces + .get_mut(&request.interface) + .expect("interface exists") + .observation + .state = ObservationState::Error(ClientError::new( + ClientErrorKind::ProtocolFailure, + error.to_string(), + )); + Ok(Completion::ProtocolFailure(error)) + } + Err(error) => self.retrieval_decode_error( + id, + session, + RetrievalSubject::Interface(request.interface), + error, + ), + } + } + + fn complete_call( + &mut self, + id: RequestId, + context: CallContext, + response: Response>, + ) -> Result { + match decode_call_operation_response( + &context.request, + &context.descriptor, + &response, + self.wire_limits, + ) { + Ok(DecodedResponse::Success(value)) => { + let session_id = context.session.clone(); + let maximum = self.limits.history; + let state = self.session_mut(&session_id)?; + if state + .entries + .get(&context.request.target.path) + .and_then(|entry| entry.object_key) + == Some(context.object_key) + { + if let Some(object) = state.objects.get_mut(&context.object_key) { + object.object.validator = value.validator.clone(); + } + state.mark_object_stale(context.object_key); + } + let record = CallRecord { + request_id: id, + context, + outcome: CallOutcome::Success(value), + }; + state.push_call(record.clone(), maximum); + Ok(Completion::Call(Box::new(record))) + } + Ok(DecodedResponse::ProtocolFailure { + error, + status_mismatch, + }) => { + 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)?; + state.mark_object_stale(context.object_key); + let record = CallRecord { + request_id: id, + context, + outcome: CallOutcome::Unknown { + reason: OutcomeUnknownReason::Protocol, + failure: None, + }, + }; + state.push_call(record.clone(), maximum); + return Ok(Completion::Call(Box::new(record))); + } + let maximum = self.limits.history; + let state = self.session_mut(&session_id)?; + match error.code { + ProtocolErrorCode::ValidatorRequired + | ProtocolErrorCode::ValidatorMismatch + | ProtocolErrorCode::InterfaceMismatch => { + state.mark_object_stale(context.object_key); + } + ProtocolErrorCode::InterfaceValidatorRequired + | ProtocolErrorCode::InterfaceValidatorMismatch => { + if let Some(interface) = state + .interfaces + .get_mut(&context.request.interface.reference) + { + interface.observation.state = ObservationState::Stale; + } + } + _ => {} + } + let record = CallRecord { + request_id: id, + context, + outcome: CallOutcome::ProtocolFailure(error), + }; + state.push_call(record.clone(), maximum); + Ok(Completion::Call(Box::new(record))) + } + Err(error) => { + let failure = classify_response_error(error); + self.finish_unknown_call(id, context, OutcomeUnknownReason::ResponseDecode, failure) + } + } + } + + fn finish_unknown_call( + &mut self, + id: RequestId, + context: CallContext, + reason: OutcomeUnknownReason, + failure: ClientError, + ) -> Result { + let session_id = context.session.clone(); + let maximum = self.limits.history; + let state = self.session_mut(&session_id)?; + state.mark_object_stale(context.object_key); + state.push_diagnostic( + Diagnostic { + request_id: id, + kind: DiagnosticKind::ClientFailure, + detail: failure.to_string(), + }, + maximum, + ); + let record = CallRecord { + request_id: id, + context, + outcome: CallOutcome::Unknown { + reason, + failure: Some(failure), + }, + }; + state.push_call(record.clone(), maximum); + Ok(Completion::Call(Box::new(record))) + } + + fn retrieval_decode_error( + &mut self, + id: RequestId, + session: &SessionId, + subject: RetrievalSubject, + error: ClientResponseError, + ) -> Result { + let error = classify_response_error(error); + match subject { + RetrievalSubject::Object(path) => { + self.session_mut(session)? + .entries + .get_mut(&path) + .expect("entry exists") + .state = ObservationState::Error(error.clone()); + } + RetrievalSubject::Tree(path) => { + self.session_mut(session)? + .trees + .get_mut(&path) + .expect("tree exists") + .state = ObservationState::Error(error.clone()); + } + RetrievalSubject::Interface(reference) => { + self.session_mut(session)? + .interfaces + .get_mut(&reference) + .expect("interface exists") + .observation + .state = ObservationState::Error(error.clone()); + } + } + self.push_failure_diagnostic(session, id, &error)?; + Err(error) + } + + fn mark_retrieval_error( + &mut self, + _id: RequestId, + pending: Pending, + error: ClientError, + ) -> Result<(), ClientError> { + match pending { + Pending::Fetch { session, request } => { + self.session_mut(&session)? + .entries + .get_mut(&request.path) + .expect("entry exists") + .state = ObservationState::Error(error); + } + Pending::Tree { session, request } => { + self.session_mut(&session)? + .trees + .get_mut(&request.path) + .expect("tree exists") + .state = ObservationState::Error(error); + } + Pending::Interface { session, request } => { + self.session_mut(&session)? + .interfaces + .get_mut(&request.interface) + .expect("interface exists") + .observation + .state = ObservationState::Error(error); + } + Pending::Call { .. } => unreachable!("handled above"), + } + 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, + id: RequestId, + error: &ClientError, + ) -> Result<(), ClientError> { + let maximum = self.limits.history; + self.session_mut(session)?.push_diagnostic( + Diagnostic { + request_id: id, + kind: DiagnosticKind::ClientFailure, + detail: error.to_string(), + }, + maximum, + ); + Ok(()) + } + + fn reject_stale( + &mut self, + session: &SessionId, + id: RequestId, + ) -> Result { + let maximum = self.limits.history; + self.session_mut(session)?.push_diagnostic( + Diagnostic { + request_id: id, + kind: DiagnosticKind::StaleResponse, + detail: "response superseded by a newer request".into(), + }, + maximum, + ); + Ok(Completion::StaleResponseRejected(id)) + } + + fn entry_request_is_current(&self, session: &SessionId, path: &str, id: RequestId) -> bool { + matches!( + self.sessions + .get(session) + .and_then(|state| state.entries.get(path)) + .map(|entry| &entry.state), + Some(ObservationState::Loading(current)) if *current == id + ) + } + + fn tree_request_is_current(&self, session: &SessionId, path: &str, id: RequestId) -> bool { + matches!( + self.sessions + .get(session) + .and_then(|state| state.trees.get(path)) + .map(|tree| &tree.state), + Some(ObservationState::Loading(current)) if *current == id + ) + } + + fn interface_request_is_current( + &self, + session: &SessionId, + reference: &str, + id: RequestId, + ) -> bool { + matches!( + self.sessions + .get(session) + .and_then(|state| state.interfaces.get(reference)) + .map(|interface| &interface.observation.state), + Some(ObservationState::Loading(current)) if *current == id + ) + } + + fn ensure_capacity(&self) -> Result<(), ClientError> { + if self.pending.len() >= self.limits.in_flight { + Err(ClientError::new( + ClientErrorKind::Capacity, + "in-flight request capacity reached", + )) + } else { + Ok(()) + } + } + + fn allocate_request_id(&mut self) -> RequestId { + let id = RequestId(self.next_request_id); + self.next_request_id = self.next_request_id.saturating_add(1); + id + } + + fn session(&self, id: &SessionId) -> Result<&Session, ClientError> { + self.sessions.get(id).ok_or_else(|| { + ClientError::new(ClientErrorKind::InvalidRequest, "unknown client session") + }) + } + + fn session_mut(&mut self, id: &SessionId) -> Result<&mut Session, ClientError> { + self.sessions.get_mut(id).ok_or_else(|| { + ClientError::new(ClientErrorKind::InvalidRequest, "unknown client session") + }) + } +} + +#[derive(Debug)] +enum RetrievalSubject { + Object(String), + Tree(String), + Interface(String), +} + +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 ensure_interface_slot( + state: &mut Session, + reference: &str, + maximum: usize, +) -> Result<(), ClientError> { + if state.interfaces.contains_key(reference) || state.interfaces.len() < maximum { + return Ok(()); + } + let candidate = state + .interfaces + .iter() + .filter(|(_, record)| !matches!(record.observation.state, ObservationState::Loading(_))) + .min_by_key(|(_, record)| record.touched) + .map(|(reference, _)| reference.clone()) + .ok_or_else(|| { + ClientError::new( + ClientErrorKind::Capacity, + "all interface cache entries are loading", + ) + })?; + state.interfaces.remove(&candidate); + Ok(()) +} + +fn insert_object( + state: &mut Session, + path: &str, + object: Object, + maximum: usize, +) -> Result { + let existing_path_key = state.entries.get(path).and_then(|entry| entry.object_key); + let key = if let Some(reference) = &object.r#ref { + if let Some(key) = state.refs.get(reference).copied() { + if let Some(existing) = state.objects.get(&key) + && existing.object.validator.is_some() + && existing.object.validator == object.validator + && existing.object != object + { + return Err(ClientError::new( + ClientErrorKind::InvalidResponse, + "object representation changed for the same ref and validator", + )); + } + key + } else { + make_object_slot(state, maximum)? + } + } else if let Some(key) = existing_path_key + && state + .objects + .get(&key) + .is_some_and(|record| record.object.r#ref.is_none()) + { + key + } else { + make_object_slot(state, maximum)? + }; + + if let Some(old_key) = existing_path_key + && old_key != key + && !state + .entries + .iter() + .any(|(entry_path, entry)| entry_path != path && entry.object_key == Some(old_key)) + { + remove_object(state, old_key); + } + if let Some(reference) = &object.r#ref { + state.refs.insert(reference.clone(), key); + } + let touched = state.tick(); + state.objects.insert(key, ObjectRecord { object, touched }); + state.entries.insert( + path.to_owned(), + EntryRecord { + object_key: Some(key), + state: ObservationState::Fresh, + }, + ); + for entry in state.entries.values_mut() { + if entry.object_key == Some(key) { + entry.state = ObservationState::Fresh; + } + } + Ok(key) +} + +fn make_object_slot(state: &mut Session, maximum: usize) -> Result { + if state.objects.len() >= maximum { + let candidate = state + .objects + .iter() + .min_by_key(|(_, record)| record.touched) + .map(|(key, _)| *key) + .ok_or_else(|| { + ClientError::new(ClientErrorKind::Capacity, "object cache capacity reached") + })?; + remove_object(state, candidate); + } + let key = state.next_object_key; + state.next_object_key = state.next_object_key.saturating_add(1); + Ok(key) +} + +fn remove_object(state: &mut Session, key: u64) { + if let Some(record) = state.objects.remove(&key) + && let Some(reference) = record.object.r#ref + && state.refs.get(&reference) == Some(&key) + { + state.refs.remove(&reference); + } + let removed_paths: Vec = state + .entries + .iter() + .filter(|(_, entry)| entry.object_key == Some(key)) + .map(|(path, _)| path.clone()) + .collect(); + state + .entries + .retain(|_, entry| entry.object_key != Some(key)); + for path in &removed_paths { + state.trees.remove(path); + } + for tree in state.trees.values_mut() { + tree.children + .retain(|path| !removed_paths.iter().any(|removed| removed == path)); + } +} + +fn apply_tree( + state: &mut Session, + root_path: &str, + response: FetchTreeResponse, + maximum: usize, +) -> Result<(), ClientError> { + let mut nodes = Vec::new(); + flatten_tree(root_path, &response.root, &mut nodes); + if nodes.len() > maximum { + return Err(ClientError::new( + ClientErrorKind::Capacity, + "tree response exceeds the local object cache bound", + )); + } + for (path, object, _) in &nodes { + insert_object(state, path, object.clone(), maximum)?; + } + for (path, _, children) in nodes { + let is_root = path == root_path; + state.trees.insert( + path.clone(), + TreeObservation { + path, + children, + truncated: is_root && response.truncated, + state: ObservationState::Fresh, + }, + ); + } + Ok(()) +} + +fn flatten_tree( + path: &str, + node: &ObjectTreeNode, + output: &mut Vec<(String, Object, Vec)>, +) { + let children: Vec = node + .children + .iter() + .map(|child| child_path(path, &child.object.name)) + .collect(); + output.push((path.to_owned(), node.object.clone(), children.clone())); + for (child, child_path) in node.children.iter().zip(children) { + flatten_tree(&child_path, child, output); + } +} + +fn child_path(parent: &str, name: &str) -> String { + if parent == "/" { + format!("/{name}") + } else { + format!("{parent}/{name}") + } +} + +fn push_bounded(history: &mut VecDeque, value: T, maximum: usize) { + if history.len() == maximum { + history.pop_front(); + } + history.push_back(value); +} diff --git a/crates/wip-client/tests/adapter.rs b/crates/wip-client/tests/adapter.rs new file mode 100644 index 0000000..5ad872f --- /dev/null +++ b/crates/wip-client/tests/adapter.rs @@ -0,0 +1,247 @@ +use std::collections::BTreeMap; + +use wip_client::{ + CallOutcome, Client, ClientErrorKind, ClientLimits, Completion, DiagnosticKind, + ObservationState, OutcomeUnknownReason, SecurityContext, +}; +use wip_https::http::{Response, StatusCode}; +use wip_https::{ + Limits, encode_fetch_interface_response, encode_fetch_response, encode_protocol_error_response, +}; +use wip_protocol::{ + FetchInterfaceRequest, FetchInterfaceResponse, FetchRequest, INTERFACE_FORMAT_V1, + InterfaceDescriptor, Object, OperationDeclaration, ParameterDeclaration, ProtocolError, + ProtocolErrorCode, ProtocolInteraction, ReturnDeclaration, TypeExpr, Value, +}; + +fn limits() -> Limits { + Limits::new(64 * 1024, 64 * 1024, 64, 64).unwrap() +} + +fn setup() -> (Client, wip_client::SessionId) { + let mut client = Client::new(ClientLimits::new(2, 16, 16, 16, 16).unwrap(), limits()); + let session = client + .open_session("https://example.test/wip", SecurityContext::new("subject")) + .unwrap(); + (client, session) +} + +fn response(status: StatusCode, body: &[u8]) -> Response> { + Response::builder() + .status(status) + .header("content-type", "application/json; charset=utf-8") + .body(body.to_vec()) + .unwrap() +} + +fn object() -> Object { + Object { + name: "item".into(), + description: None, + interfaces: vec!["math".into()], + r#ref: None, + validator: Some(b"object-v1".to_vec()), + } +} + +fn descriptor(format: &str) -> InterfaceDescriptor { + InterfaceDescriptor { + format: format.into(), + documentation: None, + types: vec![], + operations: vec![OperationDeclaration { + name: "calculate".into(), + documentation: None, + parameters: vec![ParameterDeclaration { + name: "value".into(), + required: true, + documentation: None, + r#type: TypeExpr::Integer, + }], + returns: ReturnDeclaration { + documentation: None, + r#type: TypeExpr::Integer, + }, + }], + } +} + +#[test] +fn malformed_json_unknown_fields_duplicates_and_invalid_success_are_invalid_response() { + let cases: &[&[u8]] = &[ + b"{not json", + br#"{"name":"item","interfaces":[],"unknown":true}"#, + br#"{"name":"item","interfaces":["math","math"]}"#, + br#"{"name":"wrong","interfaces":[]}"#, + ]; + for body in cases { + let (mut client, session) = setup(); + let request = client.prepare_fetch(&session, "/item").unwrap(); + let error = client + .complete(request.id, response(StatusCode::OK, body)) + .unwrap_err(); + assert_eq!(error.kind, ClientErrorKind::InvalidResponse); + assert!(matches!( + client.object(&session, "/item").unwrap().state, + ObservationState::Error(_) + )); + } +} + +#[test] +fn disallowed_code_and_status_success_contradictions_are_invalid_response() { + let cases = [ + ( + StatusCode::NOT_FOUND, + br#"{"error":{"code":"interface_not_found","message":"wrong"}}"#.as_slice(), + ), + ( + StatusCode::OK, + br#"{"error":{"code":"not_found","message":"wrong status"}}"#.as_slice(), + ), + ( + StatusCode::CREATED, + br#"{"name":"item","interfaces":[]}"#.as_slice(), + ), + ]; + for (status, body) in cases { + let (mut client, session) = setup(); + let request = client.prepare_fetch(&session, "/item").unwrap(); + assert_eq!( + client + .complete(request.id, response(status, body)) + .unwrap_err() + .kind, + ClientErrorKind::InvalidResponse + ); + } +} + +#[test] +fn valid_error_envelope_is_authoritative_and_status_mismatch_is_diagnostic() { + let (mut client, session) = setup(); + let request = client.prepare_fetch(&session, "/item").unwrap(); + let error = ProtocolError { + code: ProtocolErrorCode::NotFound, + message: "missing".into(), + }; + let mut encoded = + encode_protocol_error_response(ProtocolInteraction::Fetch, &error, limits()).unwrap(); + *encoded.status_mut() = StatusCode::IM_A_TEAPOT; + + assert_eq!( + client.complete(request.id, encoded).unwrap(), + Completion::ProtocolFailure(error) + ); + assert_eq!( + client.object(&session, "/item").unwrap().state, + ObservationState::Stale + ); + assert_eq!( + client.diagnostics(&session).unwrap().back().unwrap().kind, + DiagnosticKind::HttpStatusMismatch + ); +} + +#[test] +fn non_success_without_valid_envelope_remains_transport_failure() { + let (mut client, session) = setup(); + let request = client.prepare_fetch(&session, "/item").unwrap(); + let error = client + .complete( + request.id, + response(StatusCode::BAD_GATEWAY, b"gateway html"), + ) + .unwrap_err(); + assert_eq!(error.kind, ClientErrorKind::Transport); +} + +#[test] +fn unsupported_descriptor_format_is_distinct_from_invalid_response() { + let (mut client, session) = setup(); + let request = client.prepare_interface(&session, "math").unwrap(); + let logical = FetchInterfaceRequest { + interface: "math".into(), + }; + let value = FetchInterfaceResponse { + interface: "math".into(), + descriptor: descriptor("future-format/2"), + validator: Some(b"v2".to_vec()), + }; + let encoded = encode_fetch_interface_response(&logical, &value, limits()).unwrap(); + let error = client.complete(request.id, encoded).unwrap_err(); + assert_eq!(error.kind, ClientErrorKind::UnsupportedDescriptorFormat); +} + +#[test] +fn malformed_call_success_after_dispatch_is_outcome_unknown_and_stales_object() { + let (mut client, session) = setup(); + let fetch = client.prepare_fetch(&session, "/item").unwrap(); + client + .complete( + fetch.id, + encode_fetch_response( + &FetchRequest { + path: "/item".into(), + }, + &object(), + limits(), + ) + .unwrap(), + ) + .unwrap(); + let interface = client.prepare_interface(&session, "math").unwrap(); + client + .complete( + interface.id, + encode_fetch_interface_response( + &FetchInterfaceRequest { + interface: "math".into(), + }, + &FetchInterfaceResponse { + interface: "math".into(), + descriptor: descriptor(INTERFACE_FORMAT_V1), + validator: None, + }, + limits(), + ) + .unwrap(), + ) + .unwrap(); + let call = client + .prepare_call( + &session, + "/item", + "math", + "calculate", + BTreeMap::from([("value".into(), Value::Integer(2))]), + ) + .unwrap(); + client.mark_dispatched(call.id).unwrap(); + + let completion = client + .complete( + call.id, + response( + StatusCode::OK, + br#"{"result":"not an integer","unknown":true}"#, + ), + ) + .unwrap(); + assert!(matches!( + completion, + Completion::Call(record) + if matches!( + &record.outcome, + CallOutcome::Unknown { + reason: OutcomeUnknownReason::ResponseDecode, + failure: Some(failure), + } if failure.kind == ClientErrorKind::InvalidResponse + ) + )); + assert_eq!( + client.object(&session, "/item").unwrap().state, + ObservationState::Stale + ); + assert_eq!(client.in_flight_len(), 0); +} diff --git a/crates/wip-client/tests/runtime.rs b/crates/wip-client/tests/runtime.rs new file mode 100644 index 0000000..da5c822 --- /dev/null +++ b/crates/wip-client/tests/runtime.rs @@ -0,0 +1,517 @@ +use std::collections::BTreeMap; + +use wip_client::{ + CallOutcome, Client, ClientErrorKind, ClientLimits, Completion, DispatchedTransportFailure, + ObservationState, OutcomeUnknownReason, SecurityContext, +}; +use wip_https::http::Response; +use wip_https::{ + Limits, encode_call_operation_response, encode_fetch_interface_response, encode_fetch_response, + encode_fetch_tree_response, encode_protocol_error_response, +}; +use wip_protocol::{ + CallOperationResponse, FetchInterfaceRequest, FetchInterfaceResponse, FetchRequest, + FetchTreeRequest, FetchTreeResponse, INTERFACE_FORMAT_V1, InterfaceDescriptor, Object, + ObjectTreeNode, OperationDeclaration, ParameterDeclaration, ProtocolError, ProtocolErrorCode, + ProtocolInteraction, ReturnDeclaration, TypeExpr, Value, +}; + +fn wire_limits() -> Limits { + Limits::new(64 * 1024, 64 * 1024, 64, 64).unwrap() +} + +fn client(objects: usize, interfaces: usize, in_flight: usize, history: usize) -> Client { + Client::new( + ClientLimits::new(8, objects, interfaces, in_flight, history).unwrap(), + wire_limits(), + ) +} + +fn object(name: &str, reference: Option<&str>, validator: &[u8]) -> Object { + Object { + name: name.into(), + description: Some(format!("object {name}")), + interfaces: vec!["math".into()], + r#ref: reference.map(str::to_owned), + validator: Some(validator.to_vec()), + } +} + +fn descriptor(return_type: TypeExpr) -> InterfaceDescriptor { + InterfaceDescriptor { + format: INTERFACE_FORMAT_V1.into(), + documentation: None, + types: vec![], + operations: vec![OperationDeclaration { + name: "calculate".into(), + documentation: None, + parameters: vec![ParameterDeclaration { + name: "value".into(), + required: true, + documentation: None, + r#type: TypeExpr::Integer, + }], + returns: ReturnDeclaration { + documentation: None, + r#type: return_type, + }, + }], + } +} + +fn fetch_response(path: &str, value: &Object) -> Response> { + encode_fetch_response(&FetchRequest { path: path.into() }, value, wire_limits()).unwrap() +} + +fn observe_object(client: &mut Client, session: &wip_client::SessionId, path: &str, value: Object) { + let request = client.prepare_fetch(session, path).unwrap(); + client + .complete(request.id, fetch_response(path, &value)) + .unwrap(); +} + +fn observe_interface( + client: &mut Client, + session: &wip_client::SessionId, + value: FetchInterfaceResponse, +) { + let request = client + .prepare_interface(session, value.interface.clone()) + .unwrap(); + let response = encode_fetch_interface_response( + &FetchInterfaceRequest { + interface: value.interface.clone(), + }, + &value, + wire_limits(), + ) + .unwrap(); + client.complete(request.id, response).unwrap(); +} + +fn args(value: i64) -> BTreeMap { + BTreeMap::from([("value".into(), Value::Integer(value))]) +} + +#[test] +fn sessions_partition_known_space_by_canonical_endpoint_and_security_context() { + let mut client = client(8, 8, 8, 8); + let alice = client + .open_session( + "https://example.test/wip", + SecurityContext::new("alice-token"), + ) + .unwrap(); + let alice_canonical = client + .open_session( + "https://example.test/wip/", + SecurityContext::new("alice-token"), + ) + .unwrap(); + let bob = client + .open_session( + "https://example.test/wip", + SecurityContext::new("bob-token"), + ) + .unwrap(); + let other = client + .open_session( + "https://other.test/wip", + SecurityContext::new("alice-token"), + ) + .unwrap(); + + assert_eq!(alice, alice_canonical); + observe_object(&mut client, &alice, "/item", object("item", None, b"a")); + assert!(client.object(&alice, "/item").is_some()); + assert!(client.object(&bob, "/item").is_none()); + assert!(client.object(&other, "/item").is_none()); +} + +#[test] +fn newer_refresh_rejects_out_of_order_response_for_the_same_subject() { + let mut client = client(8, 8, 8, 8); + let session = client + .open_session("https://example.test/wip", SecurityContext::new("subject")) + .unwrap(); + let old = client.prepare_fetch(&session, "/item").unwrap(); + let new = client.prepare_fetch(&session, "/item").unwrap(); + + assert_eq!( + client + .complete( + old.id, + fetch_response("/item", &object("item", None, b"old")) + ) + .unwrap(), + Completion::StaleResponseRejected(old.id) + ); + assert!(matches!( + client.object(&session, "/item").unwrap().state, + ObservationState::Loading(id) if id == new.id + )); + + client + .complete( + new.id, + fetch_response("/item", &object("item", None, b"new")), + ) + .unwrap(); + assert_eq!( + client.object(&session, "/item").unwrap().validator, + Some(b"new".to_vec()) + ); +} + +#[test] +fn normalized_object_cache_rejects_conflicts_and_evicts_to_its_bound() { + let mut client = client(2, 8, 8, 8); + let session = client + .open_session("https://example.test/wip", SecurityContext::new("subject")) + .unwrap(); + observe_object( + &mut client, + &session, + "/a/shared", + object("shared", Some("shared-ref"), b"v1"), + ); + observe_object( + &mut client, + &session, + "/b/shared", + object("shared", Some("shared-ref"), b"v1"), + ); + + let conflict = client.prepare_fetch(&session, "/b/shared").unwrap(); + let mut changed = object("shared", Some("shared-ref"), b"v1"); + changed.description = Some("contradiction".into()); + let error = client + .complete(conflict.id, fetch_response("/b/shared", &changed)) + .unwrap_err(); + assert_eq!(error.kind, ClientErrorKind::InvalidResponse); + + observe_object( + &mut client, + &session, + "/second", + object("second", None, b"2"), + ); + observe_object(&mut client, &session, "/third", object("third", None, b"3")); + assert!(client.object(&session, "/a/shared").is_none()); + assert!(client.object(&session, "/second").is_some()); + assert!(client.object(&session, "/third").is_some()); +} + +#[test] +fn tree_expansion_is_lazy_truncated_and_refreshes_edges() { + let mut client = client(16, 8, 8, 8); + let session = client + .open_session("https://example.test/wip", SecurityContext::new("subject")) + .unwrap(); + assert!(client.tree(&session, "/").is_none()); + + let request = client.prepare_tree(&session, "/", 1, 4).unwrap(); + let logical_request = FetchTreeRequest { + path: "/".into(), + depth: 1, + limit: 4, + }; + let tree = FetchTreeResponse { + root: ObjectTreeNode { + object: object("", None, b"root"), + children: vec![ObjectTreeNode { + object: object("items", None, b"items"), + children: vec![], + }], + }, + truncated: true, + }; + client + .complete( + request.id, + encode_fetch_tree_response(&logical_request, &tree, wire_limits()).unwrap(), + ) + .unwrap(); + assert_eq!(client.tree(&session, "/").unwrap().children, ["/items"]); + assert!(client.tree(&session, "/").unwrap().truncated); + assert!(client.object(&session, "/items").is_some()); + + let refresh = client.prepare_tree(&session, "/", 1, 4).unwrap(); + let refreshed = FetchTreeResponse { + root: ObjectTreeNode { + object: object("", None, b"root-2"), + children: vec![], + }, + truncated: false, + }; + client + .complete( + refresh.id, + encode_fetch_tree_response(&logical_request, &refreshed, wire_limits()).unwrap(), + ) + .unwrap(); + assert!(client.tree(&session, "/").unwrap().children.is_empty()); + assert!(!client.tree(&session, "/").unwrap().truncated); +} + +#[test] +fn call_context_fixes_membership_validators_descriptor_and_result_validation() { + let mut client = client(8, 8, 8, 8); + let session = client + .open_session("https://example.test/wip", SecurityContext::new("subject")) + .unwrap(); + observe_object( + &mut client, + &session, + "/item", + object("item", Some("item-ref"), b"object-v1"), + ); + observe_interface( + &mut client, + &session, + FetchInterfaceResponse { + interface: "math".into(), + descriptor: descriptor(TypeExpr::Integer), + validator: Some(b"interface-v1".to_vec()), + }, + ); + + let call = client + .prepare_call(&session, "/item", "math", "calculate", args(7)) + .unwrap(); + assert_eq!( + client + .pending_call_context(call.id) + .unwrap() + .operation() + .name, + "calculate" + ); + client.mark_dispatched(call.id).unwrap(); + + // A newer interface observation cannot change this in-flight call's schema. + observe_interface( + &mut client, + &session, + FetchInterfaceResponse { + interface: "math".into(), + descriptor: descriptor(TypeExpr::String), + validator: Some(b"interface-v2".to_vec()), + }, + ); + + let history_context = match client + .complete( + call.id, + encode_call_operation_response( + &wip_protocol::CallOperationRequest { + target: wip_protocol::Target { + path: "/item".into(), + validator: Some(b"object-v1".to_vec()), + }, + interface: wip_protocol::InterfaceTarget { + reference: "math".into(), + validator: Some(b"interface-v1".to_vec()), + }, + operation: "calculate".into(), + arguments: args(7), + }, + &descriptor(TypeExpr::Integer), + &CallOperationResponse { + result: Value::Integer(14), + validator: Some(b"object-v2".to_vec()), + }, + wire_limits(), + ) + .unwrap(), + ) + .unwrap() + { + Completion::Call(record) => record, + other => panic!("unexpected completion: {other:?}"), + }; + assert_eq!( + history_context.context.descriptor(), + &descriptor(TypeExpr::Integer) + ); + assert_eq!( + history_context.context.object().validator, + Some(b"object-v1".to_vec()) + ); + assert_eq!( + history_context.context.interface().validator, + Some(b"interface-v1".to_vec()) + ); + assert!(matches!(history_context.outcome, CallOutcome::Success(_))); + assert_eq!( + client.interface(&session, "math").unwrap().descriptor, + Some(descriptor(TypeExpr::String)) + ); + assert_eq!( + client.object(&session, "/item").unwrap().state, + ObservationState::Stale + ); +} + +#[test] +fn validator_mismatch_stales_only_its_observation_and_never_retries() { + let mut client = client(8, 8, 8, 8); + let session = client + .open_session("https://example.test/wip", SecurityContext::new("subject")) + .unwrap(); + observe_object(&mut client, &session, "/item", object("item", None, b"v1")); + observe_interface( + &mut client, + &session, + FetchInterfaceResponse { + interface: "math".into(), + descriptor: descriptor(TypeExpr::Integer), + validator: Some(b"i1".to_vec()), + }, + ); + let call = client + .prepare_call(&session, "/item", "math", "calculate", args(1)) + .unwrap(); + client.mark_dispatched(call.id).unwrap(); + let error = ProtocolError { + code: ProtocolErrorCode::InterfaceValidatorMismatch, + message: "stale descriptor".into(), + }; + let response = + encode_protocol_error_response(ProtocolInteraction::CallOperation, &error, wire_limits()) + .unwrap(); + client.complete(call.id, response).unwrap(); + + assert_eq!( + client.interface(&session, "math").unwrap().state, + ObservationState::Stale + ); + assert_eq!( + client.object(&session, "/item").unwrap().state, + ObservationState::Fresh + ); + assert_eq!(client.in_flight_len(), 0); + assert_eq!(client.call_history(&session).unwrap().len(), 1); + + // Reobservation changes cache state but does not manufacture a replacement call. + observe_interface( + &mut client, + &session, + FetchInterfaceResponse { + interface: "math".into(), + descriptor: descriptor(TypeExpr::Integer), + validator: Some(b"i2".to_vec()), + }, + ); + assert_eq!(client.in_flight_len(), 0); + assert_eq!(client.call_history(&session).unwrap().len(), 1); +} + +#[test] +fn protocol_and_transport_unknown_outcomes_stale_without_retry() { + let mut client = client(8, 8, 8, 8); + let session = client + .open_session("https://example.test/wip", SecurityContext::new("subject")) + .unwrap(); + observe_object(&mut client, &session, "/item", object("item", None, b"v1")); + observe_interface( + &mut client, + &session, + FetchInterfaceResponse { + interface: "math".into(), + descriptor: descriptor(TypeExpr::Integer), + validator: None, + }, + ); + + let call = client + .prepare_call(&session, "/item", "math", "calculate", args(1)) + .unwrap(); + client.mark_dispatched(call.id).unwrap(); + let unknown = ProtocolError { + code: ProtocolErrorCode::OperationOutcomeUnknown, + message: "commit uncertain".into(), + }; + let completion = client + .complete( + call.id, + encode_protocol_error_response( + ProtocolInteraction::CallOperation, + &unknown, + wire_limits(), + ) + .unwrap(), + ) + .unwrap(); + assert!(matches!( + completion, + Completion::Call(record) + if matches!( + record.outcome, + CallOutcome::Unknown { + reason: OutcomeUnknownReason::Protocol, + .. + } + ) + )); + assert_eq!( + client.object(&session, "/item").unwrap().state, + ObservationState::Stale + ); + + observe_object(&mut client, &session, "/item", object("item", None, b"v2")); + let second = client + .prepare_call(&session, "/item", "math", "calculate", args(2)) + .unwrap(); + client.mark_dispatched(second.id).unwrap(); + let completion = client + .fail_dispatched_call( + second.id, + DispatchedTransportFailure::Timeout, + "deadline elapsed", + ) + .unwrap(); + assert!(matches!( + completion, + Completion::Call(record) + if matches!( + record.outcome, + CallOutcome::Unknown { + reason: OutcomeUnknownReason::Timeout, + .. + } + ) + )); + assert_eq!(client.in_flight_len(), 0); + assert_eq!(client.call_history(&session).unwrap().len(), 2); +} + +#[test] +fn in_flight_and_history_are_bounded() { + let mut client = client(8, 8, 1, 1); + let session = client + .open_session("https://example.test/wip", SecurityContext::new("subject")) + .unwrap(); + let first = client.prepare_fetch(&session, "/one").unwrap(); + let error = client.prepare_fetch(&session, "/two").unwrap_err(); + assert_eq!(error.kind, ClientErrorKind::Capacity); + client + .complete(first.id, fetch_response("/one", &object("one", None, b"1"))) + .unwrap(); + + // Diagnostic history evicts oldest entries at the configured bound. + let stale_old = client.prepare_fetch(&session, "/one").unwrap(); + let stale_new = client.prepare_fetch(&session, "/one").unwrap_err(); + assert_eq!(stale_new.kind, ClientErrorKind::Capacity); + client + .complete( + stale_old.id, + Response::builder() + .status(200) + .header("content-type", "application/json") + .body(b"not json".to_vec()) + .unwrap(), + ) + .unwrap_err(); + assert_eq!(client.diagnostics(&session).unwrap().len(), 1); +}