fix(client): preserve observation freshness

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<u64, ClientError> {
+    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<u64, ClientError> {
+fn make_object_slot(
+    state: &mut Session,
+    maximum: usize,
+    replaceable: Option<u64>,
+) -> Result<u64, ClientError> {
     if state.objects.len() >= maximum {
-        let candidate = state
+        let candidate = replaceable.or_else(|| {
-            .objects
+            state
-            .iter()
+                .objects
-            .min_by_key(|(_, record)| record.touched)
+                .iter()
-            .map(|(key, _)| *key)
+                .filter(|(key, _)| {
-            .ok_or_else(|| {
+                    !state.entries.values().any(|entry| {
-                ClientError::new(ClientErrorKind::Capacity, "object cache capacity reached")
+                        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<String> = 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(),

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);