From 2da2a28a67825ed7f56e66fdf23db347cb7f4da9 Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 30 Sep 2026 08:46:12 +0900 Subject: [PATCH 1/6] feat(client): add stateful WIP runtime --- Cargo.lock | 8 + README.md | 12 +- crates/wip-client/Cargo.toml | 10 + crates/wip-client/src/lib.rs | 1691 ++++++++++++++++++++++++++++ crates/wip-client/tests/adapter.rs | 247 ++++ crates/wip-client/tests/runtime.rs | 517 +++++++++ 6 files changed, 2481 insertions(+), 4 deletions(-) create mode 100644 crates/wip-client/Cargo.toml create mode 100644 crates/wip-client/src/lib.rs create mode 100644 crates/wip-client/tests/adapter.rs create mode 100644 crates/wip-client/tests/runtime.rs 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); +} From 20b3986ab6022721def06ce10d75e68460445bb2 Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 30 Sep 2026 08:59:53 +0900 Subject: [PATCH 2/6] fix(client): preserve observation freshness --- crates/wip-client/src/lib.rs | 241 ++++++++++++++++++++++++--- crates/wip-client/tests/runtime.rs | 250 +++++++++++++++++++++++++++++ 2 files changed, 470 insertions(+), 21 deletions(-) diff --git a/crates/wip-client/src/lib.rs b/crates/wip-client/src/lib.rs index 3bd8ede..c7ca8cb 100644 --- a/crates/wip-client/src/lib.rs +++ b/crates/wip-client/src/lib.rs @@ -84,7 +84,7 @@ impl SessionId { pub struct ClientLimits { /// Maximum simultaneously retained endpoint/security-context sessions. pub sessions: usize, - /// Maximum normalized objects retained by each session. + /// Maximum object, path, and lazy-tree observations retained by each session. pub objects_per_session: usize, /// Maximum interface observations retained by each session. pub interfaces_per_session: usize, @@ -624,7 +624,9 @@ impl Client { ClientError::new(ClientErrorKind::InvalidRequest, error.to_string()) })?; let id = self.allocate_request_id(); + let maximum = self.limits.objects_per_session; let state = self.session_mut(session)?; + ensure_entry_slot(state, &request.path, maximum)?; let entry = state .entries .entry(request.path.clone()) @@ -666,7 +668,9 @@ impl Client { ClientError::new(ClientErrorKind::InvalidRequest, error.to_string()) })?; let id = self.allocate_request_id(); + let maximum = self.limits.objects_per_session; let state = self.session_mut(session)?; + ensure_tree_slot(state, &request.path, maximum)?; let tree = state .trees .entry(request.path.clone()) @@ -926,6 +930,21 @@ impl Client { } other => { let session = other.session().clone(); + let is_current = match &other { + Pending::Fetch { request, .. } => { + self.entry_request_is_current(&session, &request.path, id) + } + Pending::Tree { request, .. } => { + self.tree_request_is_current(&session, &request.path, id) + } + Pending::Interface { request, .. } => { + self.interface_request_is_current(&session, &request.interface, id) + } + Pending::Call { .. } => unreachable!("calls handled above"), + }; + if !is_current { + return self.reject_stale(&session, id); + } let error = ClientError::new(ClientErrorKind::Transport, detail); self.mark_retrieval_error(id, other, error.clone())?; self.push_failure_diagnostic(&session, id, &error)?; @@ -1039,7 +1058,7 @@ impl Client { 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) { + if let Err(error) = apply_tree(&mut candidate, &request.path, tree, maximum, id) { self.session_mut(session)? .trees .get_mut(&request.path) @@ -1095,7 +1114,8 @@ impl Client { 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_none() + && (value.validator.is_some() || descriptor != &value.descriptor)) || (existing.validator.is_some() && existing.validator == value.validator && descriptor != &value.descriptor)) @@ -1522,13 +1542,77 @@ fn ensure_interface_slot( Ok(()) } +fn ensure_entry_slot(state: &mut Session, path: &str, maximum: usize) -> Result<(), ClientError> { + if state.entries.contains_key(path) || state.entries.len() < maximum { + return Ok(()); + } + let candidate = state + .entries + .iter() + .find(|(_, entry)| !matches!(entry.state, ObservationState::Loading(_))) + .map(|(path, _)| path.clone()) + .ok_or_else(|| { + ClientError::new( + ClientErrorKind::Capacity, + "all entry observations are loading", + ) + })?; + remove_entry(state, &candidate); + Ok(()) +} + +fn ensure_tree_slot(state: &mut Session, path: &str, maximum: usize) -> Result<(), ClientError> { + if state.trees.contains_key(path) || state.trees.len() < maximum { + return Ok(()); + } + let candidate = state + .trees + .iter() + .find(|(_, tree)| !matches!(tree.state, ObservationState::Loading(_))) + .map(|(path, _)| path.clone()) + .ok_or_else(|| { + ClientError::new( + ClientErrorKind::Capacity, + "all tree observations are loading", + ) + })?; + state.trees.remove(&candidate); + Ok(()) +} + +fn remove_entry(state: &mut Session, path: &str) { + let object_key = state + .entries + .remove(path) + .and_then(|entry| entry.object_key); + state.trees.remove(path); + for tree in state.trees.values_mut() { + tree.children.retain(|child| child != path); + } + if let Some(key) = object_key + && !state + .entries + .values() + .any(|entry| entry.object_key == Some(key)) + { + remove_object(state, key); + } +} + fn insert_object( state: &mut Session, path: &str, object: Object, maximum: usize, ) -> Result { + ensure_entry_slot(state, path, maximum)?; let existing_path_key = state.entries.get(path).and_then(|entry| entry.object_key); + let replaceable_key = existing_path_key.filter(|key| { + !state + .entries + .iter() + .any(|(entry_path, entry)| entry_path != path && entry.object_key == Some(*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) @@ -1543,7 +1627,7 @@ fn insert_object( } key } else { - make_object_slot(state, maximum)? + make_object_slot(state, maximum, replaceable_key)? } } else if let Some(key) = existing_path_key && state @@ -1553,7 +1637,7 @@ fn insert_object( { key } else { - make_object_slot(state, maximum)? + make_object_slot(state, maximum, replaceable_key)? }; if let Some(old_key) = existing_path_key @@ -1577,24 +1661,34 @@ fn insert_object( 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 { +fn make_object_slot( + state: &mut Session, + maximum: usize, + replaceable: Option, +) -> 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") - })?; + let candidate = replaceable.or_else(|| { + state + .objects + .iter() + .filter(|(key, _)| { + !state.entries.values().any(|entry| { + entry.object_key == Some(**key) + && matches!(entry.state, ObservationState::Loading(_)) + }) + }) + .min_by_key(|(_, record)| record.touched) + .map(|(key, _)| *key) + }); + let candidate = candidate.ok_or_else(|| { + ClientError::new( + ClientErrorKind::Capacity, + "all object cache entries are owned by loading observations", + ) + })?; remove_object(state, candidate); } let key = state.next_object_key; @@ -1627,24 +1721,129 @@ fn remove_object(state: &mut Session, key: u64) { } } +fn reserve_tree_metadata( + state: &mut Session, + paths: &[String], + maximum: usize, +) -> Result<(), ClientError> { + while state.entries.len() + + paths + .iter() + .filter(|path| !state.entries.contains_key(*path)) + .count() + > maximum + { + let candidate = state + .entries + .iter() + .find(|(path, entry)| { + !paths.contains(path) && !matches!(entry.state, ObservationState::Loading(_)) + }) + .map(|(path, _)| path.clone()) + .ok_or_else(|| { + ClientError::new( + ClientErrorKind::Capacity, + "tree response cannot evict loading entry observations", + ) + })?; + remove_entry(state, &candidate); + } + while state.trees.len() + + paths + .iter() + .filter(|path| !state.trees.contains_key(*path)) + .count() + > maximum + { + let candidate = state + .trees + .iter() + .find(|(path, tree)| { + !paths.contains(path) && !matches!(tree.state, ObservationState::Loading(_)) + }) + .map(|(path, _)| path.clone()) + .ok_or_else(|| { + ClientError::new( + ClientErrorKind::Capacity, + "tree response cannot evict loading tree observations", + ) + })?; + state.trees.remove(&candidate); + } + Ok(()) +} + +fn mark_subtree_stale(state: &mut Session, root: &str) { + let prefix = format!("{root}/"); + for (path, entry) in &mut state.entries { + if (path == root || path.starts_with(&prefix)) + && !matches!(entry.state, ObservationState::Loading(_)) + { + entry.state = ObservationState::Stale; + } + } + for (path, tree) in &mut state.trees { + if (path == root || path.starts_with(&prefix)) + && !matches!(tree.state, ObservationState::Loading(_)) + { + tree.state = ObservationState::Stale; + } + } +} + fn apply_tree( state: &mut Session, root_path: &str, response: FetchTreeResponse, maximum: usize, + request_id: RequestId, ) -> 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", + "tree response exceeds the local observation bound", )); } + let paths: Vec = nodes.iter().map(|(path, _, _)| path.clone()).collect(); + reserve_tree_metadata(state, &paths, maximum)?; + + if !response.truncated { + let mut omitted = Vec::new(); + for (path, _, children) in &nodes { + if let Some(previous) = state.trees.get(path) { + omitted.extend( + previous + .children + .iter() + .filter(|child| !children.contains(child)) + .cloned(), + ); + } + } + for path in omitted { + mark_subtree_stale(state, &path); + } + } + for (path, object, _) in &nodes { - insert_object(state, path, object.clone(), maximum)?; + let newer_owner = matches!( + state.entries.get(path).map(|entry| &entry.state), + Some(ObservationState::Loading(owner)) if *owner > request_id + ); + if !newer_owner { + insert_object(state, path, object.clone(), maximum)?; + } } for (path, _, children) in nodes { + let newer_owner = matches!( + state.trees.get(&path).map(|tree| &tree.state), + Some(ObservationState::Loading(owner)) if *owner > request_id + ); + if newer_owner { + continue; + } let is_root = path == root_path; state.trees.insert( path.clone(), diff --git a/crates/wip-client/tests/runtime.rs b/crates/wip-client/tests/runtime.rs index da5c822..de4a39d 100644 --- a/crates/wip-client/tests/runtime.rs +++ b/crates/wip-client/tests/runtime.rs @@ -252,6 +252,10 @@ fn tree_expansion_is_lazy_truncated_and_refreshes_edges() { .unwrap(); assert!(client.tree(&session, "/").unwrap().children.is_empty()); assert!(!client.tree(&session, "/").unwrap().truncated); + assert_eq!( + client.object(&session, "/items").unwrap().state, + ObservationState::Stale + ); } #[test] @@ -486,6 +490,252 @@ fn protocol_and_transport_unknown_outcomes_stale_without_retry() { assert_eq!(client.call_history(&session).unwrap().len(), 2); } +#[test] +fn overlapping_tree_work_preserves_newer_subject_request_ownership() { + let mut client = client(8, 8, 8, 8); + let session = client + .open_session("https://example.test/wip", SecurityContext::new("subject")) + .unwrap(); + let tree_request = FetchTreeRequest { + path: "/".into(), + depth: 1, + limit: 4, + }; + let tree_value = FetchTreeResponse { + root: ObjectTreeNode { + object: object("", None, b"root"), + children: vec![ObjectTreeNode { + object: object("child", None, b"tree-child"), + children: vec![], + }], + }, + truncated: false, + }; + + let old_tree = client.prepare_tree(&session, "/", 1, 4).unwrap(); + let newer_fetch = client.prepare_fetch(&session, "/child").unwrap(); + client + .complete( + old_tree.id, + encode_fetch_tree_response(&tree_request, &tree_value, wire_limits()).unwrap(), + ) + .unwrap(); + assert!(matches!( + client.object(&session, "/child").unwrap().state, + ObservationState::Loading(id) if id == newer_fetch.id + )); + client + .complete( + newer_fetch.id, + fetch_response("/child", &object("child", None, b"newer-fetch")), + ) + .unwrap(); + assert_eq!( + client.object(&session, "/child").unwrap().validator, + Some(b"newer-fetch".to_vec()) + ); + + let old_tree = client.prepare_tree(&session, "/", 1, 4).unwrap(); + let newer_child_tree = client.prepare_tree(&session, "/child", 0, 1).unwrap(); + client + .complete( + old_tree.id, + encode_fetch_tree_response(&tree_request, &tree_value, wire_limits()).unwrap(), + ) + .unwrap(); + assert!(matches!( + client.tree(&session, "/child").unwrap().state, + ObservationState::Loading(id) if id == newer_child_tree.id + )); + let child_request = FetchTreeRequest { + path: "/child".into(), + depth: 0, + limit: 1, + }; + let child_value = FetchTreeResponse { + root: ObjectTreeNode { + object: object("child", None, b"newer-tree"), + children: vec![], + }, + truncated: false, + }; + client + .complete( + newer_child_tree.id, + encode_fetch_tree_response(&child_request, &child_value, wire_limits()).unwrap(), + ) + .unwrap(); + assert_eq!( + client.tree(&session, "/child").unwrap().state, + ObservationState::Fresh + ); +} + +#[test] +fn superseded_transport_failure_does_not_poison_newer_refresh() { + 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 newer = client.prepare_fetch(&session, "/item").unwrap(); + assert_eq!( + client + .fail_transport(old.id, "old connection failed") + .unwrap(), + Completion::StaleResponseRejected(old.id) + ); + assert!(matches!( + client.object(&session, "/item").unwrap().state, + ObservationState::Loading(id) if id == newer.id + )); + client + .complete( + newer.id, + fetch_response("/item", &object("item", None, b"new")), + ) + .unwrap(); + + let old = client.prepare_tree(&session, "/", 0, 1).unwrap(); + let newer_tree = client.prepare_tree(&session, "/", 0, 1).unwrap(); + assert_eq!( + client + .fail_transport(old.id, "old tree connection failed") + .unwrap(), + Completion::StaleResponseRejected(old.id) + ); + assert!(matches!( + client.tree(&session, "/").unwrap().state, + ObservationState::Loading(id) if id == newer_tree.id + )); + client + .complete( + newer_tree.id, + encode_fetch_tree_response( + &FetchTreeRequest { + path: "/".into(), + depth: 0, + limit: 1, + }, + &FetchTreeResponse { + root: ObjectTreeNode { + object: object("", None, b"root"), + children: vec![], + }, + truncated: false, + }, + wire_limits(), + ) + .unwrap(), + ) + .unwrap(); + + let old = client.prepare_interface(&session, "math").unwrap(); + let newer = client.prepare_interface(&session, "math").unwrap(); + assert_eq!( + client + .fail_transport(old.id, "old connection failed") + .unwrap(), + Completion::StaleResponseRejected(old.id) + ); + assert!(matches!( + client.interface(&session, "math").unwrap().state, + ObservationState::Loading(id) if id == newer.id + )); + let value = FetchInterfaceResponse { + interface: "math".into(), + descriptor: descriptor(TypeExpr::Integer), + validator: Some(b"current".to_vec()), + }; + client + .complete( + newer.id, + encode_fetch_interface_response( + &FetchInterfaceRequest { + interface: "math".into(), + }, + &value, + wire_limits(), + ) + .unwrap(), + ) + .unwrap(); +} + +#[test] +fn validator_free_interface_observation_remains_immutable() { + let mut client = client(8, 8, 8, 8); + let session = client + .open_session("https://example.test/wip", SecurityContext::new("subject")) + .unwrap(); + observe_interface( + &mut client, + &session, + FetchInterfaceResponse { + interface: "math".into(), + descriptor: descriptor(TypeExpr::Integer), + validator: None, + }, + ); + + let refresh = client.prepare_interface(&session, "math").unwrap(); + let contradictory = FetchInterfaceResponse { + interface: "math".into(), + descriptor: descriptor(TypeExpr::Integer), + validator: Some(b"became-mutable".to_vec()), + }; + let error = client + .complete( + refresh.id, + encode_fetch_interface_response( + &FetchInterfaceRequest { + interface: "math".into(), + }, + &contradictory, + wire_limits(), + ) + .unwrap(), + ) + .unwrap_err(); + assert_eq!(error.kind, ClientErrorKind::InvalidResponse); + assert_eq!(client.interface(&session, "math").unwrap().validator, None); +} + +#[test] +fn failed_entry_and_tree_observation_metadata_is_bounded() { + let mut client = client(2, 8, 2, 8); + let session = client + .open_session("https://example.test/wip", SecurityContext::new("subject")) + .unwrap(); + for path in ["/one", "/two", "/three"] { + let request = client.prepare_fetch(&session, path).unwrap(); + assert_eq!( + client + .fail_transport(request.id, "retrieval failed") + .unwrap_err() + .kind, + ClientErrorKind::Transport + ); + } + assert!(client.object(&session, "/one").is_none()); + assert!(client.object(&session, "/two").is_some()); + assert!(client.object(&session, "/three").is_some()); + + for path in ["/alpha", "/beta", "/gamma"] { + let request = client.prepare_tree(&session, path, 0, 1).unwrap(); + assert_eq!( + client + .fail_transport(request.id, "tree failed") + .unwrap_err() + .kind, + ClientErrorKind::Transport + ); + } + assert!(client.tree(&session, "/alpha").is_none()); + assert!(client.tree(&session, "/beta").is_some()); + assert!(client.tree(&session, "/gamma").is_some()); +} + #[test] fn in_flight_and_history_are_bounded() { let mut client = client(8, 8, 1, 1); From e5162339e9c02204e79ce2273ec26a4ae81099d8 Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 30 Sep 2026 09:09:20 +0900 Subject: [PATCH 3/6] fix(client): retain request ownership revisions --- crates/wip-client/src/lib.rs | 92 ++++++++++++++------- crates/wip-client/tests/runtime.rs | 125 +++++++++++++++++++++++++++++ 2 files changed, 190 insertions(+), 27 deletions(-) diff --git a/crates/wip-client/src/lib.rs b/crates/wip-client/src/lib.rs index c7ca8cb..16bb57d 100644 --- a/crates/wip-client/src/lib.rs +++ b/crates/wip-client/src/lib.rs @@ -219,6 +219,8 @@ pub struct TreeObservation { pub children: Vec, /// Whether the latest expansion rooted here omitted indexable entries. pub truncated: bool, + /// Request revision that most recently owned this tree subject. + pub revision: RequestId, /// Current expansion lifecycle. pub state: ObservationState, } @@ -388,6 +390,7 @@ pub enum Completion { #[derive(Debug, Clone)] struct EntryRecord { object_key: Option, + revision: RequestId, state: ObservationState, } @@ -632,8 +635,10 @@ impl Client { .entry(request.path.clone()) .or_insert(EntryRecord { object_key: None, + revision: id, state: ObservationState::Loading(id), }); + entry.revision = id; entry.state = ObservationState::Loading(id); self.pending.insert( id, @@ -678,8 +683,10 @@ impl Client { path: request.path.clone(), children: Vec::new(), truncated: false, + revision: id, state: ObservationState::Loading(id), }); + tree.revision = id; tree.state = ObservationState::Loading(id); self.pending.insert( id, @@ -1001,7 +1008,9 @@ impl Client { 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) { + if let Err(error) = + insert_object(&mut candidate, &request.path, object, maximum, id) + { self.session_mut(session)? .entries .get_mut(&request.path) @@ -1549,7 +1558,13 @@ fn ensure_entry_slot(state: &mut Session, path: &str, maximum: usize) -> Result< let candidate = state .entries .iter() - .find(|(_, entry)| !matches!(entry.state, ObservationState::Loading(_))) + .find(|(path, entry)| { + !matches!(entry.state, ObservationState::Loading(_)) + && !state + .trees + .get(*path) + .is_some_and(|tree| matches!(tree.state, ObservationState::Loading(_))) + }) .map(|(path, _)| path.clone()) .ok_or_else(|| { ClientError::new( @@ -1604,6 +1619,7 @@ fn insert_object( path: &str, object: Object, maximum: usize, + revision: RequestId, ) -> Result { ensure_entry_slot(state, path, maximum)?; let existing_path_key = state.entries.get(path).and_then(|entry| entry.object_key); @@ -1658,6 +1674,7 @@ fn insert_object( path.to_owned(), EntryRecord { object_key: Some(key), + revision, state: ObservationState::Fresh, }, ); @@ -1675,9 +1692,12 @@ fn make_object_slot( .objects .iter() .filter(|(key, _)| { - !state.entries.values().any(|entry| { + !state.entries.iter().any(|(path, entry)| { entry.object_key == Some(**key) - && matches!(entry.state, ObservationState::Loading(_)) + && (matches!(entry.state, ObservationState::Loading(_)) + || state.trees.get(path).is_some_and(|tree| { + matches!(tree.state, ObservationState::Loading(_)) + })) }) }) .min_by_key(|(_, record)| record.touched) @@ -1712,12 +1732,24 @@ fn remove_object(state: &mut Session, key: u64) { state .entries .retain(|_, entry| entry.object_key != Some(key)); - for path in &removed_paths { - state.trees.remove(path); + let removable_paths: Vec<&String> = removed_paths + .iter() + .filter(|path| { + !state + .trees + .get(*path) + .is_some_and(|tree| matches!(tree.state, ObservationState::Loading(_))) + }) + .collect(); + for path in &removable_paths { + state.trees.remove(*path); } for tree in state.trees.values_mut() { - tree.children - .retain(|path| !removed_paths.iter().any(|removed| removed == path)); + tree.children.retain(|path| { + !removable_paths + .iter() + .any(|removed| removed.as_str() == path) + }); } } @@ -1737,7 +1769,12 @@ fn reserve_tree_metadata( .entries .iter() .find(|(path, entry)| { - !paths.contains(path) && !matches!(entry.state, ObservationState::Loading(_)) + !paths.contains(path) + && !matches!(entry.state, ObservationState::Loading(_)) + && !state + .trees + .get(*path) + .is_some_and(|tree| matches!(tree.state, ObservationState::Loading(_))) }) .map(|(path, _)| path.clone()) .ok_or_else(|| { @@ -1773,19 +1810,15 @@ fn reserve_tree_metadata( Ok(()) } -fn mark_subtree_stale(state: &mut Session, root: &str) { +fn mark_subtree_stale(state: &mut Session, root: &str, request_id: RequestId) { let prefix = format!("{root}/"); for (path, entry) in &mut state.entries { - if (path == root || path.starts_with(&prefix)) - && !matches!(entry.state, ObservationState::Loading(_)) - { + if (path == root || path.starts_with(&prefix)) && entry.revision <= request_id { entry.state = ObservationState::Stale; } } for (path, tree) in &mut state.trees { - if (path == root || path.starts_with(&prefix)) - && !matches!(tree.state, ObservationState::Loading(_)) - { + if (path == root || path.starts_with(&prefix)) && tree.revision <= request_id { tree.state = ObservationState::Stale; } } @@ -1812,7 +1845,11 @@ fn apply_tree( if !response.truncated { let mut omitted = Vec::new(); for (path, _, children) in &nodes { - if let Some(previous) = state.trees.get(path) { + if let Some(previous) = state + .trees + .get(path) + .filter(|previous| previous.revision <= request_id) + { omitted.extend( previous .children @@ -1823,24 +1860,24 @@ fn apply_tree( } } for path in omitted { - mark_subtree_stale(state, &path); + mark_subtree_stale(state, &path, request_id); } } for (path, object, _) in &nodes { - let newer_owner = matches!( - state.entries.get(path).map(|entry| &entry.state), - Some(ObservationState::Loading(owner)) if *owner > request_id - ); + let newer_owner = state + .entries + .get(path) + .is_some_and(|entry| entry.revision > request_id); if !newer_owner { - insert_object(state, path, object.clone(), maximum)?; + insert_object(state, path, object.clone(), maximum, request_id)?; } } for (path, _, children) in nodes { - let newer_owner = matches!( - state.trees.get(&path).map(|tree| &tree.state), - Some(ObservationState::Loading(owner)) if *owner > request_id - ); + let newer_owner = state + .trees + .get(&path) + .is_some_and(|tree| tree.revision > request_id); if newer_owner { continue; } @@ -1851,6 +1888,7 @@ fn apply_tree( path, children, truncated: is_root && response.truncated, + revision: request_id, state: ObservationState::Fresh, }, ); diff --git a/crates/wip-client/tests/runtime.rs b/crates/wip-client/tests/runtime.rs index de4a39d..6366342 100644 --- a/crates/wip-client/tests/runtime.rs +++ b/crates/wip-client/tests/runtime.rs @@ -535,6 +535,54 @@ fn overlapping_tree_work_preserves_newer_subject_request_ownership() { Some(b"newer-fetch".to_vec()) ); + let older_tree = client.prepare_tree(&session, "/", 1, 4).unwrap(); + let completed_newer_fetch = client.prepare_fetch(&session, "/child").unwrap(); + client + .complete( + completed_newer_fetch.id, + fetch_response("/child", &object("child", None, b"completed-newer")), + ) + .unwrap(); + client + .complete( + older_tree.id, + encode_fetch_tree_response(&tree_request, &tree_value, wire_limits()).unwrap(), + ) + .unwrap(); + assert_eq!( + client.object(&session, "/child").unwrap().validator, + Some(b"completed-newer".to_vec()) + ); + + let older_omitting_tree = client.prepare_tree(&session, "/", 1, 4).unwrap(); + let newer_fetch = client.prepare_fetch(&session, "/child").unwrap(); + client + .complete( + newer_fetch.id, + fetch_response("/child", &object("child", None, b"newer-than-omission")), + ) + .unwrap(); + client + .complete( + older_omitting_tree.id, + encode_fetch_tree_response( + &tree_request, + &FetchTreeResponse { + root: ObjectTreeNode { + object: object("", None, b"older-root"), + children: vec![], + }, + truncated: false, + }, + wire_limits(), + ) + .unwrap(), + ) + .unwrap(); + let child = client.object(&session, "/child").unwrap(); + assert_eq!(child.state, ObservationState::Fresh); + assert_eq!(child.validator, Some(b"newer-than-omission".to_vec())); + let old_tree = client.prepare_tree(&session, "/", 1, 4).unwrap(); let newer_child_tree = client.prepare_tree(&session, "/child", 0, 1).unwrap(); client @@ -569,6 +617,40 @@ fn overlapping_tree_work_preserves_newer_subject_request_ownership() { client.tree(&session, "/child").unwrap().state, ObservationState::Fresh ); + + let older_parent = client.prepare_tree(&session, "/", 1, 4).unwrap(); + let newer_child = client.prepare_tree(&session, "/child", 0, 1).unwrap(); + client + .complete( + newer_child.id, + encode_fetch_tree_response( + &child_request, + &FetchTreeResponse { + root: ObjectTreeNode { + object: object("child", None, b"completed-newer-tree"), + children: vec![], + }, + truncated: false, + }, + wire_limits(), + ) + .unwrap(), + ) + .unwrap(); + client + .complete( + older_parent.id, + encode_fetch_tree_response(&tree_request, &tree_value, wire_limits()).unwrap(), + ) + .unwrap(); + assert_eq!( + client.object(&session, "/child").unwrap().validator, + Some(b"completed-newer-tree".to_vec()) + ); + assert_eq!( + client.tree(&session, "/child").unwrap().revision, + newer_child.id + ); } #[test] @@ -701,6 +783,49 @@ fn validator_free_interface_observation_remains_immutable() { assert_eq!(client.interface(&session, "math").unwrap().validator, None); } +#[test] +fn entry_eviction_never_discards_a_loading_tree_subject() { + let mut client = client(1, 8, 8, 8); + let session = client + .open_session("https://example.test/wip", SecurityContext::new("subject")) + .unwrap(); + observe_object(&mut client, &session, "/", object("", None, b"root")); + let tree = client.prepare_tree(&session, "/", 0, 1).unwrap(); + assert_eq!( + client.prepare_fetch(&session, "/other").unwrap_err().kind, + ClientErrorKind::Capacity + ); + assert!(matches!( + client.tree(&session, "/").unwrap().state, + ObservationState::Loading(id) if id == tree.id + )); + client + .complete( + tree.id, + encode_fetch_tree_response( + &FetchTreeRequest { + path: "/".into(), + depth: 0, + limit: 1, + }, + &FetchTreeResponse { + root: ObjectTreeNode { + object: object("", None, b"tree"), + children: vec![], + }, + truncated: false, + }, + wire_limits(), + ) + .unwrap(), + ) + .unwrap(); + assert_eq!( + client.tree(&session, "/").unwrap().state, + ObservationState::Fresh + ); +} + #[test] fn failed_entry_and_tree_observation_metadata_is_bounded() { let mut client = client(2, 8, 2, 8); From f207b71f83ca8fb86a566b20681634700faaadef Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 30 Sep 2026 09:21:44 +0900 Subject: [PATCH 4/6] fix(client): preserve normalized request ownership --- crates/wip-client/src/lib.rs | 107 +++++++++++++++++++++++------ crates/wip-client/tests/runtime.rs | 98 +++++++++++++++++++++++++- 2 files changed, 182 insertions(+), 23 deletions(-) diff --git a/crates/wip-client/src/lib.rs b/crates/wip-client/src/lib.rs index 16bb57d..00cf844 100644 --- a/crates/wip-client/src/lib.rs +++ b/crates/wip-client/src/lib.rs @@ -397,6 +397,7 @@ struct EntryRecord { #[derive(Debug, Clone)] struct ObjectRecord { object: Object, + revision: RequestId, touched: u64, } @@ -675,7 +676,18 @@ impl Client { let id = self.allocate_request_id(); let maximum = self.limits.objects_per_session; let state = self.session_mut(session)?; + ensure_entry_slot(state, &request.path, maximum)?; ensure_tree_slot(state, &request.path, maximum)?; + let entry = state + .entries + .entry(request.path.clone()) + .or_insert(EntryRecord { + object_key: None, + revision: id, + state: ObservationState::Loading(id), + }); + entry.revision = id; + entry.state = ObservationState::Loading(id); let tree = state .trees .entry(request.path.clone()) @@ -1068,11 +1080,12 @@ impl Client { 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, id) { - self.session_mut(session)? - .trees - .get_mut(&request.path) - .expect("loading tree") - .state = ObservationState::Error(error.clone()); + set_tree_subject_state( + self.session_mut(session)?, + &request.path, + id, + ObservationState::Error(error.clone()), + ); self.push_failure_diagnostic(session, id, &error)?; return Err(error); } @@ -1085,11 +1098,7 @@ impl Client { 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 { + let observation_state = if error.code == ProtocolErrorCode::NotFound { ObservationState::Stale } else { ObservationState::Error(ClientError::new( @@ -1097,6 +1106,12 @@ impl Client { error.to_string(), )) }; + set_tree_subject_state( + self.session_mut(session)?, + &request.path, + id, + observation_state, + ); Ok(Completion::ProtocolFailure(error)) } Err(error) => self.retrieval_decode_error( @@ -1206,6 +1221,7 @@ impl Client { { if let Some(object) = state.objects.get_mut(&context.object_key) { object.object.validator = value.validator.clone(); + object.revision = id; } state.mark_object_stale(context.object_key); } @@ -1320,11 +1336,12 @@ impl Client { .state = ObservationState::Error(error.clone()); } RetrievalSubject::Tree(path) => { - self.session_mut(session)? - .trees - .get_mut(&path) - .expect("tree exists") - .state = ObservationState::Error(error.clone()); + set_tree_subject_state( + self.session_mut(session)?, + &path, + id, + ObservationState::Error(error.clone()), + ); } RetrievalSubject::Interface(reference) => { self.session_mut(session)? @@ -1341,7 +1358,7 @@ impl Client { fn mark_retrieval_error( &mut self, - _id: RequestId, + id: RequestId, pending: Pending, error: ClientError, ) -> Result<(), ClientError> { @@ -1354,11 +1371,12 @@ impl Client { .state = ObservationState::Error(error); } Pending::Tree { session, request } => { - self.session_mut(&session)? - .trees - .get_mut(&request.path) - .expect("tree exists") - .state = ObservationState::Error(error); + set_tree_subject_state( + self.session_mut(&session)?, + &request.path, + id, + ObservationState::Error(error), + ); } Pending::Interface { session, request } => { self.session_mut(&session)? @@ -1527,6 +1545,24 @@ fn classify_response_error(error: ClientResponseError) -> ClientError { } } +fn set_tree_subject_state( + state: &mut Session, + path: &str, + revision: RequestId, + observation_state: ObservationState, +) { + if let Some(tree) = state.trees.get_mut(path) + && tree.revision == revision + { + tree.state = observation_state.clone(); + } + if let Some(entry) = state.entries.get_mut(path) + && entry.revision == revision + { + entry.state = observation_state; + } +} + fn ensure_interface_slot( state: &mut Session, reference: &str, @@ -1614,6 +1650,16 @@ fn remove_entry(state: &mut Session, path: &str) { } } +fn object_ownership_revision(state: &Session, key: u64) -> Option { + state + .entries + .values() + .filter(|entry| entry.object_key == Some(key)) + .map(|entry| entry.revision) + .chain(state.objects.get(&key).map(|record| record.revision)) + .max() +} + fn insert_object( state: &mut Session, path: &str, @@ -1632,6 +1678,7 @@ fn insert_object( 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) + && object_ownership_revision(state, key).is_none_or(|owner| owner <= revision) && existing.object.validator.is_some() && existing.object.validator == object.validator && existing.object != object @@ -1669,7 +1716,23 @@ fn insert_object( state.refs.insert(reference.clone(), key); } let touched = state.tick(); - state.objects.insert(key, ObjectRecord { object, touched }); + let ownership_revision = object_ownership_revision(state, key).unwrap_or(revision); + match state.objects.get_mut(&key) { + Some(existing) if ownership_revision > revision => { + existing.revision = ownership_revision; + existing.touched = touched; + } + _ => { + state.objects.insert( + key, + ObjectRecord { + object, + revision, + touched, + }, + ); + } + } state.entries.insert( path.to_owned(), EntryRecord { diff --git a/crates/wip-client/tests/runtime.rs b/crates/wip-client/tests/runtime.rs index 6366342..e866126 100644 --- a/crates/wip-client/tests/runtime.rs +++ b/crates/wip-client/tests/runtime.rs @@ -202,6 +202,94 @@ fn normalized_object_cache_rejects_conflicts_and_evicts_to_its_bound() { assert!(client.object(&session, "/third").is_some()); } +#[test] +fn normalized_aliases_preserve_newer_request_ownership() { + 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, + "/a/shared", + object("shared", Some("shared-ref"), b"initial"), + ); + observe_object( + &mut client, + &session, + "/b/shared", + object("shared", Some("shared-ref"), b"initial"), + ); + let request = FetchTreeRequest { + path: "/a".into(), + depth: 1, + limit: 4, + }; + let older_tree_response = || FetchTreeResponse { + root: ObjectTreeNode { + object: object("a", None, b"root"), + children: vec![ObjectTreeNode { + object: object("shared", Some("shared-ref"), b"older"), + children: vec![], + }], + }, + truncated: false, + }; + + let older_tree = client.prepare_tree(&session, "/a", 1, 4).unwrap(); + let newer_alias = client.prepare_fetch(&session, "/b/shared").unwrap(); + client + .complete( + newer_alias.id, + fetch_response("/b/shared", &object("shared", Some("shared-ref"), b"newer")), + ) + .unwrap(); + client + .complete( + older_tree.id, + encode_fetch_tree_response(&request, &older_tree_response(), wire_limits()).unwrap(), + ) + .unwrap(); + assert_eq!( + client.object(&session, "/a/shared").unwrap().validator, + Some(b"newer".to_vec()) + ); + assert_eq!( + client.object(&session, "/b/shared").unwrap().validator, + Some(b"newer".to_vec()) + ); + + let older_tree = client.prepare_tree(&session, "/a", 1, 4).unwrap(); + let newer_alias = client.prepare_fetch(&session, "/b/shared").unwrap(); + client + .complete( + older_tree.id, + encode_fetch_tree_response(&request, &older_tree_response(), wire_limits()).unwrap(), + ) + .unwrap(); + assert!(matches!( + client.object(&session, "/b/shared").unwrap().state, + ObservationState::Loading(id) if id == newer_alias.id + )); + assert_eq!( + client.object(&session, "/b/shared").unwrap().validator, + Some(b"newer".to_vec()) + ); + client + .complete( + newer_alias.id, + fetch_response( + "/b/shared", + &object("shared", Some("shared-ref"), b"newest"), + ), + ) + .unwrap(); + assert_eq!( + client.object(&session, "/a/shared").unwrap().validator, + Some(b"newest".to_vec()) + ); +} + #[test] fn tree_expansion_is_lazy_truncated_and_refreshes_edges() { let mut client = client(16, 8, 8, 8); @@ -595,6 +683,15 @@ fn overlapping_tree_work_preserves_newer_subject_request_ownership() { client.tree(&session, "/child").unwrap().state, ObservationState::Loading(id) if id == newer_child_tree.id )); + let child_observation = client.object(&session, "/child").unwrap(); + assert!(matches!( + child_observation.state, + ObservationState::Loading(id) if id == newer_child_tree.id + )); + assert_eq!( + child_observation.validator, + Some(b"newer-than-omission".to_vec()) + ); let child_request = FetchTreeRequest { path: "/child".into(), depth: 0, @@ -857,7 +954,6 @@ fn failed_entry_and_tree_observation_metadata_is_bounded() { ); } assert!(client.tree(&session, "/alpha").is_none()); - assert!(client.tree(&session, "/beta").is_some()); assert!(client.tree(&session, "/gamma").is_some()); } From b4e38fc2c22322f4fb9894011611890eab25e6e7 Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 30 Sep 2026 09:38:28 +0900 Subject: [PATCH 5/6] fix(client): preserve operation observation ownership --- crates/wip-client/src/lib.rs | 185 +++++++++++++++++------- crates/wip-client/tests/runtime.rs | 217 +++++++++++++++++++++++++++++ 2 files changed, 354 insertions(+), 48 deletions(-) diff --git a/crates/wip-client/src/lib.rs b/crates/wip-client/src/lib.rs index 00cf844..aa3d166 100644 --- a/crates/wip-client/src/lib.rs +++ b/crates/wip-client/src/lib.rs @@ -398,6 +398,7 @@ struct EntryRecord { struct ObjectRecord { object: Object, revision: RequestId, + call_owners: usize, touched: u64, } @@ -455,7 +456,10 @@ impl Session { }) } - fn mark_object_stale(&mut self, key: u64) { + fn mark_object_stale(&mut self, key: u64, revision: RequestId) { + if let Some(record) = self.objects.get_mut(&key) { + record.revision = record.revision.max(revision); + } for entry in self.entries.values_mut() { if entry.object_key == Some(key) { entry.state = ObservationState::Stale; @@ -463,6 +467,29 @@ impl Session { } } + fn pin_call_object(&mut self, key: u64) { + let record = self + .objects + .get_mut(&key) + .expect("call context object remains observed"); + record.call_owners = record.call_owners.saturating_add(1); + } + + fn release_call_object(&mut self, key: u64) { + let Some(record) = self.objects.get_mut(&key) else { + return; + }; + record.call_owners = record.call_owners.saturating_sub(1); + let remove = record.call_owners == 0 + && !self + .entries + .values() + .any(|entry| entry.object_key == Some(key)); + if remove { + remove_object(self, key); + } + } + fn push_diagnostic(&mut self, value: Diagnostic, maximum: usize) { push_bounded(&mut self.diagnostics, value, maximum); } @@ -863,6 +890,8 @@ impl Client { ) .map_err(|error| ClientError::new(ClientErrorKind::InvalidRequest, error.to_string()))?; let id = self.allocate_request_id(); + self.session_mut(&context.session)? + .pin_call_object(context.object_key); self.pending.insert( id, Pending::Call { @@ -908,7 +937,13 @@ impl Client { Pending::Interface { session, request } => { self.complete_interface(id, &session, request, response) } - Pending::Call { context, .. } => self.complete_call(id, *context, response), + Pending::Call { context, .. } => { + let session = context.session.clone(); + let object_key = context.object_key; + let result = self.complete_call(id, *context, response); + self.session_mut(&session)?.release_call_object(object_key); + result + } } } @@ -924,16 +959,24 @@ impl Client { Pending::Call { context, dispatched: true, - } => self.finish_unknown_call( - id, - *context, - OutcomeUnknownReason::Disconnect, - ClientError::new(ClientErrorKind::Transport, detail), - ), + } => { + let session = context.session.clone(); + let object_key = context.object_key; + let result = self.finish_unknown_call( + id, + *context, + OutcomeUnknownReason::Disconnect, + ClientError::new(ClientErrorKind::Transport, detail), + ); + self.session_mut(&session)?.release_call_object(object_key); + result + } Pending::Call { context, dispatched: false, } => { + let session = context.session.clone(); + let object_key = context.object_key; let record = CallRecord { request_id: id, context: *context, @@ -943,8 +986,9 @@ impl Client { )), }; let maximum = self.limits.history; - self.session_mut(record.context.session())? - .push_call(record.clone(), maximum); + let state = self.session_mut(&session)?; + state.push_call(record.clone(), maximum); + state.release_call_object(object_key); Ok(Completion::Call(Box::new(record))) } other => { @@ -998,12 +1042,16 @@ impl Client { DispatchedTransportFailure::Timeout => OutcomeUnknownReason::Timeout, DispatchedTransportFailure::Disconnect => OutcomeUnknownReason::Disconnect, }; - self.finish_unknown_call( + let session = context.session.clone(); + let object_key = context.object_key; + let result = self.finish_unknown_call( id, *context, reason, ClientError::new(ClientErrorKind::Transport, detail), - ) + ); + self.session_mut(&session)?.release_call_object(object_key); + result } fn complete_fetch( @@ -1213,18 +1261,11 @@ impl Client { 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(); - object.revision = id; - } - state.mark_object_stale(context.object_key); + if let Some(object) = state.objects.get_mut(&context.object_key) { + object.object.validator = value.validator.clone(); + object.revision = object.revision.max(id); } + state.mark_object_stale(context.object_key, id); let record = CallRecord { request_id: id, context, @@ -1242,7 +1283,7 @@ impl Client { if error.code == ProtocolErrorCode::OperationOutcomeUnknown { let maximum = self.limits.history; let state = self.session_mut(&session_id)?; - state.mark_object_stale(context.object_key); + state.mark_object_stale(context.object_key, id); let record = CallRecord { request_id: id, context, @@ -1260,7 +1301,7 @@ impl Client { ProtocolErrorCode::ValidatorRequired | ProtocolErrorCode::ValidatorMismatch | ProtocolErrorCode::InterfaceMismatch => { - state.mark_object_stale(context.object_key); + state.mark_object_stale(context.object_key, id); } ProtocolErrorCode::InterfaceValidatorRequired | ProtocolErrorCode::InterfaceValidatorMismatch => { @@ -1298,7 +1339,7 @@ impl Client { 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.mark_object_stale(context.object_key, id); state.push_diagnostic( Diagnostic { request_id: id, @@ -1596,6 +1637,12 @@ fn ensure_entry_slot(state: &mut Session, path: &str, maximum: usize) -> Result< .iter() .find(|(path, entry)| { !matches!(entry.state, ObservationState::Loading(_)) + && !entry.object_key.is_some_and(|key| { + state + .objects + .get(&key) + .is_some_and(|record| record.call_owners > 0) + }) && !state .trees .get(*path) @@ -1605,7 +1652,7 @@ fn ensure_entry_slot(state: &mut Session, path: &str, maximum: usize) -> Result< .ok_or_else(|| { ClientError::new( ClientErrorKind::Capacity, - "all entry observations are loading", + "all entry observations are owned by in-flight work", ) })?; remove_entry(state, &candidate); @@ -1668,17 +1715,23 @@ fn insert_object( revision: RequestId, ) -> Result { ensure_entry_slot(state, path, maximum)?; - let existing_path_key = state.entries.get(path).and_then(|entry| entry.object_key); + let existing_path_entry = state.entries.get(path).cloned(); + let existing_path_key = existing_path_entry + .as_ref() + .and_then(|entry| entry.object_key); let replaceable_key = existing_path_key.filter(|key| { - !state - .entries - .iter() - .any(|(entry_path, entry)| entry_path != path && entry.object_key == Some(*key)) + state + .objects + .get(key) + .is_none_or(|record| record.call_owners == 0) + && !state + .entries + .iter() + .any(|(entry_path, entry)| entry_path != path && entry.object_key == Some(*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) - && object_ownership_revision(state, key).is_none_or(|owner| owner <= revision) && existing.object.validator.is_some() && existing.object.validator == object.validator && existing.object != object @@ -1717,8 +1770,13 @@ fn insert_object( } let touched = state.tick(); let ownership_revision = object_ownership_revision(state, key).unwrap_or(revision); + let suppressed = ownership_revision > revision; + let call_owners = state + .objects + .get(&key) + .map_or(0, |record| record.call_owners); match state.objects.get_mut(&key) { - Some(existing) if ownership_revision > revision => { + Some(existing) if suppressed => { existing.revision = ownership_revision; existing.touched = touched; } @@ -1728,19 +1786,36 @@ fn insert_object( ObjectRecord { object, revision, + call_owners, touched, }, ); } } - state.entries.insert( - path.to_owned(), + let entry = if suppressed { + match existing_path_entry { + Some(mut entry) if entry.object_key == Some(key) => { + entry.revision = ownership_revision; + if matches!(entry.state, ObservationState::Loading(current) if current == revision) + { + entry.state = ObservationState::Fresh; + } + entry + } + _ => EntryRecord { + object_key: Some(key), + revision: ownership_revision, + state: ObservationState::Stale, + }, + } + } else { EntryRecord { object_key: Some(key), revision, state: ObservationState::Fresh, - }, - ); + } + }; + state.entries.insert(path.to_owned(), entry); Ok(key) } @@ -1754,14 +1829,15 @@ fn make_object_slot( state .objects .iter() - .filter(|(key, _)| { - !state.entries.iter().any(|(path, entry)| { - entry.object_key == Some(**key) - && (matches!(entry.state, ObservationState::Loading(_)) - || state.trees.get(path).is_some_and(|tree| { - matches!(tree.state, ObservationState::Loading(_)) - })) - }) + .filter(|(key, record)| { + record.call_owners == 0 + && !state.entries.iter().any(|(path, entry)| { + entry.object_key == Some(**key) + && (matches!(entry.state, ObservationState::Loading(_)) + || state.trees.get(path).is_some_and(|tree| { + matches!(tree.state, ObservationState::Loading(_)) + })) + }) }) .min_by_key(|(_, record)| record.touched) .map(|(key, _)| *key) @@ -1769,7 +1845,7 @@ fn make_object_slot( let candidate = candidate.ok_or_else(|| { ClientError::new( ClientErrorKind::Capacity, - "all object cache entries are owned by loading observations", + "all object cache entries are owned by in-flight work", ) })?; remove_object(state, candidate); @@ -1780,6 +1856,13 @@ fn make_object_slot( } fn remove_object(state: &mut Session, key: u64) { + if state + .objects + .get(&key) + .is_some_and(|record| record.call_owners > 0) + { + return; + } if let Some(record) = state.objects.remove(&key) && let Some(reference) = record.object.r#ref && state.refs.get(&reference) == Some(&key) @@ -1834,6 +1917,12 @@ fn reserve_tree_metadata( .find(|(path, entry)| { !paths.contains(path) && !matches!(entry.state, ObservationState::Loading(_)) + && !entry.object_key.is_some_and(|key| { + state + .objects + .get(&key) + .is_some_and(|record| record.call_owners > 0) + }) && !state .trees .get(*path) @@ -1843,7 +1932,7 @@ fn reserve_tree_metadata( .ok_or_else(|| { ClientError::new( ClientErrorKind::Capacity, - "tree response cannot evict loading entry observations", + "tree response cannot evict entries owned by in-flight work", ) })?; remove_entry(state, &candidate); diff --git a/crates/wip-client/tests/runtime.rs b/crates/wip-client/tests/runtime.rs index e866126..3b539b1 100644 --- a/crates/wip-client/tests/runtime.rs +++ b/crates/wip-client/tests/runtime.rs @@ -288,6 +288,45 @@ fn normalized_aliases_preserve_newer_request_ownership() { client.object(&session, "/a/shared").unwrap().validator, Some(b"newest".to_vec()) ); + + let older_tree = client.prepare_tree(&session, "/a", 1, 4).unwrap(); + let newer_alias = client.prepare_fetch(&session, "/b/shared").unwrap(); + client + .complete( + newer_alias.id, + fetch_response( + "/b/shared", + &object("shared", Some("shared-ref"), b"newest"), + ), + ) + .unwrap(); + let mut contradictory = object("shared", Some("shared-ref"), b"newest"); + contradictory.description = Some("contradictory representation".into()); + let error = client + .complete( + older_tree.id, + encode_fetch_tree_response( + &request, + &FetchTreeResponse { + root: ObjectTreeNode { + object: object("a", None, b"root"), + children: vec![ObjectTreeNode { + object: contradictory, + children: vec![], + }], + }, + truncated: false, + }, + wire_limits(), + ) + .unwrap(), + ) + .unwrap_err(); + assert_eq!(error.kind, ClientErrorKind::InvalidResponse); + assert_eq!( + client.object(&session, "/b/shared").unwrap().validator, + Some(b"newest".to_vec()) + ); } #[test] @@ -445,6 +484,184 @@ fn call_context_fixes_membership_validators_descriptor_and_result_validation() { ); } +#[test] +fn older_retrieval_cannot_restore_freshness_after_operation_completion() { + 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("shared-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 older_alias = client.prepare_fetch(&session, "/alias/item").unwrap(); + let call = client + .prepare_call(&session, "/item", "math", "calculate", args(7)) + .unwrap(); + let request = client + .pending_call_context(call.id) + .unwrap() + .request() + .clone(); + let fixed_descriptor = client + .pending_call_context(call.id) + .unwrap() + .descriptor() + .clone(); + client.mark_dispatched(call.id).unwrap(); + client + .complete( + call.id, + encode_call_operation_response( + &request, + &fixed_descriptor, + &CallOperationResponse { + result: Value::Integer(14), + validator: Some(b"object-v2".to_vec()), + }, + wire_limits(), + ) + .unwrap(), + ) + .unwrap(); + assert_eq!( + client.object(&session, "/item").unwrap().state, + ObservationState::Stale + ); + + client + .complete( + older_alias.id, + fetch_response( + "/alias/item", + &object("item", Some("shared-ref"), b"object-v1"), + ), + ) + .unwrap(); + let alias = client.object(&session, "/alias/item").unwrap(); + assert_eq!(alias.state, ObservationState::Stale); + assert_eq!(alias.validator, Some(b"object-v2".to_vec())); + assert_eq!( + client + .prepare_call(&session, "/alias/item", "math", "calculate", args(1)) + .unwrap_err() + .kind, + ClientErrorKind::MissingObservation + ); + + observe_object( + &mut client, + &session, + "/item", + object("item", Some("shared-ref"), b"object-v2"), + ); + let older_alias = client.prepare_fetch(&session, "/second/item").unwrap(); + let call = client + .prepare_call(&session, "/item", "math", "calculate", args(8)) + .unwrap(); + client.mark_dispatched(call.id).unwrap(); + let unknown = ProtocolError { + code: ProtocolErrorCode::OperationOutcomeUnknown, + message: "commit uncertain".into(), + }; + client + .complete( + call.id, + encode_protocol_error_response( + ProtocolInteraction::CallOperation, + &unknown, + wire_limits(), + ) + .unwrap(), + ) + .unwrap(); + client + .complete( + older_alias.id, + fetch_response( + "/second/item", + &object("item", Some("shared-ref"), b"object-v2"), + ), + ) + .unwrap(); + assert_eq!( + client.object(&session, "/second/item").unwrap().state, + ObservationState::Stale + ); +} + +#[test] +fn in_flight_calls_pin_target_identity_until_staleness_is_applied() { + let mut client = client(1, 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("shared-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(); + client.mark_dispatched(call.id).unwrap(); + + assert_eq!( + client.prepare_fetch(&session, "/other").unwrap_err().kind, + ClientErrorKind::Capacity + ); + let unknown = ProtocolError { + code: ProtocolErrorCode::OperationOutcomeUnknown, + message: "commit uncertain".into(), + }; + client + .complete( + call.id, + encode_protocol_error_response( + ProtocolInteraction::CallOperation, + &unknown, + wire_limits(), + ) + .unwrap(), + ) + .unwrap(); + assert_eq!( + client.object(&session, "/item").unwrap().state, + ObservationState::Stale + ); + + observe_object( + &mut client, + &session, + "/other", + object("other", None, b"other"), + ); + assert!(client.object(&session, "/item").is_none()); + assert!(client.object(&session, "/other").is_some()); +} + #[test] fn validator_mismatch_stales_only_its_observation_and_never_retries() { let mut client = client(8, 8, 8, 8); From e1998fb8fd03748d0cad9c5b5fe745d1dca9a4f2 Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 30 Sep 2026 09:45:53 +0900 Subject: [PATCH 6/6] fix(client): validate suppressed tree objects --- crates/wip-client/src/lib.rs | 35 +++++++++++++++------- crates/wip-client/tests/runtime.rs | 47 ++++++++++++++++++++++++++++++ 2 files changed, 72 insertions(+), 10 deletions(-) diff --git a/crates/wip-client/src/lib.rs b/crates/wip-client/src/lib.rs index aa3d166..2465727 100644 --- a/crates/wip-client/src/lib.rs +++ b/crates/wip-client/src/lib.rs @@ -1707,6 +1707,29 @@ fn object_ownership_revision(state: &Session, key: u64) -> Option { .max() } +fn validate_object_consistency(state: &Session, object: &Object) -> Result<(), ClientError> { + let Some(reference) = &object.r#ref else { + return Ok(()); + }; + let Some(existing) = state + .refs + .get(reference) + .and_then(|key| state.objects.get(key)) + else { + return Ok(()); + }; + if 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", + )); + } + Ok(()) +} + fn insert_object( state: &mut Session, path: &str, @@ -1714,6 +1737,7 @@ fn insert_object( maximum: usize, revision: RequestId, ) -> Result { + validate_object_consistency(state, &object)?; ensure_entry_slot(state, path, maximum)?; let existing_path_entry = state.entries.get(path).cloned(); let existing_path_key = existing_path_entry @@ -1731,16 +1755,6 @@ fn insert_object( }); 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, replaceable_key)? @@ -2017,6 +2031,7 @@ fn apply_tree( } for (path, object, _) in &nodes { + validate_object_consistency(state, object)?; let newer_owner = state .entries .get(path) diff --git a/crates/wip-client/tests/runtime.rs b/crates/wip-client/tests/runtime.rs index 3b539b1..d2c877f 100644 --- a/crates/wip-client/tests/runtime.rs +++ b/crates/wip-client/tests/runtime.rs @@ -327,6 +327,53 @@ fn normalized_aliases_preserve_newer_request_ownership() { client.object(&session, "/b/shared").unwrap().validator, Some(b"newest".to_vec()) ); + + observe_object( + &mut client, + &session, + "/child", + object("child", Some("child-ref"), b"stable"), + ); + let older_tree = client.prepare_tree(&session, "/", 1, 4).unwrap(); + let newer_child = client.prepare_fetch(&session, "/child").unwrap(); + client + .complete( + newer_child.id, + fetch_response("/child", &object("child", Some("child-ref"), b"stable")), + ) + .unwrap(); + let mut contradictory = object("child", Some("child-ref"), b"stable"); + contradictory.description = Some("same-path contradiction".into()); + let root_request = FetchTreeRequest { + path: "/".into(), + depth: 1, + limit: 4, + }; + let error = client + .complete( + older_tree.id, + encode_fetch_tree_response( + &root_request, + &FetchTreeResponse { + root: ObjectTreeNode { + object: object("", None, b"root"), + children: vec![ObjectTreeNode { + object: contradictory, + children: vec![], + }], + }, + truncated: false, + }, + wire_limits(), + ) + .unwrap(), + ) + .unwrap_err(); + assert_eq!(error.kind, ClientErrorKind::InvalidResponse); + assert_eq!( + client.object(&session, "/child").unwrap().state, + ObservationState::Fresh + ); } #[test]