fn root_reference(name: &str) -> wip_protocol::InterfaceReference { wip_protocol::InterfaceReference { scope: "/".into(), name: name.into(), } } use std::collections::BTreeMap; use wip_client::{ CallOutcome, Client, ClientErrorKind, ClientLimits, Completion, DispatchedTransportFailure, ObservationState, OutcomeUnknownReason, SecurityContext, }; use wip_http::http::Response; use wip_http::{ EncodeHttpError, Limits, encode_call_operation_response, encode_fetch_interface_response, encode_observe_response, encode_protocol_error_response, }; use wip_protocol::{ CallOperationResponse, FetchInterfaceRequest, FetchInterfaceResponse, INTERFACE_FORMAT_V1, InterfaceDescriptor, Object, ObjectObservation, ObserveRequest, OperationDeclaration, ParameterDeclaration, ProtocolError, ProtocolErrorCode, ProtocolInteraction, ReturnDeclaration, TypeExpr, Value, }; fn wire_limits() -> Limits { Limits::new(64 * 1024, 64 * 1024, 64).unwrap() } #[derive(Clone)] struct TestObservation { object: Object, children: Vec, } struct TestObservationResponse { root: TestObservation, } fn protocol_observation(node: &TestObservation, depth: u32) -> ObjectObservation { ObjectObservation { object: node.object.clone(), children: (depth > 0).then(|| { node.children .iter() .map(|child| protocol_observation(child, depth - 1)) .collect() }), } } fn encode_test_observation_response( request: &ObserveRequest, value: &TestObservationResponse, limits: Limits, ) -> Result>, EncodeHttpError> { encode_observe_response( request, &protocol_observation(&value.root, request.depth), limits, ) } 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![root_reference("math")], 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 observation_response(path: &str, value: &Object) -> Response> { encode_observe_response( &ObserveRequest { path: path.into(), depth: 0, }, &ObjectObservation { object: value.clone(), children: None, }, wire_limits(), ) .unwrap() } fn observe_object(client: &mut Client, session: &wip_client::SessionId, path: &str, value: Object) { let request = client.refresh_observed(session, path, 0).unwrap(); client .complete(request.id, observation_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 { 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 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 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, root_reference("math")) .unwrap() .expect("missing descriptor starts a fetch"); assert!( client .ensure_interface(&session, root_reference("math")) .unwrap() .is_none() ); let value = FetchInterfaceResponse { scope_ref: None, interface: root_reference("math"), descriptor: descriptor(TypeExpr::Integer), validator: Some(b"math-v1".to_vec()), }; client .complete( request.id, encode_fetch_interface_response( &FetchInterfaceRequest { interface: root_reference("math"), }, &value, wire_limits(), ) .unwrap(), ) .unwrap(); assert!( client .ensure_interface(&session, root_reference("math")) .unwrap() .is_none() ); let failed = client .ensure_interface(&session, root_reference("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, root_reference("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, root_reference("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, root_reference("math")) .unwrap() .expect("lookahead resumes after capacity becomes available"); let value = FetchInterfaceResponse { scope_ref: None, interface: root_reference("math"), descriptor: descriptor(TypeExpr::Integer), validator: None, }; client .complete( request.id, encode_fetch_interface_response( &FetchInterfaceRequest { interface: root_reference("math"), }, &value, wire_limits(), ) .unwrap(), ) .unwrap(); assert!( client .ensure_interface(&session, root_reference("other")) .unwrap() .is_none(), "lookahead does not churn a full descriptor cache" ); assert!( client .interface(&session, &root_reference("math")) .is_some() ); assert!( client .interface(&session, &root_reference("other")) .is_none() ); let explicit = client .prepare_interface(&session, root_reference("other")) .unwrap(); assert!( client .interface(&session, &root_reference("math")) .is_none() ); assert!(matches!( client.interface(&session, &root_reference("other")).unwrap().state, ObservationState::Loading(id) if id == explicit.id )); } #[test] fn independent_retrievals_run_concurrently_and_complete_out_of_order() { let mut client = client(8, 8, 8, 8); let session = client .open_session("https://example.test/wip", SecurityContext::new("subject")) .unwrap(); let first = client.refresh_observed(&session, "/one", 0).unwrap(); let second = client.refresh_observed(&session, "/two", 0).unwrap(); let interface = client .prepare_interface(&session, root_reference("math")) .unwrap(); assert_eq!(first.request.uri().path(), "/wip/v1/observe"); assert_eq!(second.request.uri().path(), "/wip/v1/observe"); assert_eq!(interface.request.uri().path(), "/wip/v1/fetch_interface"); assert_eq!(client.in_flight_len(), 3); client .complete( second.id, observation_response("/two", &object("two", None, b"two")), ) .unwrap(); client .complete( interface.id, encode_fetch_interface_response( &FetchInterfaceRequest { interface: root_reference("math"), }, &FetchInterfaceResponse { scope_ref: None, interface: root_reference("math"), descriptor: descriptor(TypeExpr::Integer), validator: Some(b"math-v1".to_vec()), }, wire_limits(), ) .unwrap(), ) .unwrap(); client .complete( first.id, observation_response("/one", &object("one", None, b"one")), ) .unwrap(); assert_eq!( client.object(&session, "/one").unwrap().state, ObservationState::Fresh ); assert_eq!( client.object(&session, "/two").unwrap().state, ObservationState::Fresh ); assert_eq!( client .interface(&session, &root_reference("math")) .unwrap() .state, ObservationState::Fresh ); assert_eq!(client.in_flight_len(), 0); } #[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.refresh_observed(&session, "/item", 0).unwrap(); let new = client.refresh_observed(&session, "/item", 0).unwrap(); assert_eq!( client .complete( old.id, observation_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, observation_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.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 .complete(conflict.id, observation_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 normalized_aliases_preserve_newer_request_ownership() { 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, "/a/shared", object("shared", Some("shared-ref"), b"initial"), ); observe_object( &mut client, &session, "/b/shared", object("shared", Some("shared-ref"), b"initial"), ); let request = ObserveRequest { path: "/a".into(), depth: 1, }; let older_tree_response = || TestObservationResponse { root: TestObservation { object: object("a", None, b"root"), children: vec![TestObservation { object: object("shared", Some("shared-ref"), b"older"), children: vec![], }], }, }; 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, observation_response("/b/shared", &object("shared", Some("shared-ref"), b"newer")), ) .unwrap(); client .complete( older_tree.id, encode_test_observation_response(&request, &older_tree_response(), wire_limits()) .unwrap(), ) .unwrap(); assert_eq!( client.object(&session, "/a/shared").unwrap().validator, Some(b"newer".to_vec()) ); assert_eq!( client.object(&session, "/b/shared").unwrap().validator, Some(b"newer".to_vec()) ); 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, encode_test_observation_response(&request, &older_tree_response(), wire_limits()) .unwrap(), ) .unwrap(); assert!(matches!( client.object(&session, "/b/shared").unwrap().state, ObservationState::Loading(id) if id == newer_alias.id )); assert_eq!( client.object(&session, "/b/shared").unwrap().validator, Some(b"newer".to_vec()) ); client .complete( newer_alias.id, observation_response( "/b/shared", &object("shared", Some("shared-ref"), b"newest"), ), ) .unwrap(); assert_eq!( client.object(&session, "/a/shared").unwrap().validator, Some(b"newest".to_vec()) ); 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, observation_response( "/b/shared", &object("shared", Some("shared-ref"), b"newest"), ), ) .unwrap(); let mut contradictory = object("shared", Some("shared-ref"), b"newest"); contradictory.description = Some("contradictory representation".into()); let error = client .complete( older_tree.id, encode_test_observation_response( &request, &TestObservationResponse { root: TestObservation { object: object("a", None, b"root"), children: vec![TestObservation { object: contradictory, children: vec![], }], }, }, wire_limits(), ) .unwrap(), ) .unwrap_err(); assert_eq!(error.kind, ClientErrorKind::InvalidResponse); assert_eq!( client.object(&session, "/b/shared").unwrap().validator, Some(b"newest".to_vec()) ); observe_object( &mut client, &session, "/child", object("child", Some("child-ref"), b"stable"), ); let older_tree = client.refresh_observed(&session, "/", 1).unwrap(); let newer_child = client.refresh_observed(&session, "/child", 0).unwrap(); client .complete( newer_child.id, observation_response("/child", &object("child", Some("child-ref"), b"stable")), ) .unwrap(); let mut contradictory = object("child", Some("child-ref"), b"stable"); contradictory.description = Some("same-path contradiction".into()); let root_request = ObserveRequest { path: "/".into(), depth: 1, }; let error = client .complete( older_tree.id, encode_test_observation_response( &root_request, &TestObservationResponse { root: TestObservation { object: object("", None, b"root"), children: vec![TestObservation { object: contradictory, children: vec![], }], }, }, wire_limits(), ) .unwrap(), ) .unwrap_err(); assert_eq!(error.kind, ClientErrorKind::InvalidResponse); assert_eq!( client.object(&session, "/child").unwrap().state, ObservationState::Fresh ); } #[test] fn observation_expansion_is_complete_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.refresh_observed(&session, "/", 1).unwrap(); let logical_request = ObserveRequest { path: "/".into(), depth: 1, }; let tree = TestObservationResponse { root: TestObservation { object: object("", None, b"root"), children: vec![TestObservation { object: object("items", None, b"items"), children: vec![], }], }, }; client .complete( request.id, encode_test_observation_response(&logical_request, &tree, wire_limits()).unwrap(), ) .unwrap(); assert_eq!(client.tree(&session, "/").unwrap().children, ["/items"]); assert!(client.object(&session, "/items").is_some()); let refresh = client.refresh_observed(&session, "/", 1).unwrap(); let refreshed = TestObservationResponse { root: TestObservation { object: object("", None, b"root-2"), children: vec![], }, }; client .complete( refresh.id, encode_test_observation_response(&logical_request, &refreshed, wire_limits()).unwrap(), ) .unwrap(); assert!(client.tree(&session, "/").unwrap().children.is_empty()); assert_eq!( client.object(&session, "/items").unwrap().state, ObservationState::Stale ); } #[test] fn depth_zero_observation_preserves_previously_observed_children() { let mut client = client(16, 8, 8, 8); let session = client .open_session("https://example.test/wip", SecurityContext::new("subject")) .unwrap(); let deep_request = ObserveRequest { path: "/".into(), depth: 1, }; let deep = client.refresh_observed(&session, "/", 1).unwrap(); client .complete( deep.id, encode_test_observation_response( &deep_request, &TestObservationResponse { root: TestObservation { object: object("", None, b"root-v1"), children: vec![TestObservation { object: object("child", None, b"child-v1"), children: vec![], }], }, }, wire_limits(), ) .unwrap(), ) .unwrap(); let shallow = client.refresh_observed(&session, "/", 0).unwrap(); client .complete( shallow.id, observation_response("/", &object("", None, b"root-v2")), ) .unwrap(); assert_eq!(client.tree(&session, "/").unwrap().children, ["/child"]); assert_eq!( client.object(&session, "/child").unwrap().state, ObservationState::Fresh ); assert_eq!( client.object(&session, "/").unwrap().validator, Some(b"root-v2".to_vec()) ); } #[test] fn shallow_observation_supersedes_loading_deep_observation_without_leaking_loading_state() { let mut client = client(16, 8, 8, 8); let session = client .open_session("https://example.test/wip", SecurityContext::new("subject")) .unwrap(); let deep_request = ObserveRequest { path: "/".into(), depth: 1, }; let deep_value = TestObservationResponse { root: TestObservation { object: object("", None, b"old-root"), children: vec![], }, }; 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 ); assert_eq!( client .complete( deep.id, encode_test_observation_response(&deep_request, &deep_value, wire_limits()) .unwrap(), ) .unwrap(), Completion::StaleResponseRejected(deep.id) ); client .complete( shallow.id, observation_response("/", &object("", None, b"new-root")), ) .unwrap(); assert_eq!( client.tree(&session, "/").unwrap().state, ObservationState::Stale ); } #[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); 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 { scope_ref: None, interface: root_reference("math"), descriptor: descriptor(TypeExpr::Integer), validator: Some(b"interface-v1".to_vec()), }, ); let call = client .prepare_call( &session, "/item", &root_reference("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 { scope_ref: None, interface: root_reference("math"), 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 { scope_ref: None, reference: root_reference("math"), 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, &root_reference("math")) .unwrap() .descriptor, Some(descriptor(TypeExpr::String)) ); let observation = client.object(&session, "/item").unwrap(); assert_eq!(observation.state, ObservationState::Fresh); assert_eq!(observation.validator, Some(b"object-v2".to_vec())); } #[test] fn successful_call_without_validator_preserves_observed_validator() { 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"object-v1"), ); observe_interface( &mut client, &session, FetchInterfaceResponse { scope_ref: None, interface: root_reference("math"), descriptor: descriptor(TypeExpr::Integer), validator: None, }, ); let call = client .prepare_call( &session, "/item", &root_reference("math"), "calculate", args(1), ) .unwrap(); let request = client .pending_call_context(call.id) .unwrap() .request() .clone(); let fixed_descriptor = client .pending_call_context(call.id) .unwrap() .descriptor() .clone(); client.mark_dispatched(call.id).unwrap(); client .complete( call.id, encode_call_operation_response( &request, &fixed_descriptor, &CallOperationResponse { result: Value::Integer(2), validator: None, }, wire_limits(), ) .unwrap(), ) .unwrap(); let observation = client.object(&session, "/item").unwrap(); assert_eq!(observation.state, ObservationState::Fresh); assert_eq!(observation.validator, Some(b"object-v1".to_vec())); let next = client .prepare_call( &session, "/item", &root_reference("math"), "calculate", args(2), ) .unwrap(); assert_eq!( client .pending_call_context(next.id) .unwrap() .request() .target .validator, Some(b"object-v1".to_vec()) ); } #[test] fn newer_observation_prevents_older_success_validator_overwrite() { 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"object-v1"), ); observe_interface( &mut client, &session, FetchInterfaceResponse { scope_ref: None, interface: root_reference("math"), descriptor: descriptor(TypeExpr::Integer), validator: None, }, ); let call = client .prepare_call( &session, "/item", &root_reference("math"), "calculate", args(1), ) .unwrap(); let request = client .pending_call_context(call.id) .unwrap() .request() .clone(); let fixed_descriptor = client .pending_call_context(call.id) .unwrap() .descriptor() .clone(); client.mark_dispatched(call.id).unwrap(); observe_object( &mut client, &session, "/item", object("item", None, b"object-v3"), ); client .complete( call.id, encode_call_operation_response( &request, &fixed_descriptor, &CallOperationResponse { result: Value::Integer(2), validator: Some(b"object-v2".to_vec()), }, wire_limits(), ) .unwrap(), ) .unwrap(); let observation = client.object(&session, "/item").unwrap(); assert_eq!(observation.state, ObservationState::Fresh); assert_eq!(observation.validator, Some(b"object-v3".to_vec())); } #[test] fn older_retrieval_cannot_overwrite_success_validator() { 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("shared-ref"), b"object-v1"), ); observe_interface( &mut client, &session, FetchInterfaceResponse { scope_ref: None, interface: root_reference("math"), descriptor: descriptor(TypeExpr::Integer), validator: Some(b"interface-v1".to_vec()), }, ); let older_alias = client.refresh_observed(&session, "/alias/item", 0).unwrap(); let call = client .prepare_call( &session, "/item", &root_reference("math"), "calculate", args(7), ) .unwrap(); let request = client .pending_call_context(call.id) .unwrap() .request() .clone(); let fixed_descriptor = client .pending_call_context(call.id) .unwrap() .descriptor() .clone(); client.mark_dispatched(call.id).unwrap(); client .complete( call.id, encode_call_operation_response( &request, &fixed_descriptor, &CallOperationResponse { result: Value::Integer(14), validator: Some(b"object-v2".to_vec()), }, wire_limits(), ) .unwrap(), ) .unwrap(); let current = client.object(&session, "/item").unwrap(); assert_eq!(current.state, ObservationState::Fresh); assert_eq!(current.validator, Some(b"object-v2".to_vec())); client .complete( older_alias.id, observation_response( "/alias/item", &object("item", Some("shared-ref"), b"object-v1"), ), ) .unwrap(); let alias = client.object(&session, "/alias/item").unwrap(); assert_eq!(alias.state, ObservationState::Stale); assert_eq!(alias.validator, Some(b"object-v2".to_vec())); assert_eq!( client .prepare_call( &session, "/alias/item", &root_reference("math"), "calculate", args(1) ) .unwrap_err() .kind, ClientErrorKind::MissingObservation ); observe_object( &mut client, &session, "/item", object("item", Some("shared-ref"), b"object-v2"), ); let older_alias = client .refresh_observed(&session, "/second/item", 0) .unwrap(); let call = client .prepare_call( &session, "/item", &root_reference("math"), "calculate", args(8), ) .unwrap(); client.mark_dispatched(call.id).unwrap(); let unknown = ProtocolError { code: ProtocolErrorCode::OperationOutcomeUnknown, message: "commit uncertain".into(), }; client .complete( call.id, encode_protocol_error_response( ProtocolInteraction::CallOperation, &unknown, wire_limits(), ) .unwrap(), ) .unwrap(); client .complete( older_alias.id, observation_response( "/second/item", &object("item", Some("shared-ref"), b"object-v2"), ), ) .unwrap(); assert_eq!( client.object(&session, "/second/item").unwrap().state, ObservationState::Stale ); } #[test] fn in_flight_calls_pin_target_identity_until_staleness_is_applied() { let mut client = client(1, 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("shared-ref"), b"object-v1"), ); observe_interface( &mut client, &session, FetchInterfaceResponse { scope_ref: None, interface: root_reference("math"), descriptor: descriptor(TypeExpr::Integer), validator: Some(b"interface-v1".to_vec()), }, ); let call = client .prepare_call( &session, "/item", &root_reference("math"), "calculate", args(7), ) .unwrap(); client.mark_dispatched(call.id).unwrap(); assert_eq!( client .refresh_observed(&session, "/other", 0) .unwrap_err() .kind, ClientErrorKind::Capacity ); let unknown = ProtocolError { code: ProtocolErrorCode::OperationOutcomeUnknown, message: "commit uncertain".into(), }; client .complete( call.id, encode_protocol_error_response( ProtocolInteraction::CallOperation, &unknown, wire_limits(), ) .unwrap(), ) .unwrap(); assert_eq!( client.object(&session, "/item").unwrap().state, ObservationState::Stale ); observe_object( &mut client, &session, "/other", object("other", None, b"other"), ); assert!(client.object(&session, "/item").is_none()); assert!(client.object(&session, "/other").is_some()); } #[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 { scope_ref: None, interface: root_reference("math"), descriptor: descriptor(TypeExpr::Integer), validator: Some(b"i1".to_vec()), }, ); let call = client .prepare_call( &session, "/item", &root_reference("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, &root_reference("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 { scope_ref: None, interface: root_reference("math"), 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 { scope_ref: None, interface: root_reference("math"), descriptor: descriptor(TypeExpr::Integer), validator: None, }, ); let call = client .prepare_call( &session, "/item", &root_reference("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", &root_reference("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 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 = ObserveRequest { path: "/".into(), depth: 1, }; let tree_value = TestObservationResponse { root: TestObservation { object: object("", None, b"root"), children: vec![TestObservation { object: object("child", None, b"tree-child"), children: vec![], }], }, }; let old_tree = client.refresh_observed(&session, "/", 1).unwrap(); let newer_observation = client.refresh_observed(&session, "/child", 0).unwrap(); client .complete( old_tree.id, encode_test_observation_response(&tree_request, &tree_value, wire_limits()).unwrap(), ) .unwrap(); assert!(matches!( client.object(&session, "/child").unwrap().state, ObservationState::Loading(id) if id == newer_observation.id )); client .complete( newer_observation.id, observation_response("/child", &object("child", None, b"newer-observation")), ) .unwrap(); assert_eq!( client.object(&session, "/child").unwrap().validator, Some(b"newer-observation".to_vec()) ); 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, observation_response("/child", &object("child", None, b"completed-newer")), ) .unwrap(); client .complete( older_tree.id, encode_test_observation_response(&tree_request, &tree_value, wire_limits()).unwrap(), ) .unwrap(); assert_eq!( client.object(&session, "/child").unwrap().validator, Some(b"completed-newer".to_vec()) ); 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, observation_response("/child", &object("child", None, b"newer-than-omission")), ) .unwrap(); client .complete( older_omitting_tree.id, encode_test_observation_response( &tree_request, &TestObservationResponse { root: TestObservation { object: object("", None, b"older-root"), children: vec![], }, }, wire_limits(), ) .unwrap(), ) .unwrap(); let child = client.object(&session, "/child").unwrap(); assert_eq!(child.state, ObservationState::Fresh); assert_eq!(child.validator, Some(b"newer-than-omission".to_vec())); 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, encode_test_observation_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_observation = client.object(&session, "/child").unwrap(); assert!(matches!( child_observation.state, ObservationState::Loading(id) if id == newer_child_tree.id )); assert_eq!( child_observation.validator, Some(b"newer-than-omission".to_vec()) ); let child_request = ObserveRequest { path: "/child".into(), depth: 1, }; let child_value = TestObservationResponse { root: TestObservation { object: object("child", None, b"newer-tree"), children: vec![], }, }; client .complete( newer_child_tree.id, encode_test_observation_response(&child_request, &child_value, wire_limits()).unwrap(), ) .unwrap(); assert_eq!( client.tree(&session, "/child").unwrap().state, ObservationState::Fresh ); let older_parent = client.refresh_observed(&session, "/", 1).unwrap(); let newer_child = client.refresh_observed(&session, "/child", 1).unwrap(); client .complete( newer_child.id, encode_test_observation_response( &child_request, &TestObservationResponse { root: TestObservation { object: object("child", None, b"completed-newer-tree"), children: vec![], }, }, wire_limits(), ) .unwrap(), ) .unwrap(); client .complete( older_parent.id, encode_test_observation_response(&tree_request, &tree_value, wire_limits()).unwrap(), ) .unwrap(); assert_eq!( client.object(&session, "/child").unwrap().validator, Some(b"completed-newer-tree".to_vec()) ); assert_eq!( client.tree(&session, "/child").unwrap().revision, newer_child.id ); } #[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.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") .unwrap(), Completion::StaleResponseRejected(old.id) ); assert!(matches!( client.object(&session, "/item").unwrap().state, ObservationState::Loading(id) if id == newer.id )); client .complete( newer.id, observation_response("/item", &object("item", None, b"new")), ) .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") .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_test_observation_response( &ObserveRequest { path: "/".into(), depth: 1, }, &TestObservationResponse { root: TestObservation { object: object("", None, b"root"), children: vec![], }, }, wire_limits(), ) .unwrap(), ) .unwrap(); let old = client .prepare_interface(&session, root_reference("math")) .unwrap(); let newer = client .prepare_interface(&session, root_reference("math")) .unwrap(); assert_eq!( client .fail_transport(old.id, "old connection failed") .unwrap(), Completion::StaleResponseRejected(old.id) ); assert!(matches!( client.interface(&session, &root_reference("math")).unwrap().state, ObservationState::Loading(id) if id == newer.id )); let value = FetchInterfaceResponse { scope_ref: None, interface: root_reference("math"), descriptor: descriptor(TypeExpr::Integer), validator: Some(b"current".to_vec()), }; client .complete( newer.id, encode_fetch_interface_response( &FetchInterfaceRequest { interface: root_reference("math"), }, &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 { scope_ref: None, interface: root_reference("math"), descriptor: descriptor(TypeExpr::Integer), validator: None, }, ); let refresh = client .prepare_interface(&session, root_reference("math")) .unwrap(); let contradictory = FetchInterfaceResponse { scope_ref: None, interface: root_reference("math"), descriptor: descriptor(TypeExpr::Integer), validator: Some(b"became-mutable".to_vec()), }; let error = client .complete( refresh.id, encode_fetch_interface_response( &FetchInterfaceRequest { interface: root_reference("math"), }, &contradictory, wire_limits(), ) .unwrap(), ) .unwrap_err(); assert_eq!(error.kind, ClientErrorKind::InvalidResponse); assert_eq!( client .interface(&session, &root_reference("math")) .unwrap() .validator, None ); } #[test] fn entry_eviction_never_discards_a_loading_tree_subject() { let mut client = client(1, 8, 8, 8); let session = client .open_session("https://example.test/wip", SecurityContext::new("subject")) .unwrap(); observe_object(&mut client, &session, "/", object("", None, b"root")); let tree = client.refresh_observed(&session, "/", 1).unwrap(); assert_eq!( client .refresh_observed(&session, "/other", 0) .unwrap_err() .kind, ClientErrorKind::Capacity ); assert!(matches!( client.tree(&session, "/").unwrap().state, ObservationState::Loading(id) if id == tree.id )); client .complete( tree.id, encode_test_observation_response( &ObserveRequest { path: "/".into(), depth: 1, }, &TestObservationResponse { root: TestObservation { object: object("", None, b"tree"), children: vec![], }, }, wire_limits(), ) .unwrap(), ) .unwrap(); assert_eq!( client.tree(&session, "/").unwrap().state, ObservationState::Fresh ); } #[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.refresh_observed(&session, path, 0).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.refresh_observed(&session, path, 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, "/gamma").is_some()); } #[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.refresh_observed(&session, "/one", 0).unwrap(); let error = client.refresh_observed(&session, "/two", 0).unwrap_err(); assert_eq!(error.kind, ClientErrorKind::Capacity); client .complete( first.id, observation_response("/one", &object("one", None, b"1")), ) .unwrap(); // Diagnostic history evicts oldest entries at the configured bound. 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( 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); } #[test] fn structured_interface_keys_membership_and_scope_ref_are_preserved() { let mut client = client(8, 8, 8, 8); let session = client .open_session("https://example.test/wip", SecurityContext::new("subject")) .unwrap(); let root = root_reference("same#::操作"); let local = wip_protocol::InterfaceReference { scope: "/item".into(), name: root.name.clone(), }; let mut value = object("item", None, b"object-v1"); value.interfaces = vec![local.clone()]; observe_object(&mut client, &session, "/item", value); for (reference, scope_ref) in [ (root.clone(), None), (local.clone(), Some("scope-identity".into())), ] { observe_interface( &mut client, &session, FetchInterfaceResponse { interface: reference, descriptor: descriptor(TypeExpr::Integer), scope_ref, validator: None, }, ); } assert!(client.interface(&session, &root).is_some()); assert_eq!( client .interface(&session, &local) .unwrap() .scope_ref .as_deref(), Some("scope-identity") ); assert_eq!( client .prepare_call(&session, "/item", &root, "calculate", args(1)) .unwrap_err() .kind, ClientErrorKind::InvalidRequest ); let call = client .prepare_call(&session, "/item", &local, "calculate", args(1)) .unwrap(); let context = client.pending_call_context(call.id).unwrap(); assert_eq!(context.request().interface.reference, local); assert_eq!( context.request().interface.scope_ref.as_deref(), Some("scope-identity") ); assert_eq!( context.interface().scope_ref.as_deref(), Some("scope-identity") ); assert_eq!(client.interface(&session, &root).unwrap().scope_ref, None); }