From f207b71f83ca8fb86a566b20681634700faaadef Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 30 Sep 2026 09:21:44 +0900 Subject: [PATCH] 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()); }