feat: prefetch visible interface descriptors
This commit is contained in:
@@ -751,7 +751,34 @@ impl Client {
|
||||
})
|
||||
}
|
||||
|
||||
/// Starts or refreshes an interface descriptor observation.
|
||||
/// Ensures that an interface descriptor has been observed without issuing
|
||||
/// duplicate work or evicting another cached descriptor.
|
||||
///
|
||||
/// Returns `None` when the descriptor is already fresh, is currently loading,
|
||||
/// is blocked by retained stale or failure state that requires an explicit
|
||||
/// refresh, or when either the interface cache or bounded in-flight capacity is
|
||||
/// currently full. Callers may invoke this method again after applying another
|
||||
/// completion.
|
||||
pub fn ensure_interface(
|
||||
&mut self,
|
||||
session: &SessionId,
|
||||
reference: impl Into<String>,
|
||||
) -> Result<Option<PreparedRequest>, ClientError> {
|
||||
let reference = reference.into();
|
||||
let state = self.session(session)?;
|
||||
if state.interfaces.contains_key(&reference)
|
||||
|| state.interfaces.len() >= self.limits.interfaces_per_session
|
||||
|| self.pending.len() >= self.limits.in_flight
|
||||
{
|
||||
return Ok(None);
|
||||
}
|
||||
self.prepare_interface(session, reference).map(Some)
|
||||
}
|
||||
|
||||
/// Explicitly starts or refreshes an interface descriptor observation.
|
||||
///
|
||||
/// Unlike [`Client::ensure_interface`], this always creates a new request and
|
||||
/// may evict an older non-loading descriptor when the interface cache is full.
|
||||
pub fn prepare_interface(
|
||||
&mut self,
|
||||
session: &SessionId,
|
||||
|
||||
@@ -288,6 +288,117 @@ fn ensure_observed_respects_capacity_and_does_not_loop_on_failure() {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ensure_interface_deduplicates_loading_fresh_and_retained_failure() {
|
||||
let mut client = client(8, 8, 8, 8);
|
||||
let session = client
|
||||
.open_session("https://example.test/wip", SecurityContext::new("subject"))
|
||||
.unwrap();
|
||||
|
||||
let request = client
|
||||
.ensure_interface(&session, "math")
|
||||
.unwrap()
|
||||
.expect("missing descriptor starts a fetch");
|
||||
assert!(client.ensure_interface(&session, "math").unwrap().is_none());
|
||||
let value = FetchInterfaceResponse {
|
||||
interface: "math".into(),
|
||||
descriptor: descriptor(TypeExpr::Integer),
|
||||
validator: Some(b"math-v1".to_vec()),
|
||||
};
|
||||
client
|
||||
.complete(
|
||||
request.id,
|
||||
encode_fetch_interface_response(
|
||||
&FetchInterfaceRequest {
|
||||
interface: "math".into(),
|
||||
},
|
||||
&value,
|
||||
wire_limits(),
|
||||
)
|
||||
.unwrap(),
|
||||
)
|
||||
.unwrap();
|
||||
assert!(client.ensure_interface(&session, "math").unwrap().is_none());
|
||||
|
||||
let failed = client
|
||||
.ensure_interface(&session, "broken")
|
||||
.unwrap()
|
||||
.expect("new descriptor starts a fetch");
|
||||
assert_eq!(
|
||||
client
|
||||
.fail_transport(failed.id, "scripted failure")
|
||||
.unwrap_err()
|
||||
.kind,
|
||||
ClientErrorKind::Transport
|
||||
);
|
||||
assert!(
|
||||
client
|
||||
.ensure_interface(&session, "broken")
|
||||
.unwrap()
|
||||
.is_none(),
|
||||
"retained interface failures require an explicit refresh"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ensure_interface_waits_for_request_and_cache_capacity_without_evicting() {
|
||||
let mut client = client(8, 1, 1, 8);
|
||||
let session = client
|
||||
.open_session("https://example.test/wip", SecurityContext::new("subject"))
|
||||
.unwrap();
|
||||
let pending_object = client.refresh_object(&session, "/busy").unwrap();
|
||||
assert!(
|
||||
client.ensure_interface(&session, "math").unwrap().is_none(),
|
||||
"lookahead waits while the in-flight bound is full"
|
||||
);
|
||||
client
|
||||
.complete(
|
||||
pending_object.id,
|
||||
observation_response("/busy", &object("busy", None, b"busy")),
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
let request = client
|
||||
.ensure_interface(&session, "math")
|
||||
.unwrap()
|
||||
.expect("lookahead resumes after capacity becomes available");
|
||||
let value = FetchInterfaceResponse {
|
||||
interface: "math".into(),
|
||||
descriptor: descriptor(TypeExpr::Integer),
|
||||
validator: None,
|
||||
};
|
||||
client
|
||||
.complete(
|
||||
request.id,
|
||||
encode_fetch_interface_response(
|
||||
&FetchInterfaceRequest {
|
||||
interface: "math".into(),
|
||||
},
|
||||
&value,
|
||||
wire_limits(),
|
||||
)
|
||||
.unwrap(),
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
assert!(
|
||||
client
|
||||
.ensure_interface(&session, "other")
|
||||
.unwrap()
|
||||
.is_none(),
|
||||
"lookahead does not churn a full descriptor cache"
|
||||
);
|
||||
assert!(client.interface(&session, "math").is_some());
|
||||
assert!(client.interface(&session, "other").is_none());
|
||||
|
||||
let explicit = client.prepare_interface(&session, "other").unwrap();
|
||||
assert!(client.interface(&session, "math").is_none());
|
||||
assert!(matches!(
|
||||
client.interface(&session, "other").unwrap().state,
|
||||
ObservationState::Loading(id) if id == explicit.id
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn newer_refresh_rejects_out_of_order_response_for_the_same_subject() {
|
||||
let mut client = client(8, 8, 8, 8);
|
||||
|
||||
Reference in New Issue
Block a user