feat(client): add stateful WIP runtime
This commit is contained in:
@@ -0,0 +1,517 @@
|
||||
use std::collections::BTreeMap;
|
||||
|
||||
use wip_client::{
|
||||
CallOutcome, Client, ClientErrorKind, ClientLimits, Completion, DispatchedTransportFailure,
|
||||
ObservationState, OutcomeUnknownReason, SecurityContext,
|
||||
};
|
||||
use wip_https::http::Response;
|
||||
use wip_https::{
|
||||
Limits, encode_call_operation_response, encode_fetch_interface_response, encode_fetch_response,
|
||||
encode_fetch_tree_response, encode_protocol_error_response,
|
||||
};
|
||||
use wip_protocol::{
|
||||
CallOperationResponse, FetchInterfaceRequest, FetchInterfaceResponse, FetchRequest,
|
||||
FetchTreeRequest, FetchTreeResponse, INTERFACE_FORMAT_V1, InterfaceDescriptor, Object,
|
||||
ObjectTreeNode, OperationDeclaration, ParameterDeclaration, ProtocolError, ProtocolErrorCode,
|
||||
ProtocolInteraction, ReturnDeclaration, TypeExpr, Value,
|
||||
};
|
||||
|
||||
fn wire_limits() -> Limits {
|
||||
Limits::new(64 * 1024, 64 * 1024, 64, 64).unwrap()
|
||||
}
|
||||
|
||||
fn client(objects: usize, interfaces: usize, in_flight: usize, history: usize) -> Client {
|
||||
Client::new(
|
||||
ClientLimits::new(8, objects, interfaces, in_flight, history).unwrap(),
|
||||
wire_limits(),
|
||||
)
|
||||
}
|
||||
|
||||
fn object(name: &str, reference: Option<&str>, validator: &[u8]) -> Object {
|
||||
Object {
|
||||
name: name.into(),
|
||||
description: Some(format!("object {name}")),
|
||||
interfaces: vec!["math".into()],
|
||||
r#ref: reference.map(str::to_owned),
|
||||
validator: Some(validator.to_vec()),
|
||||
}
|
||||
}
|
||||
|
||||
fn descriptor(return_type: TypeExpr) -> InterfaceDescriptor {
|
||||
InterfaceDescriptor {
|
||||
format: INTERFACE_FORMAT_V1.into(),
|
||||
documentation: None,
|
||||
types: vec![],
|
||||
operations: vec![OperationDeclaration {
|
||||
name: "calculate".into(),
|
||||
documentation: None,
|
||||
parameters: vec![ParameterDeclaration {
|
||||
name: "value".into(),
|
||||
required: true,
|
||||
documentation: None,
|
||||
r#type: TypeExpr::Integer,
|
||||
}],
|
||||
returns: ReturnDeclaration {
|
||||
documentation: None,
|
||||
r#type: return_type,
|
||||
},
|
||||
}],
|
||||
}
|
||||
}
|
||||
|
||||
fn fetch_response(path: &str, value: &Object) -> Response<Vec<u8>> {
|
||||
encode_fetch_response(&FetchRequest { path: path.into() }, value, wire_limits()).unwrap()
|
||||
}
|
||||
|
||||
fn observe_object(client: &mut Client, session: &wip_client::SessionId, path: &str, value: Object) {
|
||||
let request = client.prepare_fetch(session, path).unwrap();
|
||||
client
|
||||
.complete(request.id, fetch_response(path, &value))
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
fn observe_interface(
|
||||
client: &mut Client,
|
||||
session: &wip_client::SessionId,
|
||||
value: FetchInterfaceResponse,
|
||||
) {
|
||||
let request = client
|
||||
.prepare_interface(session, value.interface.clone())
|
||||
.unwrap();
|
||||
let response = encode_fetch_interface_response(
|
||||
&FetchInterfaceRequest {
|
||||
interface: value.interface.clone(),
|
||||
},
|
||||
&value,
|
||||
wire_limits(),
|
||||
)
|
||||
.unwrap();
|
||||
client.complete(request.id, response).unwrap();
|
||||
}
|
||||
|
||||
fn args(value: i64) -> BTreeMap<String, Value> {
|
||||
BTreeMap::from([("value".into(), Value::Integer(value))])
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sessions_partition_known_space_by_canonical_endpoint_and_security_context() {
|
||||
let mut client = client(8, 8, 8, 8);
|
||||
let alice = client
|
||||
.open_session(
|
||||
"https://example.test/wip",
|
||||
SecurityContext::new("alice-token"),
|
||||
)
|
||||
.unwrap();
|
||||
let alice_canonical = client
|
||||
.open_session(
|
||||
"https://example.test/wip/",
|
||||
SecurityContext::new("alice-token"),
|
||||
)
|
||||
.unwrap();
|
||||
let bob = client
|
||||
.open_session(
|
||||
"https://example.test/wip",
|
||||
SecurityContext::new("bob-token"),
|
||||
)
|
||||
.unwrap();
|
||||
let other = client
|
||||
.open_session(
|
||||
"https://other.test/wip",
|
||||
SecurityContext::new("alice-token"),
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(alice, alice_canonical);
|
||||
observe_object(&mut client, &alice, "/item", object("item", None, b"a"));
|
||||
assert!(client.object(&alice, "/item").is_some());
|
||||
assert!(client.object(&bob, "/item").is_none());
|
||||
assert!(client.object(&other, "/item").is_none());
|
||||
}
|
||||
|
||||
#[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_fetch(&session, "/item").unwrap();
|
||||
let new = client.prepare_fetch(&session, "/item").unwrap();
|
||||
|
||||
assert_eq!(
|
||||
client
|
||||
.complete(
|
||||
old.id,
|
||||
fetch_response("/item", &object("item", None, b"old"))
|
||||
)
|
||||
.unwrap(),
|
||||
Completion::StaleResponseRejected(old.id)
|
||||
);
|
||||
assert!(matches!(
|
||||
client.object(&session, "/item").unwrap().state,
|
||||
ObservationState::Loading(id) if id == new.id
|
||||
));
|
||||
|
||||
client
|
||||
.complete(
|
||||
new.id,
|
||||
fetch_response("/item", &object("item", None, b"new")),
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
client.object(&session, "/item").unwrap().validator,
|
||||
Some(b"new".to_vec())
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn normalized_object_cache_rejects_conflicts_and_evicts_to_its_bound() {
|
||||
let mut client = client(2, 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"v1"),
|
||||
);
|
||||
observe_object(
|
||||
&mut client,
|
||||
&session,
|
||||
"/b/shared",
|
||||
object("shared", Some("shared-ref"), b"v1"),
|
||||
);
|
||||
|
||||
let conflict = client.prepare_fetch(&session, "/b/shared").unwrap();
|
||||
let mut changed = object("shared", Some("shared-ref"), b"v1");
|
||||
changed.description = Some("contradiction".into());
|
||||
let error = client
|
||||
.complete(conflict.id, fetch_response("/b/shared", &changed))
|
||||
.unwrap_err();
|
||||
assert_eq!(error.kind, ClientErrorKind::InvalidResponse);
|
||||
|
||||
observe_object(
|
||||
&mut client,
|
||||
&session,
|
||||
"/second",
|
||||
object("second", None, b"2"),
|
||||
);
|
||||
observe_object(&mut client, &session, "/third", object("third", None, b"3"));
|
||||
assert!(client.object(&session, "/a/shared").is_none());
|
||||
assert!(client.object(&session, "/second").is_some());
|
||||
assert!(client.object(&session, "/third").is_some());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn tree_expansion_is_lazy_truncated_and_refreshes_edges() {
|
||||
let mut client = client(16, 8, 8, 8);
|
||||
let session = client
|
||||
.open_session("https://example.test/wip", SecurityContext::new("subject"))
|
||||
.unwrap();
|
||||
assert!(client.tree(&session, "/").is_none());
|
||||
|
||||
let request = client.prepare_tree(&session, "/", 1, 4).unwrap();
|
||||
let logical_request = FetchTreeRequest {
|
||||
path: "/".into(),
|
||||
depth: 1,
|
||||
limit: 4,
|
||||
};
|
||||
let tree = FetchTreeResponse {
|
||||
root: ObjectTreeNode {
|
||||
object: object("", None, b"root"),
|
||||
children: vec![ObjectTreeNode {
|
||||
object: object("items", None, b"items"),
|
||||
children: vec![],
|
||||
}],
|
||||
},
|
||||
truncated: true,
|
||||
};
|
||||
client
|
||||
.complete(
|
||||
request.id,
|
||||
encode_fetch_tree_response(&logical_request, &tree, wire_limits()).unwrap(),
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(client.tree(&session, "/").unwrap().children, ["/items"]);
|
||||
assert!(client.tree(&session, "/").unwrap().truncated);
|
||||
assert!(client.object(&session, "/items").is_some());
|
||||
|
||||
let refresh = client.prepare_tree(&session, "/", 1, 4).unwrap();
|
||||
let refreshed = FetchTreeResponse {
|
||||
root: ObjectTreeNode {
|
||||
object: object("", None, b"root-2"),
|
||||
children: vec![],
|
||||
},
|
||||
truncated: false,
|
||||
};
|
||||
client
|
||||
.complete(
|
||||
refresh.id,
|
||||
encode_fetch_tree_response(&logical_request, &refreshed, wire_limits()).unwrap(),
|
||||
)
|
||||
.unwrap();
|
||||
assert!(client.tree(&session, "/").unwrap().children.is_empty());
|
||||
assert!(!client.tree(&session, "/").unwrap().truncated);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn call_context_fixes_membership_validators_descriptor_and_result_validation() {
|
||||
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,
|
||||
"/item",
|
||||
object("item", Some("item-ref"), b"object-v1"),
|
||||
);
|
||||
observe_interface(
|
||||
&mut client,
|
||||
&session,
|
||||
FetchInterfaceResponse {
|
||||
interface: "math".into(),
|
||||
descriptor: descriptor(TypeExpr::Integer),
|
||||
validator: Some(b"interface-v1".to_vec()),
|
||||
},
|
||||
);
|
||||
|
||||
let call = client
|
||||
.prepare_call(&session, "/item", "math", "calculate", args(7))
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
client
|
||||
.pending_call_context(call.id)
|
||||
.unwrap()
|
||||
.operation()
|
||||
.name,
|
||||
"calculate"
|
||||
);
|
||||
client.mark_dispatched(call.id).unwrap();
|
||||
|
||||
// A newer interface observation cannot change this in-flight call's schema.
|
||||
observe_interface(
|
||||
&mut client,
|
||||
&session,
|
||||
FetchInterfaceResponse {
|
||||
interface: "math".into(),
|
||||
descriptor: descriptor(TypeExpr::String),
|
||||
validator: Some(b"interface-v2".to_vec()),
|
||||
},
|
||||
);
|
||||
|
||||
let history_context = match client
|
||||
.complete(
|
||||
call.id,
|
||||
encode_call_operation_response(
|
||||
&wip_protocol::CallOperationRequest {
|
||||
target: wip_protocol::Target {
|
||||
path: "/item".into(),
|
||||
validator: Some(b"object-v1".to_vec()),
|
||||
},
|
||||
interface: wip_protocol::InterfaceTarget {
|
||||
reference: "math".into(),
|
||||
validator: Some(b"interface-v1".to_vec()),
|
||||
},
|
||||
operation: "calculate".into(),
|
||||
arguments: args(7),
|
||||
},
|
||||
&descriptor(TypeExpr::Integer),
|
||||
&CallOperationResponse {
|
||||
result: Value::Integer(14),
|
||||
validator: Some(b"object-v2".to_vec()),
|
||||
},
|
||||
wire_limits(),
|
||||
)
|
||||
.unwrap(),
|
||||
)
|
||||
.unwrap()
|
||||
{
|
||||
Completion::Call(record) => record,
|
||||
other => panic!("unexpected completion: {other:?}"),
|
||||
};
|
||||
assert_eq!(
|
||||
history_context.context.descriptor(),
|
||||
&descriptor(TypeExpr::Integer)
|
||||
);
|
||||
assert_eq!(
|
||||
history_context.context.object().validator,
|
||||
Some(b"object-v1".to_vec())
|
||||
);
|
||||
assert_eq!(
|
||||
history_context.context.interface().validator,
|
||||
Some(b"interface-v1".to_vec())
|
||||
);
|
||||
assert!(matches!(history_context.outcome, CallOutcome::Success(_)));
|
||||
assert_eq!(
|
||||
client.interface(&session, "math").unwrap().descriptor,
|
||||
Some(descriptor(TypeExpr::String))
|
||||
);
|
||||
assert_eq!(
|
||||
client.object(&session, "/item").unwrap().state,
|
||||
ObservationState::Stale
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn validator_mismatch_stales_only_its_observation_and_never_retries() {
|
||||
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, "/item", object("item", None, b"v1"));
|
||||
observe_interface(
|
||||
&mut client,
|
||||
&session,
|
||||
FetchInterfaceResponse {
|
||||
interface: "math".into(),
|
||||
descriptor: descriptor(TypeExpr::Integer),
|
||||
validator: Some(b"i1".to_vec()),
|
||||
},
|
||||
);
|
||||
let call = client
|
||||
.prepare_call(&session, "/item", "math", "calculate", args(1))
|
||||
.unwrap();
|
||||
client.mark_dispatched(call.id).unwrap();
|
||||
let error = ProtocolError {
|
||||
code: ProtocolErrorCode::InterfaceValidatorMismatch,
|
||||
message: "stale descriptor".into(),
|
||||
};
|
||||
let response =
|
||||
encode_protocol_error_response(ProtocolInteraction::CallOperation, &error, wire_limits())
|
||||
.unwrap();
|
||||
client.complete(call.id, response).unwrap();
|
||||
|
||||
assert_eq!(
|
||||
client.interface(&session, "math").unwrap().state,
|
||||
ObservationState::Stale
|
||||
);
|
||||
assert_eq!(
|
||||
client.object(&session, "/item").unwrap().state,
|
||||
ObservationState::Fresh
|
||||
);
|
||||
assert_eq!(client.in_flight_len(), 0);
|
||||
assert_eq!(client.call_history(&session).unwrap().len(), 1);
|
||||
|
||||
// Reobservation changes cache state but does not manufacture a replacement call.
|
||||
observe_interface(
|
||||
&mut client,
|
||||
&session,
|
||||
FetchInterfaceResponse {
|
||||
interface: "math".into(),
|
||||
descriptor: descriptor(TypeExpr::Integer),
|
||||
validator: Some(b"i2".to_vec()),
|
||||
},
|
||||
);
|
||||
assert_eq!(client.in_flight_len(), 0);
|
||||
assert_eq!(client.call_history(&session).unwrap().len(), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn protocol_and_transport_unknown_outcomes_stale_without_retry() {
|
||||
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, "/item", object("item", None, b"v1"));
|
||||
observe_interface(
|
||||
&mut client,
|
||||
&session,
|
||||
FetchInterfaceResponse {
|
||||
interface: "math".into(),
|
||||
descriptor: descriptor(TypeExpr::Integer),
|
||||
validator: None,
|
||||
},
|
||||
);
|
||||
|
||||
let call = client
|
||||
.prepare_call(&session, "/item", "math", "calculate", args(1))
|
||||
.unwrap();
|
||||
client.mark_dispatched(call.id).unwrap();
|
||||
let unknown = ProtocolError {
|
||||
code: ProtocolErrorCode::OperationOutcomeUnknown,
|
||||
message: "commit uncertain".into(),
|
||||
};
|
||||
let completion = client
|
||||
.complete(
|
||||
call.id,
|
||||
encode_protocol_error_response(
|
||||
ProtocolInteraction::CallOperation,
|
||||
&unknown,
|
||||
wire_limits(),
|
||||
)
|
||||
.unwrap(),
|
||||
)
|
||||
.unwrap();
|
||||
assert!(matches!(
|
||||
completion,
|
||||
Completion::Call(record)
|
||||
if matches!(
|
||||
record.outcome,
|
||||
CallOutcome::Unknown {
|
||||
reason: OutcomeUnknownReason::Protocol,
|
||||
..
|
||||
}
|
||||
)
|
||||
));
|
||||
assert_eq!(
|
||||
client.object(&session, "/item").unwrap().state,
|
||||
ObservationState::Stale
|
||||
);
|
||||
|
||||
observe_object(&mut client, &session, "/item", object("item", None, b"v2"));
|
||||
let second = client
|
||||
.prepare_call(&session, "/item", "math", "calculate", args(2))
|
||||
.unwrap();
|
||||
client.mark_dispatched(second.id).unwrap();
|
||||
let completion = client
|
||||
.fail_dispatched_call(
|
||||
second.id,
|
||||
DispatchedTransportFailure::Timeout,
|
||||
"deadline elapsed",
|
||||
)
|
||||
.unwrap();
|
||||
assert!(matches!(
|
||||
completion,
|
||||
Completion::Call(record)
|
||||
if matches!(
|
||||
record.outcome,
|
||||
CallOutcome::Unknown {
|
||||
reason: OutcomeUnknownReason::Timeout,
|
||||
..
|
||||
}
|
||||
)
|
||||
));
|
||||
assert_eq!(client.in_flight_len(), 0);
|
||||
assert_eq!(client.call_history(&session).unwrap().len(), 2);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn in_flight_and_history_are_bounded() {
|
||||
let mut client = client(8, 8, 1, 1);
|
||||
let session = client
|
||||
.open_session("https://example.test/wip", SecurityContext::new("subject"))
|
||||
.unwrap();
|
||||
let first = client.prepare_fetch(&session, "/one").unwrap();
|
||||
let error = client.prepare_fetch(&session, "/two").unwrap_err();
|
||||
assert_eq!(error.kind, ClientErrorKind::Capacity);
|
||||
client
|
||||
.complete(first.id, fetch_response("/one", &object("one", None, b"1")))
|
||||
.unwrap();
|
||||
|
||||
// Diagnostic history evicts oldest entries at the configured bound.
|
||||
let stale_old = client.prepare_fetch(&session, "/one").unwrap();
|
||||
let stale_new = client.prepare_fetch(&session, "/one").unwrap_err();
|
||||
assert_eq!(stale_new.kind, ClientErrorKind::Capacity);
|
||||
client
|
||||
.complete(
|
||||
stale_old.id,
|
||||
Response::builder()
|
||||
.status(200)
|
||||
.header("content-type", "application/json")
|
||||
.body(b"not json".to_vec())
|
||||
.unwrap(),
|
||||
)
|
||||
.unwrap_err();
|
||||
assert_eq!(client.diagnostics(&session).unwrap().len(), 1);
|
||||
}
|
||||
Reference in New Issue
Block a user