From 20b3986ab6022721def06ce10d75e68460445bb2 Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 30 Sep 2026 08:59:53 +0900 Subject: [PATCH] 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);