Hare/wip-rs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: retain observed scope bindings through client refresh states

Browse Source 
Address MR review sequence 5: retain last accepted Object-ref evidence during loading, ordinary staleness and retrieval failure. Mark confirmed path unavailability separately within bounded Entry records so old lifetime evidence cannot contaminate republished ref-less descriptors. Compare retained Object refs on later replacement even when newer Interface ownership previously suppressed an older scope observation.

Add accepted-descriptor and pending-first-fetch old-to-new/ref-loss matrices during Loading/Error, edge-stale evidence, and confirmed-deletion/republication/reset tracking regressions. Validation: 50 runtime + 13 adapter tests; narrow Protocol/HTTP; workspace check/tests/all-target tests; Nix Clippy -D warnings; warning-free rustdoc; fmt/diff checks.
This commit is contained in:
Hare
2026-10-08 03:29:35 +09:00
parent fe6dccda83
commit b2be41d540
2 changed files with 244 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/wip-client/src/lib.rs
+25 -3
View File
@@ -397,6 +397,9 @@ struct EntryRecord {
object_key: Option<u64>,
revision: RequestId,
state: ObservationState,
// Retrieval state can change without losing the last accepted binding.
// Only confirmed unavailability ends that path's lifetime evidence.
confirmed_missing: bool,
}
#[derive(Debug, Clone)]
@@ -751,6 +754,7 @@ impl Client {
object_key: None,
revision: id,
state: ObservationState::Loading(id),
confirmed_missing: false,
});
entry.revision = id;
entry.state = ObservationState::Loading(id);
@@ -1264,6 +1268,7 @@ impl Client {
&request.interface.scope,
current_binding.clone(),
id,
false,
Some(&request.interface),
);
let touched = state.tick();
@@ -1859,6 +1864,12 @@ fn insert_object(
let existing_path_key = existing_path_entry
.as_ref()
.and_then(|entry| entry.object_key);
let previous_scope_ref = existing_path_entry
.as_ref()
.filter(|entry| !entry.confirmed_missing)
.and_then(|entry| entry.object_key)
.and_then(|key| state.objects.get(&key))
.and_then(|record| record.object.r#ref.clone());
let replaceable_key = existing_path_key.filter(|key| {
state
.objects
@@ -1936,6 +1947,7 @@ fn insert_object(
object_key: Some(key),
revision: ownership_revision,
state: ObservationState::Stale,
confirmed_missing: false,
},
}
} else {
@@ -1943,12 +1955,14 @@ fn insert_object(
object_key: Some(key),
revision,
state: ObservationState::Fresh,
confirmed_missing: false,
}
};
state.entries.insert(path.to_owned(), entry);
if !suppressed {
let scope_ref = state.objects[&key].object.r#ref.clone();
update_scope_binding(state, path, scope_ref, revision, None);
let changed = previous_scope_ref.is_some() && previous_scope_ref != scope_ref;
update_scope_binding(state, path, scope_ref, revision, changed, None);
}
Ok(key)
}
@@ -2105,7 +2119,7 @@ fn latest_scope_binding(state: &Session, scope: &str) -> (Option<String>, Reques
let object_binding = state
.entries
.get(scope)
.filter(|entry| entry.state == ObservationState::Fresh)
.filter(|entry| !entry.confirmed_missing)
.and_then(|entry| entry.object_key)
.and_then(|key| state.objects.get(&key))
.map(|record| (record.object.r#ref.clone(), record.revision));
@@ -2124,6 +2138,7 @@ fn update_scope_binding(
scope: &str,
scope_ref: Option<String>,
revision: RequestId,
confirmed_change: bool,
except: Option<&InterfaceReference>,
) {
for (reference, interface) in &mut state.interfaces {
@@ -2134,7 +2149,9 @@ fn update_scope_binding(
{
continue;
}
if interface.scope_binding.is_some() && interface.scope_binding != scope_ref {
if confirmed_change
|| (interface.scope_binding.is_some() && interface.scope_binding != scope_ref)
{
interface.retire(revision);
}
// Corroboration is not wire metadata: never update observation.scope_ref
@@ -2155,6 +2172,11 @@ fn is_subtree_path(root: &str, path: &str) -> bool {
// Direct NotFound confirms that this scope path is no longer available. Edge
// omission and local eviction deliberately do not call this function.
fn retire_scope_subtree(state: &mut Session, root: &str, revision: RequestId) {
for (path, entry) in &mut state.entries {
if is_subtree_path(root, path) && entry.revision <= revision {
entry.confirmed_missing = true;
}
}
state.interfaces.retain(|reference, record| {
!is_subtree_path(root, &reference.scope) || record.accepted_revision > revision
});
crates/wip-client/tests/runtime.rs
+219
View File
@@ -3170,3 +3170,222 @@ fn corroborated_same_scope_ref_preserves_immutability_when_wire_metadata_becomes
let unchanged = scoped_descriptor("/scope", Some("same"), None, TypeExpr::Integer);
observe_interface(&mut client, &session, unchanged);
}
#[test]
fn retained_object_binding_survives_loading_and_failed_refresh_for_first_ref_less_fetch() {
for new_ref in [None, Some("new")] {
for accepted_descriptor in [false, true] {
for failed_refresh in [false, true] {
let (mut client, session) = scope_client();
observe_object(
&mut client,
&session,
"/scope",
object("scope", Some("old"), b"v1"),
);
let loading_scope = client.refresh_object(&session, "/scope").unwrap();
if failed_refresh {
client
.fail_transport(loading_scope.id, "temporary failure")
.unwrap_err();
}
let old = scoped_descriptor("/scope", None, None, TypeExpr::Integer);
let pending = client
.prepare_interface(&session, old.interface.clone())
.unwrap();
if accepted_descriptor {
client
.complete(pending.id, interface_response(&old))
.unwrap();
}
observe_object(
&mut client,
&session,
"/scope",
object("scope", new_ref, b"v2"),
);
let cached = client.interface(&session, &old.interface).unwrap();
assert_eq!(cached.state, ObservationState::Stale);
assert!(cached.descriptor.is_none());
assert_eq!(cached.scope_ref, None);
if !accepted_descriptor {
assert_eq!(
client
.complete(pending.id, interface_response(&old))
.unwrap(),
Completion::StaleResponseRejected(pending.id)
);
}
assert_eq!(
client
.prepare_call(
&session,
"/scope/item",
&old.interface,
"calculate",
args(1)
)
.unwrap_err()
.kind,
ClientErrorKind::MissingObservation
);
let current = scoped_descriptor("/scope", None, None, TypeExpr::String);
observe_interface(&mut client, &session, current.clone());
let call = client
.prepare_call(
&session,
"/scope/item",
&old.interface,
"calculate",
args(2),
)
.unwrap();
assert_eq!(
client
.pending_call_context(call.id)
.unwrap()
.request()
.interface
.scope_ref,
None
);
assert_eq!(
client.pending_call_context(call.id).unwrap().descriptor(),
&current.descriptor
);
if !failed_refresh {
assert_eq!(
client
.complete(
loading_scope.id,
observation_response(
"/scope",
&object("scope", Some("old"), b"v1")
)
)
.unwrap(),
Completion::StaleResponseRejected(loading_scope.id)
);
assert_eq!(
client.interface(&session, &old.interface).unwrap().state,
ObservationState::Fresh
);
}
}
}
}
}
#[test]
fn first_ref_less_descriptor_keeps_lifetime_evidence_when_scope_is_edge_stale() {
let (mut client, session) = scope_client();
let request = ObserveRequest {
path: "/".into(),
depth: 1,
};
for children in [
vec![TestObservation {
object: object("scope", Some("old"), b"scope"),
children: vec![],
}],
vec![],
] {
let pending = client.refresh_children(&session, "/").unwrap();
client
.complete(
pending.id,
encode_test_observation_response(
&request,
&TestObservationResponse {
root: TestObservation {
object: object("", None, b"root"),
children,
},
},
wire_limits(),
)
.unwrap(),
)
.unwrap();
}
assert_eq!(
client.object(&session, "/scope").unwrap().state,
ObservationState::Stale
);
let old = scoped_descriptor("/scope", None, None, TypeExpr::Integer);
observe_interface(&mut client, &session, old.clone());
observe_object(
&mut client,
&session,
"/scope",
object("scope", Some("new"), b"new"),
);
assert_eq!(
client.interface(&session, &old.interface).unwrap().state,
ObservationState::Stale
);
assert!(
client
.interface(&session, &old.interface)
.unwrap()
.descriptor
.is_none()
);
}
#[test]
fn confirmed_missing_scope_does_not_corroborate_republished_descriptor_with_expired_ref() {
let (mut client, session) = scope_client();
observe_object(
&mut client,
&session,
"/scope",
object("scope", Some("old"), b"v1"),
);
let deletion = client.refresh_object(&session, "/scope").unwrap();
client
.complete(
deletion.id,
failure_response(ProtocolInteraction::Observe, ProtocolErrorCode::NotFound),
)
.unwrap();
let current = scoped_descriptor("/scope", None, None, TypeExpr::String);
observe_interface(&mut client, &session, current.clone());
// Loading state alone must not make the confirmed ended Object binding live.
let loading = client.refresh_object(&session, "/scope").unwrap();
observe_interface(&mut client, &session, current.clone());
client
.complete(
loading.id,
observation_response("/scope", &object("scope", Some("new"), b"v2")),
)
.unwrap();
assert_eq!(
client
.interface(&session, &current.interface)
.unwrap()
.state,
ObservationState::Fresh
);
assert_eq!(
client
.interface(&session, &current.interface)
.unwrap()
.scope_ref,
None
);
// A successful new observation restores tracking for subsequent replacements.
observe_object(
&mut client,
&session,
"/scope",
object("scope", Some("another"), b"v3"),
);
assert_eq!(
client
.interface(&session, &current.interface)
.unwrap()
.state,
ObservationState::Stale
);
}