diff --git a/Cargo.lock b/Cargo.lock index 2b31c94..fd1d788 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -33,6 +33,7 @@ dependencies = [ "async-trait", "bytes", "futures", + "jsonschema", "parking_lot", "serde", "serde_json", diff --git a/Cargo.toml b/Cargo.toml index 8853e0c..09de3cc 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -19,6 +19,7 @@ default = [] [dependencies] alktype = "0.1.0" +jsonschema = { version = "0.46", default-features = false } tokio = { version = "1", features = ["full"] } serde = { version = "1", features = ["derive"] } serde_json = "1" diff --git a/docs/reviews/001-pub-and-channels-integration-review.md b/docs/reviews/001-pub-and-channels-integration-review.md index 575f847..819794f 100644 --- a/docs/reviews/001-pub-and-channels-integration-review.md +++ b/docs/reviews/001-pub-and-channels-integration-review.md @@ -127,6 +127,13 @@ publish should register like `subscribe` with `timeout: None`, not the ## P-03 [major] — `publish_schema` per-chunk validation is written but never read +**Fixed (2026-08-13, Unit 8).** `publish_schema` now round-trips through +`spec_to_json` / `operation_spec_schema()` / `rebuild_spec_for`, and both +dispatch paths (`pump_sink` and `run_loop_single_stream`) validate each +`call.published` chunk's `input` against the schema via +`alktype::validation::build_validator`; a validation failure injects +`Err(CallError::invalid_input(...))` and terminates the stream. + **ADR drift:** ADR-046 §4, and the field's own doc comment at `src/registry/spec.rs:181-186` which says "the dispatch path validates each `call.published` event's `payload.input` against this schema @@ -144,6 +151,12 @@ documented-but-unimplemented behavior. ## P-04 [major] — Initiator `call.error` does not terminate the publish stream +**Fixed (2026-08-13, Unit 8).** Both `pump_sink` and +`run_loop_single_stream` now match `call.error`, parse the `CallError` +from the payload, inject it as `Err(call_error)` into the sink's +`chunk_tx`, and terminate the feed. A malformed payload falls back to +`CallError::internal("publish error from initiator (malformed)")`. + **ADR drift:** ADR-046 §5 event table and §6 ("`Err(call_error)` for an initiator-side error (a `call.error` event from the initiator)"); plus the `SinkHandler`/`PublishStream` doc contract @@ -204,6 +217,13 @@ multiplexed on that stream. ## P-07 [major] — Handler-returns-early: response withheld until the initiator finishes +**Fixed (2026-08-13, Unit 8).** `pump_sink` now uses +`tokio::select!` over `handler.fuse()` and `reader.read_frame()`; when +the handler completes first, the response is written immediately and +the feed is short-circuited (the loop breaks, `chunk_tx` is dropped, +the handler's `PublishStream` sees EOF). The feed-wins branch awaits +the handler after the loop as before. + **ADR drift:** ADR-046 §7 (long-lived publish streams). **Verified:** YES. `pump_sink` uses `tokio::join!(feed_fut, handler)` and only writes the response after **both** complete @@ -1101,6 +1121,21 @@ ALPN. ## Unit 8 — Pub protocol semantics (P-03, P-04, P-07) +**Status: complete (2026-08-13).** All three findings fixed; 15 new +tests (7 in `dispatch.rs`, 4 in `discovery.rs`, 4 in `from_call.rs`… +the single-stream P-03/P-04 tests live in `channels/client.rs`); full +suite 480 passed, clippy/fmt/doc clean. `alktype::validation::build_validator` +is now used on both dispatch paths (`pump_sink` and +`run_loop_single_stream`); `jsonschema` was added as a direct +dependency (the validator return type is part of `alktype`'s public +API). The `SinkDispatch` struct gained a `publish_validator` field +built once at dispatch time; the single-stream `in_flight_sinks` map +gained an `InFlightSink` struct carrying the validator alongside +`chunk_tx`. The P-07 rework replaced `tokio::join!` with a +`tokio::select!` loop over `handler.fuse()` and `reader.read_frame()` +so an early handler return writes the response immediately without +waiting for the feed. + **Goal:** the publish path's decided-but-unimplemented behaviors land: per-chunk schema validation, initiator `call.error` terminates the stream, and an early handler return short-circuits the feed. diff --git a/src/channels/client.rs b/src/channels/client.rs index 1ef453a..d904c27 100644 --- a/src/channels/client.rs +++ b/src/channels/client.rs @@ -246,6 +246,47 @@ mod tests { } } + /// Read one length-prefixed `EventEnvelope` frame off `reader` and + /// parse it into a `ResponseEnvelope`. Used by the single-stream + /// direct-drive tests to read the responder's single + /// `call.responded`/`call.error` frame for a Pub. + async fn read_single_stream_response( + reader: &mut (impl tokio::io::AsyncRead + Unpin), + ) -> ResponseEnvelope { + use tokio::io::AsyncReadExt; + let mut len_buf = [0u8; 4]; + reader + .read_exact(&mut len_buf) + .await + .expect("read response length"); + let len = u32::from_be_bytes(len_buf) as usize; + let mut body = vec![0u8; len]; + reader + .read_exact(&mut body) + .await + .expect("read response body"); + let env: crate::protocol::wire::EventEnvelope = + serde_json::from_slice(&body).expect("parse response envelope"); + let request_id = env.id.clone(); + match env.r#type.as_str() { + "call.responded" => ResponseEnvelope::ok( + request_id, + env.payload + .get("output") + .cloned() + .unwrap_or(serde_json::Value::Null), + ), + "call.error" => { + let err: crate::protocol::wire::CallError = serde_json::from_value(env.payload) + .unwrap_or_else(|_| { + crate::protocol::wire::CallError::internal("malformed error payload") + }); + ResponseEnvelope::error(request_id, err) + } + other => panic!("expected call.responded or call.error, got {other}"), + } + } + /// Build the `install_channel_zero` hook for the accept side: it /// yields channel 0's `BiStream` once, splits it into the shared /// writer + reader, constructs a single-stream `CallConnection`, @@ -485,6 +526,221 @@ mod tests { ); } + /// Unit 8 / P-03 (single-stream path) — `publish_schema` validation + /// fires on the single-stream dispatch loop: a chunk that violates + /// the schema is injected as an `Err(INVALID_INPUT)` into the + /// `SinkHandler`'s stream, terminating it. The handler observes the + /// valid chunks before the violation plus the error, and its + /// response reflects that. This proves the `EVENT_PUBLISHED` arm of + /// `run_loop_single_stream` validates against the + /// `InFlightSink.publish_validator` before yielding `Ok(chunk)`. + #[tokio::test] + async fn channel_0_publish_schema_rejects_invalid_chunk_on_single_stream() { + use futures::stream::StreamExt; + + let publish_schema = serde_json::json!({ + "type": "object", + "properties": { "bytes": { "type": "string" } }, + "required": ["bytes"] + }); + let recording_sink = crate::registry::registration::make_sink_handler( + |_input, ctx, mut stream| async move { + let mut oks: Vec = Vec::new(); + let mut err: Option = None; + while let Some(item) = stream.next().await { + match item { + Ok(v) => oks.push(v), + Err(e) => { + err = Some(e); + break; + } + } + } + ResponseEnvelope::ok(ctx.request_id, serde_json::json!({ "ok": oks, "err": err })) + }, + ); + let mut registry = crate::registry::registration::OperationRegistry::new(); + registry + .register(HandlerRegistration::new( + OperationSpec::new( + "fs/upload", + OperationType::Pub, + Visibility::External, + serde_json::json!({}), + serde_json::json!({}), + vec![], + AccessControl::default(), + None, + ) + .with_publish_schema(publish_schema), + HandlerKind::Sink(recording_sink), + OperationProvenance::Local, + None, + None, + crate::core::types::Capabilities::new(), + )) + .unwrap(); + let registry = Arc::new(registry); + + let (client_end, server_end) = tokio::io::duplex(64 * 1024); + let client_conn = + Connection::from_bidi(client_end, b"alknet/channels".to_vec(), Some(TEST_ADDR)); + let server_conn = + Connection::from_bidi(server_end, b"alknet/channels".to_vec(), Some(TEST_ADDR)); + + let adapter = ChannelsAdapter::new( + make_install_channel_zero(Arc::clone(®istry)), + Arc::new(NoCap), + ); + let auth = AuthContext::anonymous(b"alknet/channels"); + let _server_handle = tokio::spawn(async move { + let _ = crate::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await; + }); + + let client = ChannelClient::from_connection(client_conn) + .await + .expect("channel client init"); + let call_conn = client + .take_call_connection() + .await + .expect("call connection present"); + + let chunks = vec![ + serde_json::json!({"bytes": "hello"}), + serde_json::json!({"bytes": 42}), + ]; + let stream: std::pin::Pin< + Box + Send>, + > = Box::pin(futures::stream::iter(chunks)); + let response = tokio::time::timeout( + std::time::Duration::from_secs(10), + call_conn.publish("fs/upload", serde_json::json!({}), stream), + ) + .await + .expect("channel-0 publish timed out"); + + let out = response.result.expect("publish ok"); + let oks = out.get("ok").and_then(|v| v.as_array()).expect("ok array"); + assert_eq!(oks.len(), 1, "only the valid chunk reached the handler"); + assert_eq!(oks[0], serde_json::json!({"bytes": "hello"})); + let err = out.get("err").expect("err field"); + assert_eq!( + err.get("code").and_then(|c| c.as_str()), + Some("INVALID_INPUT"), + "invalid chunk injects INVALID_INPUT on the single-stream path" + ); + } + + /// Unit 8 / P-04 (single-stream path) — an initiator-side + /// `call.error` during a publish over channel 0 is routed to the + /// matching in-flight sink's `chunk_tx` as `Err(call_error)`, + /// terminating the stream. The handler observes the chunks before + /// the error plus the initiator's `CallError` (not a synthetic + /// "aborted"). This drives `run_loop_single_stream` directly with + /// crafted frames (a `call.requested`, one `call.published`, then a + /// `call.error`) because the public `publish()` API takes + /// `Stream` and cannot emit an initiator `call.error` + /// — that path is for a future `Stream>` publish API. The `call.error` is injected manually + /// on the wire to prove the single-stream `EVENT_ERROR` arm parses + /// the payload and routes it to the right in-flight sink. + #[tokio::test] + async fn channel_0_initiator_call_error_terminates_publish_on_single_stream() { + use crate::protocol::connection::split_single_stream; + use crate::protocol::dispatch::Dispatcher; + use crate::protocol::wire::EventEnvelope; + use futures::stream::StreamExt; + + let recording_sink = crate::registry::registration::make_sink_handler( + |_input, ctx, mut stream| async move { + let mut oks: Vec = Vec::new(); + let mut err: Option = None; + while let Some(item) = stream.next().await { + match item { + Ok(v) => oks.push(v), + Err(e) => { + err = Some(e); + break; + } + } + } + ResponseEnvelope::ok(ctx.request_id, serde_json::json!({ "ok": oks, "err": err })) + }, + ); + let mut registry = crate::registry::registration::OperationRegistry::new(); + registry + .register(HandlerRegistration::new( + OperationSpec::new( + "fs/upload", + OperationType::Pub, + Visibility::External, + serde_json::json!({}), + serde_json::json!({}), + vec![], + AccessControl::default(), + None, + ), + HandlerKind::Sink(recording_sink), + OperationProvenance::Local, + None, + None, + crate::core::types::Capabilities::new(), + )) + .unwrap(); + let registry = Arc::new(registry); + + let request = EventEnvelope::requested( + "pub-ss-err-1", + serde_json::json!({ + "operationId": "fs/upload", + "input": {}, + }), + ); + let published = EventEnvelope::published("pub-ss-err-1", serde_json::json!({"chunk": 1})); + let initiator_error = EventEnvelope::error( + "pub-ss-err-1", + &crate::protocol::wire::CallError::new("PUBLISH_FAILED", "disk full", false), + ); + + let mut frame_buf = Vec::new(); + for env in [&request, &published, &initiator_error] { + let body = serde_json::to_vec(env).expect("serialize envelope"); + frame_buf.extend_from_slice(&(body.len() as u32).to_be_bytes()); + frame_buf.extend_from_slice(&body); + } + + let (response_send, mut response_recv) = tokio::io::duplex(64 * 1024); + let bidi = crate::core::types::BiStream::from_joined( + tokio::io::BufReader::new(std::io::Cursor::new(frame_buf)), + response_send, + ); + let (writer, reader) = split_single_stream(bidi); + + let call_connection = + Arc::new(CallConnection::new(crate::protocol::sink_empty_connection())); + let dp = Dispatcher::new(registry, Arc::new(NoopIdProvider)); + let _server_handle = tokio::spawn(async move { + dp.run_loop_single_stream(call_connection, reader, writer) + .await; + }); + + let response = read_single_stream_response(&mut response_recv).await; + let out = response.result.expect("handler ok"); + let oks = out.get("ok").and_then(|v| v.as_array()).expect("ok array"); + assert_eq!( + oks.len(), + 1, + "the chunk before the error reached the handler" + ); + assert_eq!(oks[0], serde_json::json!({"chunk": 1})); + let err = out.get("err").expect("err field"); + assert_eq!( + err.get("code").and_then(|c| c.as_str()), + Some("PUBLISH_FAILED"), + "the initiator's CallError code reaches the handler on the single-stream path" + ); + } + /// C-02 / C-03 — the end-to-end acceptance gate for Unit 3 /// (`register_openable`, ADR-047 §3 as amended 2026-08-13 — /// per-connection registration). Wires `ChannelClient` (connect diff --git a/src/client/from_call.rs b/src/client/from_call.rs index b8ba41f..02cea7b 100644 --- a/src/client/from_call.rs +++ b/src/client/from_call.rs @@ -272,6 +272,12 @@ fn rebuild_spec_for( } } + if let Some(publish_schema) = schema_json.get("publish_schema") { + if !publish_schema.is_null() { + spec = spec.with_publish_schema(publish_schema.clone()); + } + } + Ok(spec) } @@ -495,6 +501,7 @@ mod tests { use crate::core::auth::Identity; use crate::core::types::Capabilities; use crate::protocol::connection::CallConnection; + use crate::registry::discovery::spec_to_json; use crate::registry::registration::{make_handler, make_streaming_handler}; use crate::registry::spec::OperationType; use std::collections::HashMap; @@ -628,6 +635,63 @@ mod tests { ); } + #[test] + fn rebuild_spec_publish_schema_set_when_present() { + let publish_schema = json!({ + "type": "object", + "properties": { "bytes": { "type": "string" } }, + "required": ["bytes"] + }); + let mut schema = sample_schema_json("fs/upload", "pub"); + schema["publish_schema"] = publish_schema.clone(); + let spec = rebuild_spec_for(&schema, "fs/upload", &None).expect("rebuild"); + assert_eq!(spec.publish_schema.as_ref(), Some(&publish_schema)); + } + + #[test] + fn rebuild_spec_publish_schema_absent_when_omitted() { + let schema = sample_schema_json("fs/upload", "pub"); + let spec = rebuild_spec_for(&schema, "fs/upload", &None).expect("rebuild"); + assert!( + spec.publish_schema.is_none(), + "publish_schema must be absent when the discovered schema omits it" + ); + } + + #[test] + fn rebuild_spec_publish_schema_absent_when_null() { + let mut schema = sample_schema_json("fs/upload", "pub"); + schema["publish_schema"] = Value::Null; + let spec = rebuild_spec_for(&schema, "fs/upload", &None).expect("rebuild"); + assert!( + spec.publish_schema.is_none(), + "publish_schema: null must be treated as absent" + ); + } + + #[test] + fn rebuild_spec_publish_schema_round_trips_with_spec_to_json() { + let publish_schema = json!({ + "type": "object", + "properties": { "n": { "type": "integer" } }, + "required": ["n"] + }); + let spec = OperationSpec::new( + "fs/upload", + OperationType::Pub, + Visibility::External, + json!({}), + json!({}), + vec![], + AccessControl::default(), + None, + ) + .with_publish_schema(publish_schema.clone()); + let serialized = spec_to_json(&spec); + let rebuilt = rebuild_spec_for(&serialized, "fs/upload", &None).expect("rebuild"); + assert_eq!(rebuilt.publish_schema.as_ref(), Some(&publish_schema)); + } + #[test] fn derive_alpn_from_op_name_strips_channels_prefix() { assert_eq!( diff --git a/src/protocol/dispatch.rs b/src/protocol/dispatch.rs index 9af89bb..c1be99b 100644 --- a/src/protocol/dispatch.rs +++ b/src/protocol/dispatch.rs @@ -30,7 +30,7 @@ use super::abort::AbortCascade; use super::connection::CallConnection; use super::wire::{ CallError, EventEnvelope, FrameFramedReader, FrameFramedWriter, ResponseEnvelope, - EVENT_ABORTED, EVENT_COMPLETED, EVENT_PUBLISHED, EVENT_REQUESTED, + EVENT_ABORTED, EVENT_COMPLETED, EVENT_ERROR, EVENT_PUBLISHED, EVENT_REQUESTED, }; use crate::protocol::adapter::SessionOverlaySource; use crate::registry::context::{AbortPolicy, OperationContext, ScopedPeerEnv}; @@ -68,6 +68,13 @@ pub enum DispatchResult { /// `ResponseEnvelope`. /// - `chunk_tx`: the mpsc sender for feeding `call.published` chunks from /// the wire into the handler's `PublishStream`. +/// - `publish_validator`: the per-chunk validator built from the op's +/// `publish_schema` (ADR-046 §4), `None` when the op has no +/// `publish_schema`. Each `call.published` chunk's `input` is +/// validated against this before being yielded to the handler; a +/// validation failure injects an `Err(CallError::invalid_input(...))` +/// that terminates the stream (matching the `SinkHandler` contract: +/// an `Err` terminates the stream). /// /// `handle_stream` reads `call.published` events from the wire, sends /// each chunk's `input` into `chunk_tx`, and when the stream ends @@ -76,6 +83,18 @@ pub enum DispatchResult { pub struct SinkDispatch { pub(crate) handler: Pin + Send>>, pub(crate) chunk_tx: mpsc::Sender>, + pub(crate) publish_validator: Option, +} + +/// An in-flight Pub (Sink) on the single-stream dispatch loop +/// (`run_loop_single_stream`), tracked by `request_id`. Carries the +/// feed's `chunk_tx` and the per-chunk validator (built once at +/// dispatch time from the op's `publish_schema`). `call.published` +/// frames arriving after the `call.requested` are routed here; +/// `call.completed` / `call.aborted` / `call.error` remove the entry. +struct InFlightSink { + chunk_tx: mpsc::Sender>, + publish_validator: Option, } impl std::fmt::Debug for DispatchResult { @@ -315,6 +334,21 @@ impl Dispatcher { Ok(h) => h, Err(envelope) => return DispatchResult::Once(envelope), }; + let publish_validator = self + .registry + .registration(&operation_name) + .and_then(|r| r.spec.publish_schema.as_ref()) + .and_then(|schema| match alktype::validation::build_validator(schema) { + Ok(v) => Some(v), + Err(e) => { + warn!( + operation = %operation_name, + error = %e, + "publish_schema failed to compile; chunks will not be validated", + ); + None + } + }); let (chunk_tx, chunk_rx) = mpsc::channel::>(PUBLISH_CHANNEL_BUFFER); let publish_stream: PublishStream = Box::pin(chunk_rx); @@ -322,6 +356,7 @@ impl Dispatcher { DispatchResult::Sink(SinkDispatch { handler: Box::pin(handler), chunk_tx, + publish_validator, }) } } @@ -441,10 +476,32 @@ impl Dispatcher { /// is written to the wire as `call.responded` (or `call.error`). /// /// The handler runs concurrently — it consumes the - /// `PublishStream` while we feed it. On `call.aborted` for this - /// request ID, the feed injects an `Err` into `chunk_tx` and ends; - /// the handler future is awaited to completion (it is not yet - /// cancelled — the abort-cancels-Pub mechanism is review 001 Unit 9). + /// `PublishStream` while we feed it. The feed and the handler are + /// raced via `tokio::select!` (P-07): if the handler returns early + /// (e.g. rejects the upload after chunk 1), the feed is + /// short-circuited and the response is written immediately, without + /// waiting for the initiator to finish publishing. Dropping + /// `chunk_tx` on the handler-wins branch signals the feed to stop; + /// the handler's `PublishStream` sees EOF on its next `next().await`. + /// + /// Per-chunk validation (P-03, ADR-046 §4): when + /// `sink.publish_validator` is `Some`, each `call.published` chunk's + /// `input` is validated against the schema before being sent into + /// `chunk_tx`. On validation failure, an + /// `Err(CallError::invalid_input(...))` is injected and the feed + /// terminates — matching the `SinkHandler` contract that an `Err` + /// terminates the stream. When `publish_validator` is `None`, + /// chunks are yielded as-is (current behavior). + /// + /// Initiator `call.error` (P-04, ADR-046 §6): a `call.error` event + /// from the initiator is parsed into a `CallError`, injected as an + /// `Err` item, and terminates the feed — the handler sees the + /// initiator's error, not a synthetic "aborted" message. + /// + /// On `call.aborted` for this request ID, the feed injects an `Err` + /// into `chunk_tx` and ends; the handler future is awaited to + /// completion (it is not yet cancelled — the abort-cancels-Pub + /// mechanism is review 001 Unit 9). pub(crate) async fn pump_sink( &self, reader: &mut super::wire::FrameFramedReader, @@ -458,58 +515,100 @@ impl Dispatcher { let SinkDispatch { handler, mut chunk_tx, + publish_validator, } = sink; - let feed_fut = async { - loop { - match reader.read_frame().await { - Ok(envelope) => { - if envelope.id != request_id { - debug!( - event_type = %envelope.r#type, - id = %envelope.id, - "ignoring event with mismatched id during sink pump" - ); - continue; + use futures::FutureExt; + let mut handler = handler.fuse(); + let mut response: Option = None; + + loop { + tokio::select! { + biased; + resp = &mut handler => { + response = Some(resp); + break; + } + read = reader.read_frame() => { + match read { + Err(super::wire::FrameError::ConnectionClosed) => break, + Err(err) => { + warn!(error = %err, "frame read error during sink pump; ending stream"); + break; } - match envelope.r#type.as_str() { - EVENT_PUBLISHED => { - let chunk = envelope - .payload - .get("input") - .cloned() - .unwrap_or(Value::Null); - if chunk_tx.send(Ok(chunk)).await.is_err() { - break; - } - } - EVENT_COMPLETED => break, - EVENT_ABORTED => { - let _ = chunk_tx - .send(Err(CallError::internal("publish aborted by initiator"))) - .await; - break; - } - _ => { + Ok(envelope) => { + if envelope.id != request_id { debug!( event_type = %envelope.r#type, - "ignoring non-published event during sink pump" + id = %envelope.id, + "ignoring event with mismatched id during sink pump" ); + continue; + } + match envelope.r#type.as_str() { + EVENT_PUBLISHED => { + let chunk = envelope + .payload + .get("input") + .cloned() + .unwrap_or(Value::Null); + if let Some(validator) = &publish_validator { + if !validator.is_valid(&chunk) { + let details = serde_json::json!({ + "chunk": chunk, + }); + let err = CallError::invalid_input( + "published chunk failed publish_schema validation", + ) + .with_details(details); + let _ = chunk_tx.send(Err(err)).await; + break; + } + } + if chunk_tx.send(Ok(chunk)).await.is_err() { + break; + } + } + EVENT_COMPLETED => break, + EVENT_ABORTED => { + let _ = chunk_tx + .send(Err(CallError::internal( + "publish aborted by initiator", + ))) + .await; + break; + } + EVENT_ERROR => { + let call_error: CallError = + serde_json::from_value(envelope.payload).unwrap_or_else( + |_| { + CallError::internal( + "publish error from initiator (malformed)", + ) + }, + ); + let _ = chunk_tx.send(Err(call_error)).await; + break; + } + other => { + debug!( + event_type = %other, + "ignoring non-published event during sink pump" + ); + } } } } - Err(super::wire::FrameError::ConnectionClosed) => break, - Err(err) => { - warn!(error = %err, "frame read error during sink pump; ending stream"); - break; - } } } - drop(chunk_tx); - }; + } - let (feed_result, response) = tokio::join!(feed_fut, handler); - let _: () = feed_result; + drop(chunk_tx); + + let response = match response { + Some(r) => r, + None => handler.await, + }; let event: EventEnvelope = response.into(); if let Err(err) = writer.write_frame(&event).await { @@ -627,8 +726,7 @@ impl Dispatcher { }); let mut reader = FrameFramedReader::new(reader); - let mut in_flight_sinks: HashMap>> = - HashMap::new(); + let mut in_flight_sinks: HashMap = HashMap::new(); loop { let envelope = match reader.read_frame().await { @@ -664,8 +762,18 @@ impl Dispatcher { .await; } DispatchResult::Sink(sink) => { - let SinkDispatch { handler, chunk_tx } = sink; - in_flight_sinks.insert(request_id.clone(), chunk_tx); + let SinkDispatch { + handler, + chunk_tx, + publish_validator, + } = sink; + in_flight_sinks.insert( + request_id.clone(), + InFlightSink { + chunk_tx, + publish_validator, + }, + ); let writer_clone = Arc::clone(&writer); let request_id_for_handler = request_id.clone(); tokio::spawn(async move { @@ -684,8 +792,9 @@ impl Dispatcher { } EVENT_ABORTED => { let request_id = envelope.id.clone(); - if let Some(mut chunk_tx) = in_flight_sinks.remove(&request_id) { - let _ = chunk_tx + if let Some(mut entry) = in_flight_sinks.remove(&request_id) { + let _ = entry + .chunk_tx .send(Err(CallError::internal("publish aborted by initiator"))) .await; } else { @@ -699,9 +808,26 @@ impl Dispatcher { .get("input") .cloned() .unwrap_or(Value::Null); - if let Some(mut chunk_tx) = in_flight_sinks.remove(&request_id) { - let _ = chunk_tx.send(Ok(chunk)).await; - in_flight_sinks.insert(request_id, chunk_tx); + if let Some(mut entry) = in_flight_sinks.remove(&request_id) { + let validated = match &entry.publish_validator { + Some(validator) => { + if validator.is_valid(&chunk) { + Ok(chunk) + } else { + let details = serde_json::json!({ "chunk": chunk }); + Err(CallError::invalid_input( + "published chunk failed publish_schema validation", + ) + .with_details(details)) + } + } + None => Ok(chunk), + }; + let keep = validated.is_ok(); + let _ = entry.chunk_tx.send(validated).await; + if keep { + in_flight_sinks.insert(request_id, entry); + } } else { debug!( request_id = %request_id, @@ -713,6 +839,21 @@ impl Dispatcher { let request_id = envelope.id.clone(); in_flight_sinks.remove(&request_id); } + EVENT_ERROR => { + let request_id = envelope.id.clone(); + if let Some(mut entry) = in_flight_sinks.remove(&request_id) { + let call_error: CallError = serde_json::from_value(envelope.payload) + .unwrap_or_else(|_| { + CallError::internal("publish error from initiator (malformed)") + }); + let _ = entry.chunk_tx.send(Err(call_error)).await; + } else { + debug!( + request_id = %request_id, + "single-stream: call.error for unknown in-flight sink; dropping" + ); + } + } other => { debug!( event_type = %other, @@ -1854,4 +1995,385 @@ mod tests { Some(&Value::String("NOT_FOUND".into())) ); } + + // --- ADR-046 §4: publish_schema validation (P-03) --------------------- + + fn pub_spec_with_publish_schema(name: &str, publish_schema: Value) -> OperationSpec { + OperationSpec::new( + name, + OperationType::Pub, + Visibility::External, + serde_json::json!({}), + serde_json::json!({}), + vec![], + AccessControl::default(), + None, + ) + .with_publish_schema(publish_schema) + } + + fn registry_with_pub_and_schema( + name: &str, + publish_schema: Value, + handler: crate::registry::registration::SinkHandler, + ) -> Arc { + let mut registry = OperationRegistry::new(); + registry + .register(HandlerRegistration::new( + pub_spec_with_publish_schema(name, publish_schema), + HandlerKind::Sink(handler), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )) + .unwrap(); + Arc::new(registry) + } + + /// A sink handler that records every item it received (`Ok` values and + /// the first `Err`) so tests can assert what the dispatch path injected + /// into the `PublishStream`. Returns the recordings in the response + /// output as `{ ok: [...], err: Option }`. + fn recording_sink_handler() -> crate::registry::registration::SinkHandler { + use crate::registry::registration::make_sink_handler; + use futures::stream::StreamExt; + make_sink_handler(|_input, ctx, mut stream| async move { + let mut oks: Vec = Vec::new(); + let mut err: Option = None; + while let Some(item) = stream.next().await { + match item { + Ok(v) => oks.push(v), + Err(e) => { + err = Some(e); + break; + } + } + } + ResponseEnvelope::ok(ctx.request_id, serde_json::json!({ "ok": oks, "err": err })) + }) + } + + #[tokio::test] + async fn handle_stream_pub_publish_schema_validates_and_rejects_invalid_chunk() { + let schema = serde_json::json!({ + "type": "object", + "properties": { "bytes": { "type": "string" } }, + "required": ["bytes"] + }); + let registry = registry_with_pub_and_schema("fs/upload", schema, recording_sink_handler()); + let provider: Arc = Arc::new(StaticIdentityProvider::new()); + let dp = Dispatcher::new(registry, provider); + let conn = Arc::new(CallConnection::new(stub_connection())); + + let request = EventEnvelope::requested( + "pub-schema-1", + serde_json::json!({ + "operationId": "/fs/upload", + "input": { "path": "/x" }, + }), + ); + let valid_chunk = + EventEnvelope::published("pub-schema-1", serde_json::json!({"bytes": "hello"})); + let invalid_chunk = + EventEnvelope::published("pub-schema-1", serde_json::json!({"bytes": 42})); + let completed = EventEnvelope::completed("pub-schema-1"); + + let mut frame_buf = Vec::new(); + frame_buf.extend_from_slice(&encode_frame(&request)); + frame_buf.extend_from_slice(&encode_frame(&valid_chunk)); + frame_buf.extend_from_slice(&encode_frame(&invalid_chunk)); + frame_buf.extend_from_slice(&encode_frame(&completed)); + + let recv = tokio::io::BufReader::new(std::io::Cursor::new(frame_buf)); + let (send, mut sink) = tokio::io::duplex(8 * 1024); + let stream = crate::core::types::BiStream::from_joined(recv, send); + + let handle = tokio::spawn(async move { + dp.handle_stream(conn, stream).await; + }); + + let frames = read_all_frames(&mut sink).await; + handle.await.unwrap(); + assert_eq!(frames.len(), 1, "one response frame"); + assert_eq!(frames[0].r#type, EVENT_RESPONDED); + assert_eq!(frames[0].id, "pub-schema-1"); + let output = frames[0].payload.get("output").expect("output field"); + let oks = output + .get("ok") + .and_then(|v| v.as_array()) + .expect("ok array"); + assert_eq!(oks.len(), 1, "only the valid chunk reached the handler"); + assert_eq!(oks[0], serde_json::json!({"bytes": "hello"})); + let err = output.get("err").expect("err field"); + assert!( + err.get("code").and_then(|c| c.as_str()) == Some("INVALID_INPUT"), + "invalid chunk injects INVALID_INPUT, got: {err}" + ); + } + + #[tokio::test] + async fn handle_stream_pub_publish_schema_passes_valid_chunks_unchanged() { + let schema = serde_json::json!({ + "type": "object", + "properties": { "n": { "type": "integer" } }, + "required": ["n"] + }); + let registry = registry_with_pub_and_schema("fs/upload", schema, recording_sink_handler()); + let provider: Arc = Arc::new(StaticIdentityProvider::new()); + let dp = Dispatcher::new(registry, provider); + let conn = Arc::new(CallConnection::new(stub_connection())); + + let request = EventEnvelope::requested( + "pub-schema-2", + serde_json::json!({ + "operationId": "/fs/upload", + "input": {}, + }), + ); + let c1 = EventEnvelope::published("pub-schema-2", serde_json::json!({"n": 1})); + let c2 = EventEnvelope::published("pub-schema-2", serde_json::json!({"n": 2})); + let c3 = EventEnvelope::published("pub-schema-2", serde_json::json!({"n": 3})); + let completed = EventEnvelope::completed("pub-schema-2"); + + let mut frame_buf = Vec::new(); + frame_buf.extend_from_slice(&encode_frame(&request)); + frame_buf.extend_from_slice(&encode_frame(&c1)); + frame_buf.extend_from_slice(&encode_frame(&c2)); + frame_buf.extend_from_slice(&encode_frame(&c3)); + frame_buf.extend_from_slice(&encode_frame(&completed)); + + let recv = tokio::io::BufReader::new(std::io::Cursor::new(frame_buf)); + let (send, mut sink) = tokio::io::duplex(8 * 1024); + let stream = crate::core::types::BiStream::from_joined(recv, send); + + let handle = tokio::spawn(async move { + dp.handle_stream(conn, stream).await; + }); + + let frames = read_all_frames(&mut sink).await; + handle.await.unwrap(); + assert_eq!(frames.len(), 1); + let output = frames[0].payload.get("output").expect("output field"); + let oks = output + .get("ok") + .and_then(|v| v.as_array()) + .expect("ok array"); + assert_eq!(oks.len(), 3, "all three valid chunks reached the handler"); + assert!(output + .get("err") + .and_then(|e| e.as_object()) + .is_none_or(|o| o.is_empty())); + } + + #[tokio::test] + async fn handle_stream_pub_no_publish_schema_yields_chunks_as_is() { + let registry = registry_with_pub("fs/upload", recording_sink_handler()); + let provider: Arc = Arc::new(StaticIdentityProvider::new()); + let dp = Dispatcher::new(registry, provider); + let conn = Arc::new(CallConnection::new(stub_connection())); + + let request = EventEnvelope::requested( + "pub-schema-3", + serde_json::json!({ + "operationId": "/fs/upload", + "input": {}, + }), + ); + let c1 = EventEnvelope::published("pub-schema-3", serde_json::json!({"anything": true})); + let completed = EventEnvelope::completed("pub-schema-3"); + + let mut frame_buf = Vec::new(); + frame_buf.extend_from_slice(&encode_frame(&request)); + frame_buf.extend_from_slice(&encode_frame(&c1)); + frame_buf.extend_from_slice(&encode_frame(&completed)); + + let recv = tokio::io::BufReader::new(std::io::Cursor::new(frame_buf)); + let (send, mut sink) = tokio::io::duplex(8 * 1024); + let stream = crate::core::types::BiStream::from_joined(recv, send); + + let handle = tokio::spawn(async move { + dp.handle_stream(conn, stream).await; + }); + + let frames = read_all_frames(&mut sink).await; + handle.await.unwrap(); + assert_eq!(frames.len(), 1); + let output = frames[0].payload.get("output").expect("output field"); + let oks = output + .get("ok") + .and_then(|v| v.as_array()) + .expect("ok array"); + assert_eq!(oks.len(), 1); + assert_eq!(oks[0], serde_json::json!({"anything": true})); + } + + // --- ADR-046 §6: initiator call.error terminates the stream (P-04) ---- + + #[tokio::test] + async fn handle_stream_pub_initiator_call_error_terminates_stream_with_initiator_error() { + let registry = registry_with_pub("fs/upload", recording_sink_handler()); + let provider: Arc = Arc::new(StaticIdentityProvider::new()); + let dp = Dispatcher::new(registry, provider); + let conn = Arc::new(CallConnection::new(stub_connection())); + + let request = EventEnvelope::requested( + "pub-err-1", + serde_json::json!({ + "operationId": "/fs/upload", + "input": {}, + }), + ); + let c1 = EventEnvelope::published("pub-err-1", serde_json::json!({"chunk": 1})); + let initiator_error = EventEnvelope::error( + "pub-err-1", + &CallError::new("UPLOAD_FAILED", "disk full on initiator", false), + ); + + let mut frame_buf = Vec::new(); + frame_buf.extend_from_slice(&encode_frame(&request)); + frame_buf.extend_from_slice(&encode_frame(&c1)); + frame_buf.extend_from_slice(&encode_frame(&initiator_error)); + + let recv = tokio::io::BufReader::new(std::io::Cursor::new(frame_buf)); + let (send, mut sink) = tokio::io::duplex(8 * 1024); + let stream = crate::core::types::BiStream::from_joined(recv, send); + + let handle = tokio::spawn(async move { + dp.handle_stream(conn, stream).await; + }); + + let frames = read_all_frames(&mut sink).await; + handle.await.unwrap(); + assert_eq!(frames.len(), 1, "one response frame"); + let output = frames[0].payload.get("output").expect("output field"); + let oks = output + .get("ok") + .and_then(|v| v.as_array()) + .expect("ok array"); + assert_eq!(oks.len(), 1, "the one valid chunk before the error"); + let err = output.get("err").expect("err field"); + assert_eq!( + err.get("code").and_then(|c| c.as_str()), + Some("UPLOAD_FAILED"), + "the initiator's CallError code reaches the handler, not a synthetic 'aborted'" + ); + assert_eq!( + err.get("message").and_then(|m| m.as_str()), + Some("disk full on initiator") + ); + } + + #[tokio::test] + async fn handle_stream_pub_initiator_call_error_malformed_payload_falls_back_to_internal() { + let registry = registry_with_pub("fs/upload", recording_sink_handler()); + let provider: Arc = Arc::new(StaticIdentityProvider::new()); + let dp = Dispatcher::new(registry, provider); + let conn = Arc::new(CallConnection::new(stub_connection())); + + let request = EventEnvelope::requested( + "pub-err-2", + serde_json::json!({ + "operationId": "/fs/upload", + "input": {}, + }), + ); + let malformed_error = EventEnvelope::new( + "call.error", + "pub-err-2", + serde_json::json!("not an object"), + ); + + let mut frame_buf = Vec::new(); + frame_buf.extend_from_slice(&encode_frame(&request)); + frame_buf.extend_from_slice(&encode_frame(&malformed_error)); + + let recv = tokio::io::BufReader::new(std::io::Cursor::new(frame_buf)); + let (send, mut sink) = tokio::io::duplex(8 * 1024); + let stream = crate::core::types::BiStream::from_joined(recv, send); + + let handle = tokio::spawn(async move { + dp.handle_stream(conn, stream).await; + }); + + let frames = read_all_frames(&mut sink).await; + handle.await.unwrap(); + assert_eq!(frames.len(), 1); + let output = frames[0].payload.get("output").expect("output field"); + let err = output.get("err").expect("err field"); + assert_eq!( + err.get("code").and_then(|c| c.as_str()), + Some("INTERNAL"), + "malformed call.error payload falls back to INTERNAL" + ); + } + + // --- ADR-046 §7: early handler return short-circuits the feed (P-07) -- + + #[tokio::test] + async fn handle_stream_pub_early_handler_return_short_circuits_slow_feed() { + use std::sync::Arc as StdArc; + use std::sync::Mutex as StdMutex; + + let seen_chunks: StdArc>> = StdArc::new(StdMutex::new(Vec::new())); + let seen_clone = StdArc::clone(&seen_chunks); + let early_handler: crate::registry::registration::SinkHandler = { + use crate::registry::registration::make_sink_handler; + use futures::stream::StreamExt; + make_sink_handler(move |_input, ctx, mut stream| { + let seen = StdArc::clone(&seen_clone); + async move { + if let Some(Ok(v)) = stream.next().await { + seen.lock().unwrap().push(v); + } + ResponseEnvelope::ok( + ctx.request_id, + serde_json::json!({ "rejected": true, "after": 1 }), + ) + } + }) + }; + let registry = registry_with_pub("fs/upload", early_handler); + let provider: Arc = Arc::new(StaticIdentityProvider::new()); + let dp = Dispatcher::new(registry, provider); + let conn = Arc::new(CallConnection::new(stub_connection())); + + let request = EventEnvelope::requested( + "pub-early-1", + serde_json::json!({ + "operationId": "/fs/upload", + "input": {}, + }), + ); + let c1 = EventEnvelope::published("pub-early-1", serde_json::json!({"chunk": 1})); + + let mut frame_buf = Vec::new(); + frame_buf.extend_from_slice(&encode_frame(&request)); + frame_buf.extend_from_slice(&encode_frame(&c1)); + + let recv = tokio::io::BufReader::new(std::io::Cursor::new(frame_buf)); + let (send, mut sink) = tokio::io::duplex(8 * 1024); + let stream = crate::core::types::BiStream::from_joined(recv, send); + + let start = std::time::Instant::now(); + dp.handle_stream(conn, stream).await; + let elapsed = start.elapsed(); + + let frames = read_all_frames(&mut sink).await; + assert_eq!(frames.len(), 1, "response written once handler returns"); + assert_eq!(frames[0].r#type, EVENT_RESPONDED); + assert_eq!(frames[0].id, "pub-early-1"); + let output = frames[0].payload.get("output").expect("output field"); + assert_eq!(output["rejected"], serde_json::json!(true)); + assert!( + elapsed < std::time::Duration::from_secs(5), + "response must not be deferred behind a slow/absent feed (elapsed: {elapsed:?})" + ); + let seen = seen_chunks.lock().unwrap(); + assert_eq!( + seen.len(), + 1, + "handler consumed exactly one chunk before returning" + ); + } } diff --git a/src/registry/discovery.rs b/src/registry/discovery.rs index a7c56cf..ff6995a 100644 --- a/src/registry/discovery.rs +++ b/src/registry/discovery.rs @@ -145,6 +145,9 @@ fn operation_spec_schema() -> Value { "channel_open": { "type": ["boolean", "null"], "description": "Marker (ADR-047): when true, the op's stream is binary and the channels layer allocates a data channel for it. Absent/null for JSON-stream ops." + }, + "publish_schema": { + "description": "Schema for each published chunk's `input` (Pub ops only, ADR-046 §4). Absent/null for Query/Mutation/Sub ops and for Pub ops with no per-chunk validation. When present, the dispatch path validates each `call.published` event's `payload.input` against this schema before yielding it to the SinkHandler." } }, "required": [ @@ -194,7 +197,7 @@ fn error_definition_to_json(def: &super::spec::ErrorDefinition) -> Value { }) } -fn spec_to_json(spec: &OperationSpec) -> Value { +pub(crate) fn spec_to_json(spec: &OperationSpec) -> Value { let error_schemas: Vec = spec .error_schemas .iter() @@ -213,6 +216,9 @@ fn spec_to_json(spec: &OperationSpec) -> Value { if spec.channel_open.is_some() { json["channel_open"] = json!(true); } + if let Some(publish_schema) = &spec.publish_schema { + json["publish_schema"] = publish_schema.clone(); + } json } @@ -839,6 +845,60 @@ mod tests { ); } + #[test] + fn spec_to_json_emits_publish_schema_when_set() { + let publish_schema = json!({ + "type": "object", + "properties": { "bytes": { "type": "string" } }, + "required": ["bytes"] + }); + let spec = OperationSpec::new( + "fs/upload", + OperationType::Pub, + Visibility::External, + json!({}), + json!({}), + vec![], + AccessControl::default(), + None, + ) + .with_publish_schema(publish_schema.clone()); + let json_val = spec_to_json(&spec); + assert_eq!(json_val.get("publish_schema"), Some(&publish_schema)); + } + + #[test] + fn spec_to_json_omits_publish_schema_when_absent() { + let spec = OperationSpec::new( + "fs/readFile", + OperationType::Query, + Visibility::External, + json!({}), + json!({}), + vec![], + AccessControl::default(), + None, + ); + let json_val = spec_to_json(&spec); + assert!( + json_val.get("publish_schema").is_none(), + "publish_schema must be absent when not set" + ); + } + + #[test] + fn operation_spec_schema_documents_publish_schema_property() { + let schema = operation_spec_schema(); + let props = schema + .get("properties") + .and_then(|v| v.as_object()) + .expect("properties object"); + assert!( + props.contains_key("publish_schema"), + "operation_spec_schema must advertise publish_schema" + ); + } + #[tokio::test] async fn services_list_filters_by_access_control_authorized_peer() { let registry = registry_with_access_controlled_ops();