feat: prefetch visible observation frontier

This commit is contained in:
2026-09-30 19:26:50 +09:00
parent ea7b0f00ea
commit 06edf62d4e
9 changed files with 440 additions and 68 deletions
+128 -3
View File
@@ -1,7 +1,8 @@
//! 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
//! supplied security context. It reconciles requested observation coverage without
//! duplicating fresh or loading work, 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.
@@ -630,8 +631,64 @@ impl Client {
}
}
/// Starts or refreshes an object observation through the requested depth.
pub fn prepare_observe(
/// Ensures that an observation is available through `depth` without issuing
/// duplicate work.
///
/// Returns `None` when the requested coverage is already fresh, is currently
/// loading, is blocked by a retained failure or stale object state that requires
/// an explicit refresh, or when the bounded in-flight capacity is currently
/// full. Callers may invoke this method again after applying another completion.
pub fn ensure_observed(
&mut self,
session: &SessionId,
path: impl Into<String>,
depth: u32,
) -> Result<Option<PreparedRequest>, ClientError> {
let path = path.into();
wip_protocol::validate_path(&path).map_err(|error| {
ClientError::new(ClientErrorKind::InvalidRequest, error.to_string())
})?;
let coverage = observation_coverage(self.session(session)?, &path, depth);
if coverage != ObservationCoverage::Missing || self.pending.len() >= self.limits.in_flight {
return Ok(None);
}
self.refresh_observed(session, path, depth).map(Some)
}
/// Ensures that an object's direct children have been observed without
/// duplicating fresh or loading work.
pub fn ensure_children_observed(
&mut self,
session: &SessionId,
path: impl Into<String>,
) -> Result<Option<PreparedRequest>, ClientError> {
self.ensure_observed(session, path, 1)
}
/// Explicitly refreshes only one object while preserving previously observed
/// children in Known Space.
pub fn refresh_object(
&mut self,
session: &SessionId,
path: impl Into<String>,
) -> Result<PreparedRequest, ClientError> {
self.refresh_observed(session, path, 0)
}
/// Explicitly refreshes one object and its direct children.
pub fn refresh_children(
&mut self,
session: &SessionId,
path: impl Into<String>,
) -> Result<PreparedRequest, ClientError> {
self.refresh_observed(session, path, 1)
}
/// Explicitly starts or refreshes an object observation through `depth`.
///
/// Unlike [`Client::ensure_observed`], this always creates a new request and
/// supersedes older requests for the same root path.
pub fn refresh_observed(
&mut self,
session: &SessionId,
path: impl Into<String>,
@@ -1430,6 +1487,64 @@ enum RetrievalSubject {
Interface(String),
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ObservationCoverage {
Missing,
Loading,
Covered,
Blocked,
}
fn observation_coverage(state: &Session, path: &str, depth: u32) -> ObservationCoverage {
let Some(entry) = state.entries.get(path) else {
return ObservationCoverage::Missing;
};
match &entry.state {
ObservationState::Loading(_) => return ObservationCoverage::Loading,
ObservationState::Stale | ObservationState::Error(_) => {
return ObservationCoverage::Blocked;
}
ObservationState::Fresh => {}
}
if entry
.object_key
.is_none_or(|key| !state.objects.contains_key(&key))
{
return ObservationCoverage::Missing;
}
if depth == 0 {
return ObservationCoverage::Covered;
}
let Some(tree) = state.trees.get(path) else {
return ObservationCoverage::Missing;
};
match &tree.state {
ObservationState::Loading(_) => return ObservationCoverage::Loading,
ObservationState::Stale => return ObservationCoverage::Missing,
ObservationState::Error(_) => return ObservationCoverage::Blocked,
ObservationState::Fresh => {}
}
let mut saw_loading = false;
let mut saw_missing = false;
for child in &tree.children {
match observation_coverage(state, child, depth - 1) {
ObservationCoverage::Blocked => return ObservationCoverage::Blocked,
ObservationCoverage::Loading => saw_loading = true,
ObservationCoverage::Missing => saw_missing = true,
ObservationCoverage::Covered => {}
}
}
if saw_loading {
ObservationCoverage::Loading
} else if saw_missing {
ObservationCoverage::Missing
} else {
ObservationCoverage::Covered
}
}
fn session_endpoint(session: &SessionId) -> Result<Endpoint, ClientError> {
Endpoint::parse(&session.endpoint)
.map_err(|error| ClientError::new(ClientErrorKind::InvalidRequest, error.to_string()))
@@ -1883,6 +1998,16 @@ fn apply_observation(
.collect();
reserve_observation_metadata(state, &paths, &observed_children_paths, maximum)?;
for (path, _, children) in &nodes {
if children.is_none()
&& let Some(tree) = state.trees.get_mut(path)
&& tree.revision < request_id
&& matches!(tree.state, ObservationState::Loading(_))
{
tree.state = ObservationState::Stale;
}
}
let mut omitted = Vec::new();
for (path, _, children) in &nodes {
let Some(children) = children else {
+5 -5
View File
@@ -77,7 +77,7 @@ fn malformed_json_unknown_fields_duplicates_and_invalid_success_are_invalid_resp
];
for body in cases {
let (mut client, session) = setup();
let request = client.prepare_observe(&session, "/item", 0).unwrap();
let request = client.refresh_observed(&session, "/item", 0).unwrap();
let error = client
.complete(request.id, response(StatusCode::OK, body))
.unwrap_err();
@@ -107,7 +107,7 @@ fn disallowed_code_and_status_success_contradictions_are_invalid_response() {
];
for (status, body) in cases {
let (mut client, session) = setup();
let request = client.prepare_observe(&session, "/item", 0).unwrap();
let request = client.refresh_observed(&session, "/item", 0).unwrap();
assert_eq!(
client
.complete(request.id, response(status, body))
@@ -121,7 +121,7 @@ fn disallowed_code_and_status_success_contradictions_are_invalid_response() {
#[test]
fn valid_error_envelope_is_authoritative_and_status_mismatch_is_diagnostic() {
let (mut client, session) = setup();
let request = client.prepare_observe(&session, "/item", 0).unwrap();
let request = client.refresh_observed(&session, "/item", 0).unwrap();
let error = ProtocolError {
code: ProtocolErrorCode::NotFound,
message: "missing".into(),
@@ -147,7 +147,7 @@ fn valid_error_envelope_is_authoritative_and_status_mismatch_is_diagnostic() {
#[test]
fn non_success_without_valid_envelope_remains_transport_failure() {
let (mut client, session) = setup();
let request = client.prepare_observe(&session, "/item", 0).unwrap();
let request = client.refresh_observed(&session, "/item", 0).unwrap();
let error = client
.complete(
request.id,
@@ -177,7 +177,7 @@ fn unsupported_descriptor_format_is_distinct_from_invalid_response() {
#[test]
fn malformed_call_success_after_dispatch_is_outcome_unknown_and_stales_object() {
let (mut client, session) = setup();
let observation = client.prepare_observe(&session, "/item", 0).unwrap();
let observation = client.refresh_observed(&session, "/item", 0).unwrap();
client
.complete(
observation.id,
+234 -43
View File
@@ -109,7 +109,7 @@ fn observation_response(path: &str, value: &Object) -> Response<Vec<u8>> {
}
fn observe_object(client: &mut Client, session: &wip_client::SessionId, path: &str, value: Object) {
let request = client.prepare_observe(session, path, 0).unwrap();
let request = client.refresh_observed(session, path, 0).unwrap();
client
.complete(request.id, observation_response(path, &value))
.unwrap();
@@ -173,14 +173,129 @@ fn sessions_partition_known_space_by_canonical_endpoint_and_security_context() {
assert!(client.object(&other, "/item").is_none());
}
#[test]
fn ensure_observed_deduplicates_loading_and_fresh_coverage() {
let mut client = client(8, 8, 8, 8);
let session = client
.open_session("https://example.test/wip", SecurityContext::new("subject"))
.unwrap();
let request = ObserveRequest {
path: "/".into(),
depth: 1,
};
let value = TestObservationResponse {
root: TestObservation {
object: object("", None, b"root"),
children: vec![TestObservation {
object: object("child", None, b"child"),
children: vec![],
}],
},
};
let root = client
.ensure_observed(&session, "/", 1)
.unwrap()
.expect("missing coverage starts an observation");
assert!(client.ensure_observed(&session, "/", 1).unwrap().is_none());
client
.complete(
root.id,
encode_test_observation_response(&request, &value, wire_limits()).unwrap(),
)
.unwrap();
assert!(client.ensure_observed(&session, "/", 1).unwrap().is_none());
let child = client
.ensure_observed(&session, "/child", 1)
.unwrap()
.expect("depth-boundary children remain missing coverage");
assert!(
client
.ensure_observed(&session, "/child", 1)
.unwrap()
.is_none()
);
client
.complete(
child.id,
encode_test_observation_response(
&ObserveRequest {
path: "/child".into(),
depth: 1,
},
&TestObservationResponse {
root: TestObservation {
object: object("child", None, b"child"),
children: vec![],
},
},
wire_limits(),
)
.unwrap(),
)
.unwrap();
assert!(
client
.ensure_observed(&session, "/child", 1)
.unwrap()
.is_none()
);
}
#[test]
fn ensure_observed_respects_capacity_and_does_not_loop_on_failure() {
let mut client = client(8, 8, 1, 8);
let session = client
.open_session("https://example.test/wip", SecurityContext::new("subject"))
.unwrap();
let first = client
.ensure_observed(&session, "/one", 0)
.unwrap()
.expect("first observation fits");
assert_eq!(
client
.ensure_observed(&session, "relative", 0)
.unwrap_err()
.kind,
ClientErrorKind::InvalidRequest
);
assert!(
client
.ensure_observed(&session, "/two", 0)
.unwrap()
.is_none()
);
assert_eq!(
client
.fail_transport(first.id, "scripted failure")
.unwrap_err()
.kind,
ClientErrorKind::Transport
);
assert!(
client
.ensure_observed(&session, "/one", 0)
.unwrap()
.is_none(),
"retained failures require an explicit refresh"
);
assert!(
client
.ensure_observed(&session, "/two", 0)
.unwrap()
.is_some()
);
}
#[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_observe(&session, "/item", 0).unwrap();
let new = client.prepare_observe(&session, "/item", 0).unwrap();
let old = client.refresh_observed(&session, "/item", 0).unwrap();
let new = client.refresh_observed(&session, "/item", 0).unwrap();
assert_eq!(
client
@@ -227,7 +342,7 @@ fn normalized_object_cache_rejects_conflicts_and_evicts_to_its_bound() {
object("shared", Some("shared-ref"), b"v1"),
);
let conflict = client.prepare_observe(&session, "/b/shared", 0).unwrap();
let conflict = client.refresh_observed(&session, "/b/shared", 0).unwrap();
let mut changed = object("shared", Some("shared-ref"), b"v1");
changed.description = Some("contradiction".into());
let error = client
@@ -279,8 +394,8 @@ fn normalized_aliases_preserve_newer_request_ownership() {
},
};
let older_tree = client.prepare_observe(&session, "/a", 1).unwrap();
let newer_alias = client.prepare_observe(&session, "/b/shared", 0).unwrap();
let older_tree = client.refresh_observed(&session, "/a", 1).unwrap();
let newer_alias = client.refresh_observed(&session, "/b/shared", 0).unwrap();
client
.complete(
newer_alias.id,
@@ -303,8 +418,8 @@ fn normalized_aliases_preserve_newer_request_ownership() {
Some(b"newer".to_vec())
);
let older_tree = client.prepare_observe(&session, "/a", 1).unwrap();
let newer_alias = client.prepare_observe(&session, "/b/shared", 0).unwrap();
let older_tree = client.refresh_observed(&session, "/a", 1).unwrap();
let newer_alias = client.refresh_observed(&session, "/b/shared", 0).unwrap();
client
.complete(
older_tree.id,
@@ -334,8 +449,8 @@ fn normalized_aliases_preserve_newer_request_ownership() {
Some(b"newest".to_vec())
);
let older_tree = client.prepare_observe(&session, "/a", 1).unwrap();
let newer_alias = client.prepare_observe(&session, "/b/shared", 0).unwrap();
let older_tree = client.refresh_observed(&session, "/a", 1).unwrap();
let newer_alias = client.refresh_observed(&session, "/b/shared", 0).unwrap();
client
.complete(
newer_alias.id,
@@ -378,8 +493,8 @@ fn normalized_aliases_preserve_newer_request_ownership() {
"/child",
object("child", Some("child-ref"), b"stable"),
);
let older_tree = client.prepare_observe(&session, "/", 1).unwrap();
let newer_child = client.prepare_observe(&session, "/child", 0).unwrap();
let older_tree = client.refresh_observed(&session, "/", 1).unwrap();
let newer_child = client.refresh_observed(&session, "/child", 0).unwrap();
client
.complete(
newer_child.id,
@@ -426,7 +541,7 @@ fn observation_expansion_is_complete_and_refreshes_edges() {
.unwrap();
assert!(client.tree(&session, "/").is_none());
let request = client.prepare_observe(&session, "/", 1).unwrap();
let request = client.refresh_observed(&session, "/", 1).unwrap();
let logical_request = ObserveRequest {
path: "/".into(),
depth: 1,
@@ -449,7 +564,7 @@ fn observation_expansion_is_complete_and_refreshes_edges() {
assert_eq!(client.tree(&session, "/").unwrap().children, ["/items"]);
assert!(client.object(&session, "/items").is_some());
let refresh = client.prepare_observe(&session, "/", 1).unwrap();
let refresh = client.refresh_observed(&session, "/", 1).unwrap();
let refreshed = TestObservationResponse {
root: TestObservation {
object: object("", None, b"root-2"),
@@ -479,7 +594,7 @@ fn depth_zero_observation_preserves_previously_observed_children() {
path: "/".into(),
depth: 1,
};
let deep = client.prepare_observe(&session, "/", 1).unwrap();
let deep = client.refresh_observed(&session, "/", 1).unwrap();
client
.complete(
deep.id,
@@ -500,7 +615,7 @@ fn depth_zero_observation_preserves_previously_observed_children() {
)
.unwrap();
let shallow = client.prepare_observe(&session, "/", 0).unwrap();
let shallow = client.refresh_observed(&session, "/", 0).unwrap();
client
.complete(
shallow.id,
@@ -536,8 +651,8 @@ fn shallow_observation_supersedes_loading_deep_observation_without_leaking_loadi
},
};
let deep = client.prepare_observe(&session, "/", 1).unwrap();
let shallow = client.prepare_observe(&session, "/", 0).unwrap();
let deep = client.refresh_observed(&session, "/", 1).unwrap();
let shallow = client.refresh_observed(&session, "/", 0).unwrap();
assert_eq!(
client.tree(&session, "/").unwrap().state,
ObservationState::Stale
@@ -564,6 +679,80 @@ fn shallow_observation_supersedes_loading_deep_observation_without_leaking_loadi
);
}
#[test]
fn newer_parent_boundary_observation_replans_superseded_child_lookahead() {
let mut client = client(16, 8, 8, 8);
let session = client
.open_session("https://example.test/wip", SecurityContext::new("subject"))
.unwrap();
let root_request = ObserveRequest {
path: "/".into(),
depth: 1,
};
let root_value = |validator: &'static [u8]| TestObservationResponse {
root: TestObservation {
object: object("", None, validator),
children: vec![TestObservation {
object: object("child", None, validator),
children: vec![],
}],
},
};
let initial = client.refresh_observed(&session, "/", 1).unwrap();
client
.complete(
initial.id,
encode_test_observation_response(&root_request, &root_value(b"initial"), wire_limits())
.unwrap(),
)
.unwrap();
let older_child = client.refresh_observed(&session, "/child", 1).unwrap();
let newer_parent = client.refresh_observed(&session, "/", 1).unwrap();
client
.complete(
newer_parent.id,
encode_test_observation_response(&root_request, &root_value(b"newer"), wire_limits())
.unwrap(),
)
.unwrap();
assert_eq!(
client.tree(&session, "/child").unwrap().state,
ObservationState::Stale
);
let replanned = client
.ensure_observed(&session, "/child", 1)
.unwrap()
.expect("superseded lookahead is missing rather than permanently loading");
assert_eq!(
client
.complete(
older_child.id,
encode_test_observation_response(
&ObserveRequest {
path: "/child".into(),
depth: 1,
},
&TestObservationResponse {
root: TestObservation {
object: object("child", None, b"older"),
children: vec![],
},
},
wire_limits(),
)
.unwrap(),
)
.unwrap(),
Completion::StaleResponseRejected(older_child.id)
);
assert!(matches!(
client.object(&session, "/child").unwrap().state,
ObservationState::Loading(id) if id == replanned.id
));
}
#[test]
fn call_context_fixes_membership_validators_descriptor_and_result_validation() {
let mut client = client(8, 8, 8, 8);
@@ -818,7 +1007,7 @@ fn older_retrieval_cannot_overwrite_success_validator() {
},
);
let older_alias = client.prepare_observe(&session, "/alias/item", 0).unwrap();
let older_alias = client.refresh_observed(&session, "/alias/item", 0).unwrap();
let call = client
.prepare_call(&session, "/item", "math", "calculate", args(7))
.unwrap();
@@ -878,7 +1067,9 @@ fn older_retrieval_cannot_overwrite_success_validator() {
"/item",
object("item", Some("shared-ref"), b"object-v2"),
);
let older_alias = client.prepare_observe(&session, "/second/item", 0).unwrap();
let older_alias = client
.refresh_observed(&session, "/second/item", 0)
.unwrap();
let call = client
.prepare_call(&session, "/item", "math", "calculate", args(8))
.unwrap();
@@ -941,7 +1132,7 @@ fn in_flight_calls_pin_target_identity_until_staleness_is_applied() {
assert_eq!(
client
.prepare_observe(&session, "/other", 0)
.refresh_observed(&session, "/other", 0)
.unwrap_err()
.kind,
ClientErrorKind::Capacity
@@ -1129,8 +1320,8 @@ fn overlapping_tree_work_preserves_newer_subject_request_ownership() {
},
};
let old_tree = client.prepare_observe(&session, "/", 1).unwrap();
let newer_observation = client.prepare_observe(&session, "/child", 0).unwrap();
let old_tree = client.refresh_observed(&session, "/", 1).unwrap();
let newer_observation = client.refresh_observed(&session, "/child", 0).unwrap();
client
.complete(
old_tree.id,
@@ -1152,8 +1343,8 @@ fn overlapping_tree_work_preserves_newer_subject_request_ownership() {
Some(b"newer-observation".to_vec())
);
let older_tree = client.prepare_observe(&session, "/", 1).unwrap();
let completed_newer_observation = client.prepare_observe(&session, "/child", 0).unwrap();
let older_tree = client.refresh_observed(&session, "/", 1).unwrap();
let completed_newer_observation = client.refresh_observed(&session, "/child", 0).unwrap();
client
.complete(
completed_newer_observation.id,
@@ -1171,8 +1362,8 @@ fn overlapping_tree_work_preserves_newer_subject_request_ownership() {
Some(b"completed-newer".to_vec())
);
let older_omitting_tree = client.prepare_observe(&session, "/", 1).unwrap();
let newer_observation = client.prepare_observe(&session, "/child", 0).unwrap();
let older_omitting_tree = client.refresh_observed(&session, "/", 1).unwrap();
let newer_observation = client.refresh_observed(&session, "/child", 0).unwrap();
client
.complete(
newer_observation.id,
@@ -1199,8 +1390,8 @@ fn overlapping_tree_work_preserves_newer_subject_request_ownership() {
assert_eq!(child.state, ObservationState::Fresh);
assert_eq!(child.validator, Some(b"newer-than-omission".to_vec()));
let old_tree = client.prepare_observe(&session, "/", 1).unwrap();
let newer_child_tree = client.prepare_observe(&session, "/child", 1).unwrap();
let old_tree = client.refresh_observed(&session, "/", 1).unwrap();
let newer_child_tree = client.refresh_observed(&session, "/child", 1).unwrap();
client
.complete(
old_tree.id,
@@ -1241,8 +1432,8 @@ fn overlapping_tree_work_preserves_newer_subject_request_ownership() {
ObservationState::Fresh
);
let older_parent = client.prepare_observe(&session, "/", 1).unwrap();
let newer_child = client.prepare_observe(&session, "/child", 1).unwrap();
let older_parent = client.refresh_observed(&session, "/", 1).unwrap();
let newer_child = client.refresh_observed(&session, "/child", 1).unwrap();
client
.complete(
newer_child.id,
@@ -1281,8 +1472,8 @@ fn superseded_transport_failure_does_not_poison_newer_refresh() {
let session = client
.open_session("https://example.test/wip", SecurityContext::new("subject"))
.unwrap();
let old = client.prepare_observe(&session, "/item", 0).unwrap();
let newer = client.prepare_observe(&session, "/item", 0).unwrap();
let old = client.refresh_observed(&session, "/item", 0).unwrap();
let newer = client.refresh_observed(&session, "/item", 0).unwrap();
assert_eq!(
client
.fail_transport(old.id, "old connection failed")
@@ -1300,8 +1491,8 @@ fn superseded_transport_failure_does_not_poison_newer_refresh() {
)
.unwrap();
let old = client.prepare_observe(&session, "/", 1).unwrap();
let newer_tree = client.prepare_observe(&session, "/", 1).unwrap();
let old = client.refresh_observed(&session, "/", 1).unwrap();
let newer_tree = client.refresh_observed(&session, "/", 1).unwrap();
assert_eq!(
client
.fail_transport(old.id, "old tree connection failed")
@@ -1410,10 +1601,10 @@ fn entry_eviction_never_discards_a_loading_tree_subject() {
.open_session("https://example.test/wip", SecurityContext::new("subject"))
.unwrap();
observe_object(&mut client, &session, "/", object("", None, b"root"));
let tree = client.prepare_observe(&session, "/", 1).unwrap();
let tree = client.refresh_observed(&session, "/", 1).unwrap();
assert_eq!(
client
.prepare_observe(&session, "/other", 0)
.refresh_observed(&session, "/other", 0)
.unwrap_err()
.kind,
ClientErrorKind::Capacity
@@ -1454,7 +1645,7 @@ fn failed_entry_and_tree_observation_metadata_is_bounded() {
.open_session("https://example.test/wip", SecurityContext::new("subject"))
.unwrap();
for path in ["/one", "/two", "/three"] {
let request = client.prepare_observe(&session, path, 0).unwrap();
let request = client.refresh_observed(&session, path, 0).unwrap();
assert_eq!(
client
.fail_transport(request.id, "retrieval failed")
@@ -1468,7 +1659,7 @@ fn failed_entry_and_tree_observation_metadata_is_bounded() {
assert!(client.object(&session, "/three").is_some());
for path in ["/alpha", "/beta", "/gamma"] {
let request = client.prepare_observe(&session, path, 1).unwrap();
let request = client.refresh_observed(&session, path, 1).unwrap();
assert_eq!(
client
.fail_transport(request.id, "tree failed")
@@ -1487,8 +1678,8 @@ fn in_flight_and_history_are_bounded() {
let session = client
.open_session("https://example.test/wip", SecurityContext::new("subject"))
.unwrap();
let first = client.prepare_observe(&session, "/one", 0).unwrap();
let error = client.prepare_observe(&session, "/two", 0).unwrap_err();
let first = client.refresh_observed(&session, "/one", 0).unwrap();
let error = client.refresh_observed(&session, "/two", 0).unwrap_err();
assert_eq!(error.kind, ClientErrorKind::Capacity);
client
.complete(
@@ -1498,8 +1689,8 @@ fn in_flight_and_history_are_bounded() {
.unwrap();
// Diagnostic history evicts oldest entries at the configured bound.
let stale_old = client.prepare_observe(&session, "/one", 0).unwrap();
let stale_new = client.prepare_observe(&session, "/one", 0).unwrap_err();
let stale_old = client.refresh_observed(&session, "/one", 0).unwrap();
let stale_new = client.refresh_observed(&session, "/one", 0).unwrap_err();
assert_eq!(stale_new.kind, ClientErrorKind::Capacity);
client
.complete(