diff --git a/Cargo.lock b/Cargo.lock index 48a11ab..ee57f4b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1301,7 +1301,9 @@ version = "0.1.0" name = "wip-public-api-example" version = "0.1.0" dependencies = [ + "wip-client", "wip-host", + "wip-http", "wip-interface", "wip-protocol", ] diff --git a/crates/wip-public-api-example/Cargo.toml b/crates/wip-public-api-example/Cargo.toml index 30c8f04..30dbd41 100644 --- a/crates/wip-public-api-example/Cargo.toml +++ b/crates/wip-public-api-example/Cargo.toml @@ -12,3 +12,7 @@ publish = false wip-host.workspace = true wip-interface.workspace = true wip-protocol.workspace = true + +[dev-dependencies] +wip-client.workspace = true +wip-http.workspace = true diff --git a/crates/wip-public-api-example/src/lib.rs b/crates/wip-public-api-example/src/lib.rs index ff5bb43..44a8ccf 100644 --- a/crates/wip-public-api-example/src/lib.rs +++ b/crates/wip-public-api-example/src/lib.rs @@ -229,7 +229,7 @@ pub fn run() -> Result> { let published = dispatch_publish( &provider, - catalog.reference(), + &catalog_response, PublishRequest { title: "Public APIs".into(), body: Some("protocol values in, validated values out".into()), @@ -243,7 +243,7 @@ pub fn run() -> Result> { )?; let rejected = dispatch_publish( &provider, - catalog.reference(), + &catalog_response, PublishRequest { title: String::new(), body: None, @@ -256,7 +256,8 @@ pub fn run() -> Result> { &provider, request( OBJECT_PATH, - catalog.reference(), + &catalog_response.interface, + catalog_response.scope_ref.as_deref(), Some(CATALOG_VALIDATOR), "status", BTreeMap::new(), @@ -266,7 +267,8 @@ pub fn run() -> Result> { &provider, request( OBJECT_PATH, - audit.reference(), + &audit_response.interface, + audit_response.scope_ref.as_deref(), None, "status", BTreeMap::new(), @@ -276,7 +278,8 @@ pub fn run() -> Result> { let host_error = provider .call_operation(request( OBJECT_PATH, - catalog.reference(), + &catalog_response.interface, + catalog_response.scope_ref.as_deref(), Some(CATALOG_VALIDATOR), "backend_status", BTreeMap::new(), @@ -288,7 +291,8 @@ pub fn run() -> Result> { let unknown_protocol_failure = provider .call_operation(request( OBJECT_PATH, - catalog.reference(), + &catalog_response.interface, + catalog_response.scope_ref.as_deref(), Some(CATALOG_VALIDATOR), "uncertain_backend_status", BTreeMap::new(), @@ -299,7 +303,8 @@ pub fn run() -> Result> { let permission_failure = provider .call_operation(request( OBJECT_PATH, - catalog.reference(), + &catalog_response.interface, + catalog_response.scope_ref.as_deref(), Some(CATALOG_VALIDATOR), "restricted_status", BTreeMap::new(), @@ -329,7 +334,8 @@ pub fn run() -> Result> { let interface_mismatch = provider .call_operation(request( OBJECT_PATH, - diagnostics.reference(), + &diagnostics.reference().protocol_reference(), + catalog_response.scope_ref.as_deref(), None, "diagnostics", BTreeMap::new(), @@ -351,7 +357,8 @@ pub fn run() -> Result> { let mut stale_scope = request( OBJECT_PATH, - catalog.reference(), + &catalog_response.interface, + catalog_response.scope_ref.as_deref(), Some(CATALOG_VALIDATOR), "status", BTreeMap::new(), @@ -421,13 +428,14 @@ fn ensure_kind(error: &wip_host::CallError, expected: CallErrorKind) -> Result<( fn dispatch_publish( provider: &ApplicationProvider, - catalog: &InterfaceRef, + catalog: &wip_protocol::FetchInterfaceResponse, request_value: PublishRequest, ) -> Result> { let encoded_request = request_value.encode()?; let result = provider.call_operation(request( OBJECT_PATH, - catalog, + &catalog.interface, + catalog.scope_ref.as_deref(), Some(CATALOG_VALIDATOR), "publish", BTreeMap::from([("request".into(), encoded_request)]), @@ -445,7 +453,8 @@ fn dispatch_string( fn request( path: &str, - interface: &InterfaceRef, + interface: &InterfaceReference, + scope_ref: Option<&str>, interface_validator: Option<&[u8]>, operation: &str, arguments: BTreeMap, @@ -456,8 +465,8 @@ fn request( validator: Some(OBJECT_VALIDATOR.to_vec()), }, interface: InterfaceTarget { - scope_ref: Some("article-service-instance".into()), - reference: interface.protocol_reference(), + scope_ref: scope_ref.map(str::to_owned), + reference: interface.clone(), validator: interface_validator.map(<[u8]>::to_vec), }, operation: operation.into(), diff --git a/crates/wip-public-api-example/tests/end_to_end.rs b/crates/wip-public-api-example/tests/end_to_end.rs index 4bf1257..6849890 100644 --- a/crates/wip-public-api-example/tests/end_to_end.rs +++ b/crates/wip-public-api-example/tests/end_to_end.rs @@ -258,6 +258,40 @@ fn assert_schema_neutral(value: &Value) { } } +#[test] +fn scoped_dtos_keep_identity_binding_and_validators_separate() { + // Exact literals/destructuring guard the public Rust contract, including + // omission of refs rather than requiring a fabricated identity. + let reference = wip_protocol::InterfaceReference { + scope: "/article-service".into(), + name: "catalog::read#raw".into(), + }; + let request = wip_protocol::FetchInterfaceRequest { + interface: reference.clone(), + }; + request.validate().unwrap(); + let wip_protocol::FetchInterfaceResponse { + interface, + scope_ref, + descriptor, + validator, + } = wip_protocol::FetchInterfaceResponse { + interface: reference.clone(), + scope_ref: None, + descriptor: CatalogDefinition::descriptor(), + validator: None, + }; + let target = wip_protocol::InterfaceTarget { + reference: interface, + scope_ref, + validator, + }; + assert_eq!(target.reference, reference); + assert!(target.scope_ref.is_none()); + assert!(target.validator.is_none()); + assert_eq!(descriptor.format, INTERFACE_FORMAT_V1); +} + fn root_reference(name: &str) -> wip_protocol::InterfaceReference { wip_protocol::InterfaceReference { scope: "/".into(), diff --git a/crates/wip-public-api-example/tests/scoped_http.rs b/crates/wip-public-api-example/tests/scoped_http.rs new file mode 100644 index 0000000..abef293 --- /dev/null +++ b/crates/wip-public-api-example/tests/scoped_http.rs @@ -0,0 +1,559 @@ +//! External-style consumer: a transactional provider, real TCP HTTP, and Client. +use std::collections::BTreeMap; +use std::io::{Read, Write}; +use std::net::{TcpListener, TcpStream}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; +use std::thread; +use std::time::Duration; + +use wip_client::{ + CallOutcome, Client, ClientLimits, Completion, ObservationState, SecurityContext, +}; +use wip_host::{ + ExecutionContext, InterfaceRegistry, ObjectProvider, ProviderResult, PublishedObject, + RegisteredInterface, Scope, ScopeObservation, Validator, WipObject, +}; +use wip_http::{http, *}; +use wip_interface as wip; +use wip_protocol::{InterfaceReference, ObjectObservation, ProtocolErrorCode, ProtocolInteraction}; + +#[wip::interface] +pub trait Status { + fn status(&self) -> wip::OperationResult; + fn uncertain(&self) -> wip::OperationResult; +} + +struct Service { + name: String, + reference: Option, + validator: Validator, + calls: Arc, +} +impl WipObject for Service { + fn name(&self) -> &str { + &self.name + } + fn reference(&self) -> Option<&str> { + self.reference.as_deref() + } + fn validator(&self) -> Option<&Validator> { + Some(&self.validator) + } +} +impl Status for Service { + fn status(&self) -> wip::OperationResult { + self.calls.fetch_add(1, Ordering::SeqCst); + Ok(self.name.clone()) + } + fn uncertain(&self) -> wip::OperationResult { + self.calls.fetch_add(1, Ordering::SeqCst); + Err(wip::HostError::internal( + "commit acknowledgement lost", + wip::EffectGuarantee::Unknown, + )) + } +} +struct Publication { + scope: Scope, + api: RegisteredInterface, + reference: Option, +} +struct State { + registry: InterfaceRegistry, + root: Publication, + service: Option, + calls: Arc, +} +impl State { + fn new(reference: Option<&str>) -> Self { + let mut registry = InterfaceRegistry::new(); + let scope = Scope::new("/").unwrap(); + let root = Publication { + api: registry + .register(&scope, "status", StatusDefinition, None) + .unwrap(), + scope, + reference: None, + }; + let mut state = Self { + registry, + root, + service: None, + calls: Arc::default(), + }; + state.republish(reference); + state + } + fn remove(&mut self) { + if let Some(old) = self.service.take() { + self.registry.remove_scope(&old.scope); + } + } + fn republish(&mut self, reference: Option<&str>) { + self.remove(); + let scope = Scope::new("/service").unwrap(); + self.service = Some(Publication { + api: self + .registry + .register( + &scope, + "status", + StatusDefinition, + Some(Validator::new(b"api-v1")), + ) + .unwrap(), + scope, + reference: reference.map(str::to_owned), + }); + } +} + +// The HTTP adapter holds the State lock across all provider lookups, preconditions +// and dispatch. No separate scope prelookup can race a replacement. +struct Provider<'a> { + state: &'a State, + visible: bool, +} +impl ObjectProvider for Provider<'_> { + type Object = Service; + fn with_object( + &self, + _: &ExecutionContext, + path: &str, + operation: impl FnOnce(&PublishedObject) -> R, + ) -> ProviderResult { + let publication = match path { + "/" => &self.state.root, + "/service" if self.visible => match &self.state.service { + Some(value) => value, + None => return ProviderResult::NotFound, + }, + _ => return ProviderResult::NotFound, + }; + let object = Service { + name: if path == "/" { "" } else { "service" }.into(), + reference: publication.reference.clone(), + validator: Validator::new(b"object-v1"), + calls: self.state.calls.clone(), + }; + let current = ScopeObservation::new(&publication.scope, &object); + let mut object = PublishedObject::new(object) + .bind_scope(current) + .expose(&publication.api) + .unwrap(); + if path == "/service" { + let root_object = Service { + name: "".into(), + reference: None, + validator: Validator::new(b"object-v1"), + calls: self.state.calls.clone(), + }; + object = object + .bind_scope(ScopeObservation::new(&self.state.root.scope, &root_object)) + .expose(&self.state.root.api) + .unwrap(); + } + ProviderResult::Found(operation(&object)) + } +} +fn limits() -> Limits { + Limits::new(65536, 65536, 64).unwrap() +} +fn reference(scope: &str) -> InterfaceReference { + InterfaceReference { + scope: scope.into(), + name: "status".into(), + } +} + +fn handle(state: &State, path: &str, body: &[u8], visible: bool) -> http::Response> { + let provider = Provider { state, visible }; + match path { + "/wip/v1/observe" => { + let request = decode_observe_request(body, limits()).unwrap(); + match provider.observe_object(&request.path) { + Ok(object) => encode_observe_response( + &request, + &ObjectObservation { + object, + children: None, + }, + limits(), + ) + .unwrap(), + Err(error) => encode_protocol_error_response( + ProtocolInteraction::Observe, + &error.protocol_error(), + limits(), + ) + .unwrap(), + } + } + "/wip/v1/fetch_interface" => { + let request = decode_fetch_interface_request(body, limits()).unwrap(); + match provider.fetch_interface(&state.registry, &request.interface) { + Ok(observation) => encode_fetch_interface_response( + &request, + &observation.protocol_response().unwrap(), + limits(), + ) + .unwrap(), + Err(error) => encode_protocol_error_response( + ProtocolInteraction::FetchInterface, + &error, + limits(), + ) + .unwrap(), + } + } + "/wip/v1/call_operation" => { + let metadata = decode_call_operation_metadata(body, limits()).unwrap(); + let decoded = Mutex::new(None); + let result = provider.call_operation_with_context_and_decoder( + &ExecutionContext::default(), + metadata.target.clone(), + metadata.interface.clone(), + metadata.operation.clone(), + |descriptor| { + let request = metadata.decode_request(body, descriptor, limits())?; + let arguments = request.arguments.clone(); + *decoded.lock().unwrap() = Some((request, descriptor.clone())); + Ok::<_, CodecError>(arguments) + }, + ); + match result { + Ok(result) => { + let (request, descriptor) = decoded.into_inner().unwrap().unwrap(); + encode_call_operation_response(&request, &descriptor, &result, limits()) + .unwrap() + } + Err(error) => encode_protocol_error_response( + ProtocolInteraction::CallOperation, + &error.protocol_error(), + limits(), + ) + .unwrap(), + } + } + _ => panic!("unexpected route {path}"), + } +} + +// Small bounded test-only HTTP/1.1 listener. Production serving is deliberately +// not an example dependency; both directions still cross an actual TCP socket. +fn exchange( + state: &Arc>, + request: http::Request>, + visible: bool, +) -> http::Response> { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let state = state.clone(); + let server = thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + let mut raw = Vec::new(); + let (split, length) = loop { + let mut chunk = [0; 1024]; + let count = stream.read(&mut chunk).unwrap(); + assert!(count > 0); + raw.extend_from_slice(&chunk[..count]); + assert!(raw.len() <= 70000); + if let Some(split) = raw.windows(4).position(|part| part == b"\r\n\r\n") { + let head = std::str::from_utf8(&raw[..split]).unwrap(); + let length = head + .lines() + .find_map(|line| line.strip_prefix("Content-Length: ")) + .unwrap() + .parse::() + .unwrap(); + if raw.len() >= split + 4 + length { + break (split, length); + } + } + }; + let path = std::str::from_utf8(&raw[..split]) + .unwrap() + .lines() + .next() + .unwrap() + .split_whitespace() + .nth(1) + .unwrap(); + let response = handle( + &state.lock().unwrap(), + path, + &raw[split + 4..split + 4 + length], + visible, + ); + write!(stream, "HTTP/1.1 {} OK\r\nConnection: close\r\nContent-Length: {}\r\nContent-Type: application/json\r\n\r\n", response.status().as_u16(), response.body().len()).unwrap(); + stream.write_all(response.body()).unwrap(); + }); + let mut stream = TcpStream::connect(address).unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + write!(stream, "POST {} HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n", request.uri().path(), request.body().len()).unwrap(); + stream.write_all(request.body()).unwrap(); + let mut raw = Vec::new(); + stream.read_to_end(&mut raw).unwrap(); + server.join().unwrap(); + let split = raw.windows(4).position(|part| part == b"\r\n\r\n").unwrap(); + let status = std::str::from_utf8(&raw[..split]) + .unwrap() + .split_whitespace() + .nth(1) + .unwrap() + .parse::() + .unwrap(); + http::Response::builder() + .status(status) + .header("content-type", "application/json") + .body(raw[split + 4..].to_vec()) + .unwrap() +} + +fn setup() -> (Client, wip_client::SessionId) { + let mut client = Client::new(ClientLimits::new(1, 16, 16, 16, 16).unwrap(), limits()); + let session = client + .open_session("http://localhost/wip", SecurityContext::new("alice")) + .unwrap(); + (client, session) +} +fn observe( + client: &mut Client, + session: &wip_client::SessionId, + state: &Arc>, + path: &str, +) { + let request = client.refresh_object(session, path).unwrap(); + let response = exchange(state, request.request, true); + client.complete(request.id, response).unwrap(); +} +fn fetch( + client: &mut Client, + session: &wip_client::SessionId, + state: &Arc>, + reference: &InterfaceReference, +) { + let request = client + .prepare_interface(session, reference.clone()) + .unwrap(); + let body = std::str::from_utf8(request.request.body()).unwrap(); + assert!(body.contains("\"scope\"")); + assert!(body.contains("\"name\"")); + let response = exchange(state, request.request, true); + assert!(matches!( + client.complete(request.id, response).unwrap(), + Completion::Interface(_) + )); +} + +#[test] +fn real_http_root_self_same_name_and_optional_scope_refs() { + for scope_ref in [Some("instance-a"), None] { + let state = Arc::new(Mutex::new(State::new(scope_ref))); + let (mut client, session) = setup(); + observe(&mut client, &session, &state, "/"); + let root = client.object(&session, "/").unwrap(); + assert_eq!(root.object.unwrap().name, ""); + observe(&mut client, &session, &state, "/service"); + for scope in ["/", "/service"] { + let selected = reference(scope); + fetch(&mut client, &session, &state, &selected); + let observation = client.interface(&session, &selected).unwrap().clone(); + assert_eq!( + observation.scope_ref.as_deref(), + if scope == "/" { None } else { scope_ref } + ); + let request = client + .prepare_call(&session, "/service", &selected, "status", BTreeMap::new()) + .unwrap(); + let logical = decode_call_operation_request( + request.request.body(), + observation.descriptor.as_ref().unwrap(), + limits(), + ) + .unwrap(); + assert_eq!(logical.interface.reference, selected); + assert_eq!(logical.interface.scope_ref, observation.scope_ref); + let response = exchange(&state, request.request, true); + assert!(matches!( + client.complete(request.id, response).unwrap(), + Completion::Call(record) if matches!(record.outcome, CallOutcome::Success(_)) + )); + } + assert_eq!(state.lock().unwrap().calls.load(Ordering::SeqCst), 2); + } +} + +#[test] +fn stale_scope_republication_and_ref_less_replacement_require_explicit_calls() { + let selected = reference("/service"); + let state = Arc::new(Mutex::new(State::new(Some("instance-a")))); + let (mut client, session) = setup(); + observe(&mut client, &session, &state, "/service"); + fetch(&mut client, &session, &state, &selected); + let request = client + .prepare_call(&session, "/service", &selected, "status", BTreeMap::new()) + .unwrap(); + state.lock().unwrap().republish(Some("instance-b")); + let response = exchange(&state, request.request, true); + assert_eq!(response.status(), http::StatusCode::CONFLICT); + assert!(matches!(client.complete(request.id, response).unwrap(), + Completion::Call(record) if matches!(record.outcome, + CallOutcome::ProtocolFailure(ref error) if error.code == ProtocolErrorCode::InterfaceMismatch))); + assert_eq!(state.lock().unwrap().calls.load(Ordering::SeqCst), 0); + assert_ne!( + client.interface(&session, &selected).unwrap().state, + ObservationState::Fresh + ); + assert!( + client + .prepare_call(&session, "/service", &selected, "status", BTreeMap::new()) + .is_err() + ); + observe(&mut client, &session, &state, "/service"); + fetch(&mut client, &session, &state, &selected); + assert_eq!( + client + .interface(&session, &selected) + .unwrap() + .scope_ref + .as_deref(), + Some("instance-b") + ); + assert_eq!( + state.lock().unwrap().calls.load(Ordering::SeqCst), + 0, + "re-observation is not execution" + ); + + state.lock().unwrap().remove(); + observe(&mut client, &session, &state, "/service"); + assert!(client.interface(&session, &selected).is_none()); + state.lock().unwrap().republish(None); + observe(&mut client, &session, &state, "/service"); + fetch(&mut client, &session, &state, &selected); + let request = client + .prepare_call(&session, "/service", &selected, "status", BTreeMap::new()) + .unwrap(); + // Ref-less replacement is intentionally not identity-trackable. Current + // membership, live publication and both validators still apply at dispatch. + state.lock().unwrap().republish(None); + let response = exchange(&state, request.request, true); + client.complete(request.id, response).unwrap(); + assert_eq!(state.lock().unwrap().calls.load(Ordering::SeqCst), 1); +} + +#[test] +fn real_http_unknown_outcome_is_never_automatically_retried() { + let state = Arc::new(Mutex::new(State::new(Some("instance-a")))); + let (mut client, session) = setup(); + observe(&mut client, &session, &state, "/service"); + let selected = reference("/service"); + fetch(&mut client, &session, &state, &selected); + let request = client + .prepare_call( + &session, + "/service", + &selected, + "uncertain", + BTreeMap::new(), + ) + .unwrap(); + let response = exchange(&state, request.request, true); + let completion = client.complete(request.id, response).unwrap(); + assert!( + matches!(completion, Completion::Call(record) if matches!(record.outcome, CallOutcome::Unknown { .. })) + ); + assert_eq!(client.in_flight_len(), 0); + assert_eq!(state.lock().unwrap().calls.load(Ordering::SeqCst), 1); + observe(&mut client, &session, &state, "/service"); + assert_eq!(state.lock().unwrap().calls.load(Ordering::SeqCst), 1); +} + +#[test] +fn principal_visibility_and_validator_precedence_cross_real_http() { + let state = Arc::new(Mutex::new(State::new(Some("instance-a")))); + let endpoint = Endpoint::parse("http://localhost/wip").unwrap(); + let selected = reference("/service"); + let fetch = wip_protocol::FetchInterfaceRequest { + interface: selected.clone(), + }; + let response = exchange( + &state, + encode_fetch_interface_request(&endpoint, &fetch, limits()).unwrap(), + false, + ); + assert!( + matches!(decode_fetch_interface_response(&fetch, &response, limits()).unwrap(), DecodedResponse::ProtocolFailure { error, .. } if error.code == ProtocolErrorCode::InterfaceNotFound) + ); + let descriptor = StatusDefinition::descriptor(); + let mut call = wip_protocol::CallOperationRequest { + target: wip_protocol::Target { + path: "/service".into(), + validator: Some(b"stale-object".to_vec()), + }, + interface: wip_protocol::InterfaceTarget { + reference: selected, + scope_ref: Some("stale-scope".into()), + validator: Some(b"stale-api".to_vec()), + }, + operation: "status".into(), + arguments: BTreeMap::new(), + }; + for (visible, object_validator, scope_ref, interface_validator, expected) in [ + ( + false, + b"stale-object".as_slice(), + Some("stale-scope"), + Some(b"stale-api".as_slice()), + ProtocolErrorCode::NotFound, + ), + ( + true, + b"stale-object".as_slice(), + Some("stale-scope"), + Some(b"stale-api".as_slice()), + ProtocolErrorCode::ValidatorMismatch, + ), + ( + true, + b"object-v1".as_slice(), + Some("stale-scope"), + Some(b"stale-api".as_slice()), + ProtocolErrorCode::InterfaceMismatch, + ), + ( + true, + b"object-v1".as_slice(), + Some("instance-a"), + None, + ProtocolErrorCode::InterfaceValidatorRequired, + ), + ( + true, + b"object-v1".as_slice(), + None, + Some(b"stale-api".as_slice()), + ProtocolErrorCode::InterfaceValidatorMismatch, + ), + ] { + call.target.validator = Some(object_validator.to_vec()); + call.interface.scope_ref = scope_ref.map(str::to_owned); + call.interface.validator = interface_validator.map(<[u8]>::to_vec); + let response = exchange( + &state, + encode_call_operation_request(&endpoint, &call, &descriptor, limits()).unwrap(), + visible, + ); + assert!( + matches!(decode_call_operation_response(&call, &descriptor, &response, limits()).unwrap(), DecodedResponse::ProtocolFailure { error, .. } if error.code == expected) + ); + } + assert_eq!(state.lock().unwrap().calls.load(Ordering::SeqCst), 0); +}