glm-5.3-flash
f84d214173
Remediates all six findings of review 004 (per-connection dispatch resolution and client-side op serving). All claims re-verified in source before remediation; F-02's member list gains ScopedPeerEnv (also Clone — fork surface simpler than estimated). - OperationRegistry: interior mutability (parking_lot RwLock on both maps); register takes &self; registration/list_operations return owned clones; fork() deep-copies registrations + cached publish-schema validators (F-02/F-03); OperationRegistryBuilder::from_registry. - install_bootstrap_discovery: services/list, services/list-peers, services/schema registered closed over the fork itself, so per-session openables are discoverable and services/schema answers from the fork (F-06). - Dispatcher::serve_single_stream: full-duplex single-stream loop — call.requested dispatches inbound; responded/completed/error resolve outbound pendings; aborted tries both tables (in-flight sink aborts + pending cascade); published routes inbound sinks (F-04). - ChannelClient::from_connection_with_serving(connection, Option<ServingConfig>): opt-in serving; from_connection keeps the pure-consumer default. - registry::op_register: OpRegisterRequest wire DTO (spec in services/schema JSON + replace flag), op_register_spec, op_register_handler (rebuild -> forwarding stub -> register_imported, forced Internal/FromCall), announce_op; CallError::already_exists; spec_to_json_pub; from_call's rebuild_spec_for + forwarding-handler constructors crate-shared (F-05). - ADR-047 §4 amendment #2: per-session fork is the dispatch-registry mechanism; overlay stays nested-invocation/peer-announced landing zone (F-01/F-02). - ADR-022 amendment 2026-09-03: bootstrap-op set (services/list, services/schema, op/register), opt-in connect-side serving, op/register wire shape (F-04/F-05). - alkhttp ADR-048 reconciliation note + OQ-05 re-pointed at the alkcall ADRs (Unit 1b). - Review 004 status -> remediated; remediation log with gates. Verification: - cargo test: 581 passed, 0 failed (565 baseline + 16 new) - cargo test --all-features: 598 passed, 0 failed - cargo clippy --all-targets -- -D warnings: clean - cargo clippy --all-features --all-targets -- -D warnings: clean - cargo clippy --target wasm32-unknown-unknown -- -D warnings: clean - cargo fmt --check: clean - cargo doc --no-deps: clean Gates: fork_registry_open_op_resolves_and_is_discoverable (open op via fork + services/list shows openable + services/schema validates), serving_loop_hub_to_consumer_call_resolves (hub->consumer call through consumer's serving loop, consumer->hub still resolves), op_register_announce_then_hub_call_routes_back_to_consumer (announce -> overlay -> hub call -> forwarding stub -> consumer serves).
2902 lines
117 KiB
Rust
2902 lines
117 KiB
Rust
//! Shared dispatch loop for `alk/call` connections.
|
|
//!
|
|
//! Both [`super::adapter::CallAdapter`]'s accept path and
|
|
//! [`crate::client::CallClient`]'s connect path produce a [`CallConnection`]
|
|
//! and hand it to the same dispatch loop here (ADR-017 §1): the loop reads
|
|
//! `EventEnvelope` frames off accepted bidirectional streams, dispatches
|
|
//! `call.requested` events against the operation registry, and writes the
|
|
//! response back on the same stream. The connection-establishment half
|
|
//! differs (accept vs dial); the dispatch half is shared.
|
|
//!
|
|
//! See `docs/architecture/` for the spec.
|
|
|
|
use std::collections::HashMap;
|
|
use std::future::Future;
|
|
use std::pin::Pin;
|
|
use std::sync::Arc;
|
|
use std::time::{Duration, Instant};
|
|
|
|
use crate::core::auth::{AuthToken, Identity, IdentityProvider};
|
|
use crate::core::ownership::OwnershipProvider;
|
|
use crate::core::types::StreamError;
|
|
use futures::channel::mpsc;
|
|
use futures::stream::StreamExt;
|
|
use futures::SinkExt;
|
|
use parking_lot::Mutex as ParkingLotMutex;
|
|
use serde_json::Value;
|
|
use tokio::sync::oneshot;
|
|
use tokio::task::JoinHandle;
|
|
use tracing::{debug, warn};
|
|
|
|
use super::abort::AbortCascade;
|
|
use super::connection::CallConnection;
|
|
use super::wire::{
|
|
CallError, EventEnvelope, FrameFramedReader, FrameFramedWriter, ResponseEnvelope,
|
|
EVENT_ABORTED, EVENT_COMPLETED, EVENT_ERROR, EVENT_PUBLISHED, EVENT_REQUESTED, EVENT_RESPONDED,
|
|
};
|
|
use crate::protocol::adapter::SessionOverlaySource;
|
|
use crate::registry::context::{AbortPolicy, OperationContext, ScopedPeerEnv};
|
|
use crate::registry::env::{LocalOperationEnv, OperationEnv, PeerCompositeEnv};
|
|
use crate::registry::registration::{OperationRegistry, PublishStream, ResponseStream};
|
|
use crate::registry::spec::OperationType;
|
|
|
|
const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);
|
|
const SWEEPER_INTERVAL: Duration = Duration::from_secs(10);
|
|
const PUBLISH_CHANNEL_BUFFER: usize = 64;
|
|
|
|
/// Outcome of dispatching a `call.requested` event. The dispatcher branches on
|
|
/// the registered operation's `op_type` (ADR-021 §6, ADR-046):
|
|
/// `Query`/`Mutation` produce a single [`ResponseEnvelope`] (`Once`),
|
|
/// `Sub` produces a [`ResponseStream`] (`Stream`) that `handle_stream`
|
|
/// pumps to the wire, `Pub` produces a [`SinkDispatch`] (`Sink`) that
|
|
/// `handle_stream` feeds from the wire.
|
|
///
|
|
/// This enum is the branch point the spec describes ("branches on
|
|
/// `op_type` in `handle_stream`"): `dispatch` returns it and
|
|
/// `handle_stream` matches on it, keeping the Once path (one frame, no
|
|
/// `call.completed`), the Stream path (each envelope → frame,
|
|
/// `call.completed` on natural end), and the Sink path (feed
|
|
/// `call.published` chunks → one `call.responded`) visibly distinct.
|
|
pub enum DispatchResult {
|
|
Once(ResponseEnvelope),
|
|
Stream(ResponseStream),
|
|
Sink(SinkDispatch),
|
|
}
|
|
|
|
/// The sink dispatch result for a `Pub` operation (ADR-046). Carries:
|
|
///
|
|
/// - `handler`: the future returned by `invoke_sink()` — the
|
|
/// `SinkHandler` consuming the `PublishStream` and producing a single
|
|
/// `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
|
|
/// (`call.completed` or wire read end), drops `chunk_tx` and awaits
|
|
/// `handler` to get the single `ResponseEnvelope` to write to the wire.
|
|
pub struct SinkDispatch {
|
|
pub(crate) handler: Pin<Box<dyn Future<Output = ResponseEnvelope> + Send>>,
|
|
pub(crate) chunk_tx: mpsc::Sender<Result<Value, CallError>>,
|
|
pub(crate) publish_validator: Option<jsonschema::Validator>,
|
|
}
|
|
|
|
/// 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`, the per-chunk validator (built once at dispatch
|
|
/// time from the op's `publish_schema`), and the spawned handler task's
|
|
/// `JoinHandle` so that `call.aborted` can cancel the handler future
|
|
/// (ADR-046 §5, review 001 Unit 9). `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<Result<Value, CallError>>,
|
|
publish_validator: Option<jsonschema::Validator>,
|
|
handler_handle: JoinHandle<()>,
|
|
}
|
|
|
|
impl std::fmt::Debug for DispatchResult {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
match self {
|
|
DispatchResult::Once(env) => f.debug_tuple("Once").field(env).finish(),
|
|
DispatchResult::Stream(_) => {
|
|
f.debug_tuple("Stream").field(&"<ResponseStream>").finish()
|
|
}
|
|
DispatchResult::Sink(_) => f.debug_tuple("Sink").field(&"<SinkDispatch>").finish(),
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Shared dispatcher for an established `CallConnection`. Constructed by
|
|
/// both `CallAdapter` (accept path) and `CallClient` (connect path) and used
|
|
/// to run the dispatch loop. Holds no per-connection state; the
|
|
/// `CallConnection` is passed into `run_loop`.
|
|
///
|
|
/// `in_flight_sink_aborts` maps in-flight Pub `request_id`s to
|
|
/// `oneshot::Sender`s so that a cross-stream `call.aborted` (arriving on a
|
|
/// different `BiStream` than the one the Pub is pumping) can signal the
|
|
/// running `pump_sink` to cancel the handler future (ADR-046 §5, review 001
|
|
/// Unit 9). The map is shared across all `handle_stream` tasks spawned by
|
|
/// `run_loop`; each `pump_sink` inserts its abort sender on entry and
|
|
/// removes it on exit.
|
|
pub struct Dispatcher {
|
|
pub registry: Arc<OperationRegistry>,
|
|
pub identity_provider: Arc<dyn IdentityProvider>,
|
|
pub session_source: Option<Arc<dyn SessionOverlaySource + Send + Sync>>,
|
|
pub ownership_provider: Option<Arc<dyn OwnershipProvider>>,
|
|
pub default_timeout: Duration,
|
|
in_flight_sink_aborts: Arc<ParkingLotMutex<HashMap<String, oneshot::Sender<()>>>>,
|
|
}
|
|
|
|
impl Dispatcher {
|
|
pub fn new(
|
|
registry: Arc<OperationRegistry>,
|
|
identity_provider: Arc<dyn IdentityProvider>,
|
|
) -> Self {
|
|
Self {
|
|
registry,
|
|
identity_provider,
|
|
session_source: None,
|
|
ownership_provider: None,
|
|
default_timeout: DEFAULT_TIMEOUT,
|
|
in_flight_sink_aborts: Arc::new(ParkingLotMutex::new(HashMap::new())),
|
|
}
|
|
}
|
|
|
|
pub fn with_session_source(
|
|
mut self,
|
|
source: Arc<dyn SessionOverlaySource + Send + Sync>,
|
|
) -> Self {
|
|
self.session_source = Some(source);
|
|
self
|
|
}
|
|
|
|
pub fn with_timeout(mut self, timeout: Duration) -> Self {
|
|
self.default_timeout = timeout;
|
|
self
|
|
}
|
|
|
|
pub fn with_ownership_provider(mut self, provider: Arc<dyn OwnershipProvider>) -> Self {
|
|
self.ownership_provider = Some(provider);
|
|
self
|
|
}
|
|
|
|
fn strip_leading_slash(operation_id: &str) -> &str {
|
|
operation_id.strip_prefix('/').unwrap_or(operation_id)
|
|
}
|
|
|
|
pub(crate) fn resolve_identity(
|
|
&self,
|
|
connection_identity: Option<Identity>,
|
|
payload: &Value,
|
|
) -> Option<Identity> {
|
|
let auth_token = payload.get("auth_token").and_then(|v| v.as_str());
|
|
match auth_token {
|
|
Some(token_str) => {
|
|
let token = AuthToken {
|
|
raw: token_str.as_bytes().to_vec(),
|
|
};
|
|
match self.identity_provider.resolve_from_token(&token) {
|
|
Some(identity) => Some(identity),
|
|
None => connection_identity,
|
|
}
|
|
}
|
|
None => connection_identity,
|
|
}
|
|
}
|
|
|
|
pub fn compose_root_env(
|
|
&self,
|
|
connection: &CallConnection,
|
|
context: &OperationContext,
|
|
) -> Arc<dyn OperationEnv + Send + Sync> {
|
|
let base: Arc<dyn OperationEnv + Send + Sync> =
|
|
Arc::new(LocalOperationEnv::new(Arc::clone(&self.registry)));
|
|
let session = self
|
|
.session_source
|
|
.as_ref()
|
|
.and_then(|s| s.overlay_for(context));
|
|
|
|
let mut env = PeerCompositeEnv::new(base);
|
|
if let Some(session) = session {
|
|
env = env.with_session(session);
|
|
}
|
|
if let Some(peer_id) = connection.identity().map(|identity| identity.id.clone()) {
|
|
env.attach_peer(peer_id, connection.overlay_env());
|
|
}
|
|
Arc::new(env)
|
|
}
|
|
|
|
pub(crate) fn build_root_context(
|
|
&self,
|
|
request_id: String,
|
|
operation_name: &str,
|
|
identity: Option<Identity>,
|
|
forwarded_for: Option<Identity>,
|
|
connection: &CallConnection,
|
|
) -> OperationContext {
|
|
let registration = self.registry.registration(operation_name);
|
|
let (composition_authority, capabilities, scoped_env) = match registration {
|
|
Some(r) => (
|
|
r.composition_authority.clone(),
|
|
r.capabilities.clone(),
|
|
r.scoped_env.clone().unwrap_or_else(ScopedPeerEnv::empty),
|
|
),
|
|
None => (
|
|
None,
|
|
crate::core::types::Capabilities::new(),
|
|
ScopedPeerEnv::empty(),
|
|
),
|
|
};
|
|
|
|
let stub_env: Arc<dyn OperationEnv + Send + Sync> =
|
|
Arc::new(LocalOperationEnv::new(Arc::clone(&self.registry)));
|
|
let mut context = OperationContext {
|
|
request_id,
|
|
parent_request_id: None,
|
|
identity: identity.clone(),
|
|
handler_identity: composition_authority,
|
|
forwarded_for,
|
|
capabilities,
|
|
metadata: HashMap::new(),
|
|
deadline: Some(Instant::now() + self.default_timeout),
|
|
scoped_env,
|
|
env: stub_env,
|
|
abort_policy: AbortPolicy::default(),
|
|
internal: false,
|
|
ownership: self.ownership_provider.clone(),
|
|
};
|
|
context.env = self.compose_root_env(connection, &context);
|
|
context
|
|
}
|
|
|
|
pub async fn dispatch_requested(
|
|
&self,
|
|
connection: &Arc<CallConnection>,
|
|
request_id: String,
|
|
payload: Value,
|
|
) -> ResponseEnvelope {
|
|
let request_id_for_error = request_id.clone();
|
|
match self.dispatch(connection, request_id, payload).await {
|
|
DispatchResult::Once(envelope) => envelope,
|
|
DispatchResult::Stream(mut stream) => stream.next().await.unwrap_or_else(|| {
|
|
ResponseEnvelope::error(
|
|
request_id_for_error,
|
|
CallError::internal(
|
|
"dispatch_requested called on a Sub op; use the streaming path",
|
|
),
|
|
)
|
|
}),
|
|
DispatchResult::Sink(_) => ResponseEnvelope::error(
|
|
request_id_for_error,
|
|
CallError::internal("dispatch_requested called on a Pub op; use the sink path"),
|
|
),
|
|
}
|
|
}
|
|
|
|
/// Dispatch a `call.requested` event, branching on the registered
|
|
/// operation's `op_type` (ADR-021 §6, ADR-046). `Query`/`Mutation` →
|
|
/// `invoke()` → [`DispatchResult::Once`]; `Sub` →
|
|
/// `invoke_streaming()` → [`DispatchResult::Stream`]; `Pub` →
|
|
/// `invoke_sink()` → [`DispatchResult::Sink`]. Unknown ops and ACL
|
|
/// failures resolve via the registry's own envelope/error paths.
|
|
///
|
|
/// For the `Sub` and `Pub` branches the root context's deadline is
|
|
/// cleared (`deadline: None`): streaming and publish operations are
|
|
/// long-running and unbounded — the 30s request/response deadline
|
|
/// does not apply. The Once branch keeps the deadline from
|
|
/// `build_root_context`.
|
|
pub async fn dispatch(
|
|
&self,
|
|
connection: &Arc<CallConnection>,
|
|
request_id: String,
|
|
payload: Value,
|
|
) -> DispatchResult {
|
|
let operation_id = payload
|
|
.get("operationId")
|
|
.and_then(|v| v.as_str())
|
|
.unwrap_or("");
|
|
let operation_name = Self::strip_leading_slash(operation_id).to_string();
|
|
|
|
let connection_identity = connection.identity().cloned();
|
|
let identity = self.resolve_identity(connection_identity, &payload);
|
|
let forwarded_for = payload
|
|
.get("forwarded_for")
|
|
.and_then(|v| serde_json::from_value::<Identity>(v.clone()).ok());
|
|
|
|
let input = payload.get("input").cloned().unwrap_or(Value::Null);
|
|
|
|
let op_type = self
|
|
.registry
|
|
.registration(&operation_name)
|
|
.map(|r| r.spec.op_type)
|
|
.unwrap_or(OperationType::Query);
|
|
|
|
let mut context = self.build_root_context(
|
|
request_id.clone(),
|
|
&operation_name,
|
|
identity,
|
|
forwarded_for,
|
|
connection,
|
|
);
|
|
|
|
match op_type {
|
|
OperationType::Query | OperationType::Mutation => {
|
|
let envelope = self.registry.invoke(&operation_name, input, context).await;
|
|
DispatchResult::Once(envelope)
|
|
}
|
|
OperationType::Sub => {
|
|
context.deadline = None;
|
|
let stream = self
|
|
.registry
|
|
.invoke_streaming(&operation_name, input, context);
|
|
DispatchResult::Stream(stream)
|
|
}
|
|
OperationType::Pub => {
|
|
context.deadline = None;
|
|
let sink_handler =
|
|
match self
|
|
.registry
|
|
.resolve_sink_handler(&operation_name, &input, &context)
|
|
{
|
|
Ok(h) => h,
|
|
Err(envelope) => return DispatchResult::Once(envelope),
|
|
};
|
|
let publish_validator = self.registry.publish_validator(&operation_name);
|
|
let (chunk_tx, chunk_rx) =
|
|
mpsc::channel::<Result<Value, CallError>>(PUBLISH_CHANNEL_BUFFER);
|
|
let publish_stream: PublishStream = Box::pin(chunk_rx);
|
|
let handler = (sink_handler)(input, context, publish_stream);
|
|
DispatchResult::Sink(SinkDispatch {
|
|
handler: Box::pin(handler),
|
|
chunk_tx,
|
|
publish_validator,
|
|
})
|
|
}
|
|
}
|
|
}
|
|
|
|
pub async fn handle_abort(&self, connection: &Arc<CallConnection>, request_id: &str) {
|
|
if let Some(tx) = self.in_flight_sink_aborts.lock().remove(request_id) {
|
|
let _ = tx.send(());
|
|
debug!(request_id, "signalled in-flight sink abort");
|
|
}
|
|
let mut pending = connection.pending().lock();
|
|
let mut cascade = AbortCascade::new(&mut pending);
|
|
let aborted = cascade.cascade_abort(request_id, AbortPolicy::AbortDependents);
|
|
pending.handle_aborted(request_id);
|
|
if !aborted.is_empty() {
|
|
debug!(count = aborted.len(), "abort cascade evicted descendants");
|
|
}
|
|
}
|
|
|
|
pub(crate) async fn handle_stream(
|
|
&self,
|
|
connection: Arc<CallConnection>,
|
|
stream: crate::core::types::BiStream,
|
|
) {
|
|
// `stream` is a `BiStream` (ADR-009) — `AsyncRead + AsyncWrite + Send
|
|
// + Unpin`. Split into the read and write halves the call protocol's
|
|
// frame reader/writer consume. The split is the stdlib idiom; no
|
|
// per-handler wrapper.
|
|
let (recv, send) = tokio::io::split(stream);
|
|
let mut reader = FrameFramedReader::new(recv);
|
|
let mut writer = FrameFramedWriter::new(send);
|
|
|
|
loop {
|
|
let envelope = match reader.read_frame().await {
|
|
Ok(env) => env,
|
|
Err(super::wire::FrameError::ConnectionClosed) => break,
|
|
Err(err) => {
|
|
warn!(error = %err, "stream frame read error; closing stream");
|
|
break;
|
|
}
|
|
};
|
|
|
|
match envelope.r#type.as_str() {
|
|
EVENT_REQUESTED => {
|
|
let request_id = envelope.id.clone();
|
|
let payload = envelope.payload.clone();
|
|
|
|
match self
|
|
.dispatch(&connection, request_id.clone(), payload)
|
|
.await
|
|
{
|
|
DispatchResult::Once(response) => {
|
|
let event: EventEnvelope = response.into();
|
|
if let Err(err) = writer.write_frame(&event).await {
|
|
warn!(error = %err, "failed to write response frame; closing stream");
|
|
break;
|
|
}
|
|
}
|
|
DispatchResult::Stream(stream) => {
|
|
self.pump_stream(&mut writer, &request_id, stream).await;
|
|
}
|
|
DispatchResult::Sink(sink) => {
|
|
self.pump_sink(&mut reader, &mut writer, request_id, sink)
|
|
.await;
|
|
}
|
|
}
|
|
}
|
|
EVENT_ABORTED => {
|
|
let request_id = envelope.id.clone();
|
|
self.handle_abort(&connection, &request_id).await;
|
|
}
|
|
other => {
|
|
debug!(event_type = %other, id = %envelope.id, "ignoring non-requested/non-aborted event on inbound stream");
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Pump a subscription's [`ResponseStream`] to the wire: each
|
|
/// [`ResponseEnvelope`] becomes an [`EventEnvelope`] frame (`call.responded`
|
|
/// for `Ok`, `call.error` for `Err`). On natural stream end (the stream
|
|
/// returned `None` without the last item being an `Err`), write a
|
|
/// `call.completed` frame. An `Err` envelope is terminal — the stream
|
|
/// ends after it and we do NOT write `call.completed` (ADR-021 §6).
|
|
///
|
|
/// If a frame write fails the pump stops early; the stream is dropped on
|
|
/// return, releasing the handler's resources via `Drop` (ADR-016). A
|
|
/// `call.aborted` for this request ID mutates the `PendingRequestMap`
|
|
/// (via `handle_abort` on another stream) but does not yet cancel this
|
|
/// task — the pump ends when the stream ends naturally or the I/O fails
|
|
/// (the abort-cancels mechanism for Sub/Pub is review 001 Unit 9).
|
|
pub(crate) async fn pump_stream<W: tokio::io::AsyncWrite + Unpin>(
|
|
&self,
|
|
writer: &mut super::wire::FrameFramedWriter<W>,
|
|
request_id: &str,
|
|
mut stream: ResponseStream,
|
|
) {
|
|
let mut last_was_error = false;
|
|
while let Some(envelope) = stream.next().await {
|
|
last_was_error = envelope.result.is_err();
|
|
let event: EventEnvelope = envelope.into();
|
|
if let Err(err) = writer.write_frame(&event).await {
|
|
warn!(error = %err, "failed to write streaming frame; closing stream");
|
|
return;
|
|
}
|
|
}
|
|
|
|
if !last_was_error {
|
|
let completed = EventEnvelope::completed(request_id);
|
|
if let Err(err) = writer.write_frame(&completed).await {
|
|
warn!(error = %err, "failed to write call.completed");
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Pump a `Pub` operation's sink to the wire (ADR-046): feed
|
|
/// `call.published` chunks from the wire into the `SinkDispatch`'s
|
|
/// `chunk_tx`, and when the publish stream ends (`call.completed`
|
|
/// initiator→responder, or wire read end), drop `chunk_tx` and await
|
|
/// the handler future to get the single `ResponseEnvelope`, which
|
|
/// is written to the wire as `call.responded` (or `call.error`).
|
|
///
|
|
/// The handler runs concurrently — it consumes the
|
|
/// `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 (same-stream or
|
|
/// cross-stream), the handler future is dropped — the abort
|
|
/// cancels the Pub (ADR-046 §5, review 001 Unit 9). A cross-stream
|
|
/// abort arrives via `handle_abort` on a different `BiStream` and
|
|
/// signals through the shared `in_flight_sink_aborts` map; a
|
|
/// same-stream abort arrives as a `call.aborted` frame on this
|
|
/// stream. In both cases the handler future is dropped (not
|
|
/// awaited), `chunk_tx` is dropped (the `PublishStream` sees EOF),
|
|
/// and no response is written to the wire — the aborting initiator
|
|
/// already knows the request is cancelled.
|
|
pub(crate) async fn pump_sink<R, W>(
|
|
&self,
|
|
reader: &mut super::wire::FrameFramedReader<R>,
|
|
writer: &mut super::wire::FrameFramedWriter<W>,
|
|
request_id: String,
|
|
sink: SinkDispatch,
|
|
) where
|
|
R: tokio::io::AsyncRead + Unpin,
|
|
W: tokio::io::AsyncWrite + Unpin,
|
|
{
|
|
let SinkDispatch {
|
|
handler,
|
|
mut chunk_tx,
|
|
publish_validator,
|
|
} = sink;
|
|
|
|
let (abort_tx, mut abort_rx) = oneshot::channel::<()>();
|
|
self.in_flight_sink_aborts
|
|
.lock()
|
|
.insert(request_id.clone(), abort_tx);
|
|
|
|
use futures::FutureExt;
|
|
let mut handler = handler.fuse();
|
|
let mut response: Option<ResponseEnvelope> = None;
|
|
let mut aborted = false;
|
|
|
|
loop {
|
|
tokio::select! {
|
|
biased;
|
|
resp = &mut handler => {
|
|
response = Some(resp);
|
|
break;
|
|
}
|
|
_ = &mut abort_rx => {
|
|
aborted = true;
|
|
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;
|
|
}
|
|
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;
|
|
}
|
|
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 => {
|
|
aborted = true;
|
|
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"
|
|
);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
self.in_flight_sink_aborts.lock().remove(&request_id);
|
|
drop(chunk_tx);
|
|
|
|
if aborted {
|
|
drop(handler);
|
|
return;
|
|
}
|
|
|
|
let response = match response {
|
|
Some(r) => r,
|
|
None => handler.await,
|
|
};
|
|
|
|
let event: EventEnvelope = response.into();
|
|
if let Err(err) = writer.write_frame(&event).await {
|
|
warn!(error = %err, "failed to write sink response frame");
|
|
}
|
|
}
|
|
|
|
/// Run the shared dispatch loop over an established `CallConnection`:
|
|
/// spawn the pending-entry sweeper, accept bidirectional streams until the
|
|
/// connection closes, dispatch each stream via `handle_stream`, and fail
|
|
/// outstanding pending requests on close. Returns when the connection is
|
|
/// closed (accept loop yields `ConnectionClosed`/`StreamClosed`/`Timeout`).
|
|
pub async fn run_loop(self, connection: Arc<CallConnection>) {
|
|
let pending = Arc::clone(connection.pending());
|
|
|
|
let quic = match connection.connection() {
|
|
Some(c) => Arc::clone(c),
|
|
None => {
|
|
warn!("run_loop called with an overlay-only CallConnection; returning");
|
|
return;
|
|
}
|
|
};
|
|
|
|
let sweeper_pending = Arc::clone(&pending);
|
|
let sweeper_handle: JoinHandle<()> = tokio::spawn(async move {
|
|
let mut interval = tokio::time::interval(SWEEPER_INTERVAL);
|
|
interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
|
|
loop {
|
|
interval.tick().await;
|
|
let evicted = sweeper_pending.lock().evict_expired();
|
|
if !evicted.is_empty() {
|
|
debug!(
|
|
count = evicted.len(),
|
|
"sweeper evicted expired pending entries"
|
|
);
|
|
}
|
|
}
|
|
});
|
|
|
|
loop {
|
|
match quic.accept_bi().await {
|
|
Ok(stream) => {
|
|
let conn = Arc::clone(&connection);
|
|
let dispatcher = self.clone();
|
|
tokio::spawn(async move {
|
|
dispatcher.handle_stream(conn, stream).await;
|
|
});
|
|
}
|
|
Err(StreamError::ConnectionClosed) => break,
|
|
Err(StreamError::StreamClosed) => break,
|
|
Err(StreamError::Timeout) => break,
|
|
Err(err) => {
|
|
warn!(error = %err, "accept_bi error; stopping accept loop");
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
|
|
let failed = pending
|
|
.lock()
|
|
.fail_all(CallError::internal("connection closed"));
|
|
if !failed.is_empty() {
|
|
debug!(
|
|
count = failed.len(),
|
|
"failed pending requests on connection close"
|
|
);
|
|
}
|
|
|
|
sweeper_handle.abort();
|
|
}
|
|
|
|
/// Run the single-stream dispatch loop (ADR-036 amendment — channel
|
|
/// 0 single-stream mode). Reads `EventEnvelope` frames off `reader`
|
|
/// (channel 0's read half) and dispatches them, writing responses
|
|
/// through `writer` (the shared frame writer). All frames —
|
|
/// requests, responses, published chunks, aborts — are multiplexed
|
|
/// on the one `BiStream` (channel 0).
|
|
///
|
|
/// Unlike `run_loop` (stream-per-request), there is one read loop
|
|
/// for the whole connection. In-flight Pub (Sink) operations are
|
|
/// tracked by `request_id` in `in_flight_sinks`: their
|
|
/// `call.published` / `call.completed` / `call.aborted` frames
|
|
/// arrive *after* the `call.requested` and are routed to the
|
|
/// matching sink's `chunk_tx` while new `call.requested` frames for
|
|
/// other requests continue to be dispatched. Query/Mutation and Sub
|
|
/// responses are written through `writer` immediately after
|
|
/// dispatch.
|
|
///
|
|
/// Returns when the read half closes (transport EOF). Outstanding
|
|
/// pending requests are failed with `connection closed`, and
|
|
/// in-flight sinks' `chunk_tx` are dropped (the handler's
|
|
/// `PublishStream` sees EOF).
|
|
pub async fn run_loop_single_stream(
|
|
self,
|
|
connection: Arc<CallConnection>,
|
|
reader: Box<dyn tokio::io::AsyncRead + Send + Unpin>,
|
|
writer: Arc<super::connection::SharedFrameWriter>,
|
|
) {
|
|
let pending = Arc::clone(connection.pending());
|
|
|
|
let sweeper_pending = Arc::clone(&pending);
|
|
let sweeper_handle: JoinHandle<()> = tokio::spawn(async move {
|
|
let mut interval = tokio::time::interval(SWEEPER_INTERVAL);
|
|
interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
|
|
loop {
|
|
interval.tick().await;
|
|
let evicted = sweeper_pending.lock().evict_expired();
|
|
if !evicted.is_empty() {
|
|
debug!(
|
|
count = evicted.len(),
|
|
"sweeper evicted expired pending entries"
|
|
);
|
|
}
|
|
}
|
|
});
|
|
|
|
let mut reader = FrameFramedReader::new(reader);
|
|
let mut in_flight_sinks: HashMap<String, InFlightSink> = HashMap::new();
|
|
|
|
loop {
|
|
let envelope = match reader.read_frame().await {
|
|
Ok(env) => env,
|
|
Err(super::wire::FrameError::ConnectionClosed) => break,
|
|
Err(err) => {
|
|
warn!(error = %err, "single-stream frame read error; closing loop");
|
|
break;
|
|
}
|
|
};
|
|
|
|
match envelope.r#type.as_str() {
|
|
EVENT_REQUESTED => {
|
|
let request_id = envelope.id.clone();
|
|
let payload = envelope.payload.clone();
|
|
let dispatch_result = self
|
|
.dispatch(&connection, request_id.clone(), payload)
|
|
.await;
|
|
|
|
match dispatch_result {
|
|
DispatchResult::Once(response) => {
|
|
let event: EventEnvelope = response.into();
|
|
if let Err(err) = writer.write_frame(&event).await {
|
|
warn!(
|
|
error = %err,
|
|
"single-stream: failed to write Once response; closing loop"
|
|
);
|
|
break;
|
|
}
|
|
}
|
|
DispatchResult::Stream(stream) => {
|
|
self.pump_stream_single_stream(&writer, &request_id, stream)
|
|
.await;
|
|
}
|
|
DispatchResult::Sink(sink) => {
|
|
let SinkDispatch {
|
|
handler,
|
|
chunk_tx,
|
|
publish_validator,
|
|
} = sink;
|
|
let writer_clone = Arc::clone(&writer);
|
|
let request_id_for_handler = request_id.clone();
|
|
let handle = tokio::spawn(async move {
|
|
let response = handler.await;
|
|
let event: EventEnvelope = response.into();
|
|
if let Err(err) = writer_clone.write_frame(&event).await {
|
|
warn!(
|
|
error = %err,
|
|
request_id = %request_id_for_handler,
|
|
"single-stream: failed to write sink response frame"
|
|
);
|
|
}
|
|
});
|
|
in_flight_sinks.insert(
|
|
request_id.clone(),
|
|
InFlightSink {
|
|
chunk_tx,
|
|
publish_validator,
|
|
handler_handle: handle,
|
|
},
|
|
);
|
|
}
|
|
}
|
|
}
|
|
EVENT_ABORTED => {
|
|
let request_id = envelope.id.clone();
|
|
if let Some(mut entry) = in_flight_sinks.remove(&request_id) {
|
|
entry.handler_handle.abort();
|
|
let _ = entry
|
|
.chunk_tx
|
|
.send(Err(CallError::internal("publish aborted by initiator")))
|
|
.await;
|
|
} else {
|
|
self.handle_abort(&connection, &request_id).await;
|
|
}
|
|
}
|
|
EVENT_PUBLISHED => {
|
|
let request_id = envelope.id.clone();
|
|
let chunk = envelope
|
|
.payload
|
|
.get("input")
|
|
.cloned()
|
|
.unwrap_or(Value::Null);
|
|
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,
|
|
"single-stream: call.published for unknown in-flight sink; dropping"
|
|
);
|
|
}
|
|
}
|
|
EVENT_COMPLETED => {
|
|
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,
|
|
id = %envelope.id,
|
|
"single-stream: ignoring non-requested/non-published/non-aborted/non-completed event"
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
in_flight_sinks.clear();
|
|
|
|
let failed = pending
|
|
.lock()
|
|
.fail_all(CallError::internal("connection closed"));
|
|
if !failed.is_empty() {
|
|
debug!(
|
|
count = failed.len(),
|
|
"single-stream: failed pending requests on connection close"
|
|
);
|
|
}
|
|
|
|
sweeper_handle.abort();
|
|
}
|
|
|
|
/// Pump a subscription's `ResponseStream` to the wire through the
|
|
/// shared single-stream writer (ADR-036 amendment). Each
|
|
/// `ResponseEnvelope` becomes a `call.responded` / `call.error`
|
|
/// frame; on natural end, a `call.completed` frame. The shared
|
|
/// writer serializes frames so this pump's frames do not interleave
|
|
/// with concurrent calls' frames.
|
|
async fn pump_stream_single_stream(
|
|
&self,
|
|
writer: &Arc<super::connection::SharedFrameWriter>,
|
|
request_id: &str,
|
|
mut stream: ResponseStream,
|
|
) {
|
|
let mut last_was_error = false;
|
|
while let Some(envelope) = stream.next().await {
|
|
last_was_error = envelope.result.is_err();
|
|
let event: EventEnvelope = envelope.into();
|
|
if let Err(err) = writer.write_frame(&event).await {
|
|
warn!(error = %err, "single-stream: failed to write streaming frame");
|
|
return;
|
|
}
|
|
}
|
|
if !last_was_error {
|
|
let completed = EventEnvelope::completed(request_id);
|
|
if let Err(err) = writer.write_frame(&completed).await {
|
|
warn!(error = %err, "single-stream: failed to write call.completed");
|
|
}
|
|
}
|
|
}
|
|
|
|
/// The full-duplex single-stream serving loop (review 004 F-04):
|
|
/// composes the dispatch arms (`run_loop_single_stream`) and the
|
|
/// pending-resolution arms (`read_single_stream_until_closed`) in
|
|
/// one loop. Both sides of a channels connection can be both
|
|
/// producer and consumer (AGENTS.md §8; ADR-022 §2) — on channel 0
|
|
/// the two directions' frames are multiplexed on one byte stream,
|
|
/// so the loop branches per frame:
|
|
///
|
|
/// - `call.requested` → dispatch against this loop's registry and
|
|
/// write the response frame(s) (the serving half — what the
|
|
/// accept side's `run_loop_single_stream` does).
|
|
/// - `call.responded` / `call.completed` / `call.error` → resolve
|
|
/// the matching outbound pending entry (the calling half — what
|
|
/// the client read pump does). `call.responded` frames for
|
|
/// *inbound* Sub responses are `call.responded` too; direction is
|
|
/// disambiguated by the pending map (an id that is one of *our*
|
|
/// outbound pendings resolves there; an unknown id is dropped
|
|
/// with a debug line, never an error — the peer may legitimately
|
|
/// stream a Sub's responses that this side does not track).
|
|
/// - `call.aborted` → try both tables: the in-flight sink aborts
|
|
/// *and* the outbound pending map's abort-cascade
|
|
/// (`handle_abort`), then fall through to serving-side sink
|
|
/// cancellation (the `run_loop_single_stream` arm) when the id
|
|
/// matches an inbound sink. An id in neither table is a no-op.
|
|
/// - `call.published` / `call.completed` (initiator→responder) →
|
|
/// route to the matching inbound in-flight sink; ids that are
|
|
/// outbound pending *call* entries whose `call.responded` already
|
|
/// removed them are no-ops.
|
|
///
|
|
/// IDs are UUID-generated per side (`generate_request_id()`), so
|
|
/// cross-correlation between the two directions is not a hazard.
|
|
///
|
|
/// On read-half close, outbound pendings are failed (`connection
|
|
/// closed`) and in-flight inbound sinks are dropped — the same
|
|
/// teardown both arms perform separately today.
|
|
pub async fn serve_single_stream(
|
|
self,
|
|
connection: Arc<CallConnection>,
|
|
reader: Box<dyn tokio::io::AsyncRead + Send + Unpin>,
|
|
writer: Arc<super::connection::SharedFrameWriter>,
|
|
) {
|
|
let pending = Arc::clone(connection.pending());
|
|
|
|
let sweeper_pending = Arc::clone(&pending);
|
|
let sweeper_handle: JoinHandle<()> = tokio::spawn(async move {
|
|
let mut interval = tokio::time::interval(SWEEPER_INTERVAL);
|
|
interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
|
|
loop {
|
|
interval.tick().await;
|
|
let evicted = sweeper_pending.lock().evict_expired();
|
|
if !evicted.is_empty() {
|
|
debug!(
|
|
count = evicted.len(),
|
|
"serve loop: sweeper evicted expired pending entries"
|
|
);
|
|
}
|
|
}
|
|
});
|
|
|
|
let mut reader = FrameFramedReader::new(reader);
|
|
let mut in_flight_sinks: HashMap<String, InFlightSink> = HashMap::new();
|
|
|
|
loop {
|
|
let envelope = match reader.read_frame().await {
|
|
Ok(env) => env,
|
|
Err(super::wire::FrameError::ConnectionClosed) => break,
|
|
Err(err) => {
|
|
warn!(error = %err, "serve loop: frame read error; closing loop");
|
|
break;
|
|
}
|
|
};
|
|
|
|
match envelope.r#type.as_str() {
|
|
EVENT_REQUESTED => {
|
|
let request_id = envelope.id.clone();
|
|
let payload = envelope.payload.clone();
|
|
let dispatch_result = self
|
|
.dispatch(&connection, request_id.clone(), payload)
|
|
.await;
|
|
|
|
match dispatch_result {
|
|
DispatchResult::Once(response) => {
|
|
let event: EventEnvelope = response.into();
|
|
if let Err(err) = writer.write_frame(&event).await {
|
|
warn!(
|
|
error = %err,
|
|
"serve loop: failed to write Once response; closing loop"
|
|
);
|
|
break;
|
|
}
|
|
}
|
|
DispatchResult::Stream(stream) => {
|
|
self.pump_stream_single_stream(&writer, &request_id, stream)
|
|
.await;
|
|
}
|
|
DispatchResult::Sink(sink) => {
|
|
let SinkDispatch {
|
|
handler,
|
|
chunk_tx,
|
|
publish_validator,
|
|
} = sink;
|
|
let writer_clone = Arc::clone(&writer);
|
|
let request_id_for_handler = request_id.clone();
|
|
let handle = tokio::spawn(async move {
|
|
let response = handler.await;
|
|
let event: EventEnvelope = response.into();
|
|
if let Err(err) = writer_clone.write_frame(&event).await {
|
|
warn!(
|
|
error = %err,
|
|
request_id = %request_id_for_handler,
|
|
"serve loop: failed to write sink response frame"
|
|
);
|
|
}
|
|
});
|
|
in_flight_sinks.insert(
|
|
request_id.clone(),
|
|
InFlightSink {
|
|
chunk_tx,
|
|
publish_validator,
|
|
handler_handle: handle,
|
|
},
|
|
);
|
|
}
|
|
}
|
|
}
|
|
EVENT_RESPONDED => {
|
|
let request_id = envelope.id.clone();
|
|
let output = envelope
|
|
.payload
|
|
.get("output")
|
|
.cloned()
|
|
.unwrap_or(Value::Null);
|
|
pending.lock().handle_responded(&request_id, output);
|
|
}
|
|
EVENT_COMPLETED => {
|
|
let request_id = envelope.id.clone();
|
|
if in_flight_sinks.remove(&request_id).is_none() {
|
|
pending.lock().handle_completed(&request_id);
|
|
}
|
|
}
|
|
EVENT_ABORTED => {
|
|
let request_id = envelope.id.clone();
|
|
if let Some(mut entry) = in_flight_sinks.remove(&request_id) {
|
|
entry.handler_handle.abort();
|
|
let _ = entry
|
|
.chunk_tx
|
|
.send(Err(CallError::internal("publish aborted by initiator")))
|
|
.await;
|
|
} else {
|
|
self.handle_abort(&connection, &request_id).await;
|
|
}
|
|
}
|
|
EVENT_PUBLISHED => {
|
|
let request_id = envelope.id.clone();
|
|
let chunk = envelope
|
|
.payload
|
|
.get("input")
|
|
.cloned()
|
|
.unwrap_or(Value::Null);
|
|
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,
|
|
"serve loop: call.published for unknown in-flight sink; dropping"
|
|
);
|
|
}
|
|
}
|
|
EVENT_ERROR => {
|
|
let request_id = envelope.id.clone();
|
|
let call_error: CallError = serde_json::from_value(envelope.payload)
|
|
.unwrap_or_else(|_| {
|
|
CallError::internal("publish error from initiator (malformed)")
|
|
});
|
|
if let Some(mut entry) = in_flight_sinks.remove(&request_id) {
|
|
let _ = entry.chunk_tx.send(Err(call_error)).await;
|
|
} else {
|
|
pending.lock().handle_error(&request_id, call_error);
|
|
}
|
|
}
|
|
other => {
|
|
debug!(
|
|
event_type = %other,
|
|
id = %envelope.id,
|
|
"serve loop: ignoring unknown event type"
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
in_flight_sinks.clear();
|
|
|
|
let failed = pending
|
|
.lock()
|
|
.fail_all(CallError::internal("connection closed"));
|
|
if !failed.is_empty() {
|
|
debug!(
|
|
count = failed.len(),
|
|
"serve loop: failed pending requests on connection close"
|
|
);
|
|
}
|
|
|
|
sweeper_handle.abort();
|
|
}
|
|
}
|
|
|
|
impl Clone for Dispatcher {
|
|
fn clone(&self) -> Self {
|
|
Self {
|
|
registry: Arc::clone(&self.registry),
|
|
identity_provider: Arc::clone(&self.identity_provider),
|
|
session_source: self.session_source.clone(),
|
|
ownership_provider: self.ownership_provider.clone(),
|
|
default_timeout: self.default_timeout,
|
|
in_flight_sink_aborts: Arc::clone(&self.in_flight_sink_aborts),
|
|
}
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use crate::core::auth::{AuthToken, Identity, IdentityProvider};
|
|
use crate::core::types::Capabilities;
|
|
use crate::protocol::wire::{EVENT_COMPLETED, EVENT_ERROR, EVENT_RESPONDED};
|
|
use crate::registry::registration::{
|
|
make_handler, make_streaming_handler, HandlerKind, HandlerRegistration, OperationProvenance,
|
|
};
|
|
use crate::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
|
|
use std::collections::HashMap;
|
|
use std::sync::Mutex as StdMutex;
|
|
|
|
use crate::protocol::sink_empty_connection as stub_connection;
|
|
|
|
struct StaticIdentityProvider {
|
|
tokens: StdMutex<HashMap<String, Identity>>,
|
|
}
|
|
|
|
impl StaticIdentityProvider {
|
|
fn new() -> Self {
|
|
Self {
|
|
tokens: StdMutex::new(HashMap::new()),
|
|
}
|
|
}
|
|
|
|
fn with_token(self, token: &str, identity: Identity) -> Self {
|
|
self.tokens
|
|
.lock()
|
|
.unwrap()
|
|
.insert(token.to_string(), identity);
|
|
self
|
|
}
|
|
}
|
|
|
|
impl IdentityProvider for StaticIdentityProvider {
|
|
fn resolve_from_fingerprint(&self, _fp: &str) -> Option<Identity> {
|
|
None
|
|
}
|
|
fn resolve_from_token(&self, token: &AuthToken) -> Option<Identity> {
|
|
let token_str = String::from_utf8_lossy(&token.raw);
|
|
self.tokens.lock().unwrap().get(token_str.as_ref()).cloned()
|
|
}
|
|
}
|
|
|
|
fn identity_with_scopes(id: &str, scopes: &[&str]) -> Identity {
|
|
Identity {
|
|
id: id.to_string(),
|
|
scopes: scopes.iter().map(|s| s.to_string()).collect(),
|
|
resources: HashMap::new(),
|
|
}
|
|
}
|
|
|
|
fn external_spec(name: &str, acl: AccessControl) -> OperationSpec {
|
|
OperationSpec::new(
|
|
name,
|
|
OperationType::Query,
|
|
Visibility::External,
|
|
serde_json::json!({}),
|
|
serde_json::json!({}),
|
|
vec![],
|
|
acl,
|
|
None,
|
|
)
|
|
}
|
|
|
|
fn registry_with(name: &str, visibility: Visibility, acl: AccessControl) -> OperationRegistry {
|
|
let registry = OperationRegistry::new();
|
|
registry
|
|
.register(HandlerRegistration::new(
|
|
OperationSpec::new(
|
|
name,
|
|
OperationType::Query,
|
|
visibility,
|
|
serde_json::json!({}),
|
|
serde_json::json!({}),
|
|
vec![],
|
|
acl,
|
|
None,
|
|
),
|
|
HandlerKind::Once(make_handler(|input, context| async move {
|
|
ResponseEnvelope::ok(context.request_id, input)
|
|
})),
|
|
OperationProvenance::Local,
|
|
None,
|
|
None,
|
|
Capabilities::new(),
|
|
))
|
|
.unwrap();
|
|
registry
|
|
}
|
|
|
|
fn dispatcher() -> Dispatcher {
|
|
Dispatcher::new(
|
|
Arc::new(OperationRegistry::new()),
|
|
Arc::new(StaticIdentityProvider::new()),
|
|
)
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn dispatch_authorized_peer_dispatches_and_populates_capabilities() {
|
|
let caps = Capabilities::new().with_api_key("google", "k".to_string());
|
|
let registry = OperationRegistry::new();
|
|
let handler = make_handler(|_input, context| async move {
|
|
let has_google = context.capabilities.get("google").is_some();
|
|
ResponseEnvelope::ok(
|
|
context.request_id,
|
|
serde_json::json!({ "has_google": has_google }),
|
|
)
|
|
});
|
|
registry
|
|
.register(HandlerRegistration::new(
|
|
external_spec("admin/run", AccessControl::default()),
|
|
HandlerKind::Once(handler),
|
|
OperationProvenance::Local,
|
|
None,
|
|
None,
|
|
caps,
|
|
))
|
|
.unwrap();
|
|
let registry = Arc::new(registry);
|
|
let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
|
|
let dp = Dispatcher::new(registry, provider);
|
|
let conn = Arc::new(CallConnection::new(stub_connection()));
|
|
|
|
let payload = serde_json::json!({
|
|
"operationId": "/admin/run",
|
|
"input": {},
|
|
});
|
|
let response = dp
|
|
.dispatch_requested(&conn, "req-1".to_string(), payload)
|
|
.await;
|
|
let out = response.result.expect("dispatch ok");
|
|
assert_eq!(out["has_google"], Value::Bool(true));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn dispatch_unauthorized_peer_returns_forbidden_capabilities_never_populated() {
|
|
let caps = Capabilities::new().with_api_key("google", "k".to_string());
|
|
let registry = OperationRegistry::new();
|
|
let handler = make_handler(|_input, context| async move {
|
|
let has_google = context.capabilities.get("google").is_some();
|
|
ResponseEnvelope::ok(
|
|
context.request_id,
|
|
serde_json::json!({ "has_google": has_google }),
|
|
)
|
|
});
|
|
registry
|
|
.register(HandlerRegistration::new(
|
|
external_spec(
|
|
"admin/run",
|
|
AccessControl {
|
|
required_scopes: vec!["admin".to_string()],
|
|
..Default::default()
|
|
},
|
|
),
|
|
HandlerKind::Once(handler),
|
|
OperationProvenance::Local,
|
|
None,
|
|
None,
|
|
caps,
|
|
))
|
|
.unwrap();
|
|
let registry = Arc::new(registry);
|
|
let provider: Arc<dyn IdentityProvider> = Arc::new(
|
|
StaticIdentityProvider::new()
|
|
.with_token("alk_user", identity_with_scopes("regular-user", &["user"])),
|
|
);
|
|
let dp = Dispatcher::new(registry, provider);
|
|
let conn = Arc::new(CallConnection::new(stub_connection()));
|
|
|
|
let payload = serde_json::json!({
|
|
"operationId": "/admin/run",
|
|
"input": {},
|
|
"auth_token": "alk_user",
|
|
});
|
|
let response = dp
|
|
.dispatch_requested(&conn, "req-2".to_string(), payload)
|
|
.await;
|
|
match response.result {
|
|
Err(e) => {
|
|
assert_eq!(e.code, "FORBIDDEN");
|
|
assert!(e.message.contains("admin"));
|
|
}
|
|
other => panic!("expected FORBIDDEN, got {other:?}"),
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn dispatch_internal_op_from_wire_returns_not_found_before_acl() {
|
|
let registry = Arc::new(registry_with(
|
|
"secret/op",
|
|
Visibility::Internal,
|
|
AccessControl::default(),
|
|
));
|
|
let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
|
|
let dp = Dispatcher::new(registry, provider);
|
|
let conn = Arc::new(CallConnection::new(stub_connection()));
|
|
|
|
let payload = serde_json::json!({
|
|
"operationId": "/secret/op",
|
|
"input": {},
|
|
});
|
|
let response = dp
|
|
.dispatch_requested(&conn, "req-3".to_string(), payload)
|
|
.await;
|
|
match response.result {
|
|
Err(e) => {
|
|
assert_eq!(e.code, "NOT_FOUND");
|
|
assert!(e.message.contains("secret/op"));
|
|
}
|
|
other => panic!("expected NOT_FOUND, got {other:?}"),
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn dispatch_connection_with_no_identity_produces_no_peer_id_in_env() {
|
|
let registry = Arc::new(registry_with(
|
|
"fs/readFile",
|
|
Visibility::External,
|
|
AccessControl::default(),
|
|
));
|
|
let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
|
|
let dp = Dispatcher::new(registry, provider);
|
|
let conn = CallConnection::new(stub_connection());
|
|
|
|
let context = dp.build_root_context("req-4".to_string(), "fs/readFile", None, None, &conn);
|
|
|
|
assert!(
|
|
context.identity.is_none(),
|
|
"no connection identity → context.identity is None"
|
|
);
|
|
assert!(
|
|
context.env.peer_ids().is_empty(),
|
|
"no peer overlay attached when connection has no identity"
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn dispatch_connection_with_identity_attaches_peer_overlay_keyed_by_identity_id() {
|
|
let registry = Arc::new(registry_with(
|
|
"fs/readFile",
|
|
Visibility::External,
|
|
AccessControl::default(),
|
|
));
|
|
let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
|
|
let dp = Dispatcher::new(registry, provider);
|
|
let conn = CallConnection::new(stub_connection());
|
|
conn.connection()
|
|
.expect("quic connection present")
|
|
.set_identity(identity_with_scopes("worker-a", &[]))
|
|
.expect("identity not yet set");
|
|
|
|
let context = dp.build_root_context("req-5".to_string(), "fs/readFile", None, None, &conn);
|
|
|
|
assert_eq!(
|
|
context.env.peer_ids(),
|
|
vec!["worker-a".to_string()],
|
|
"PeerId for connection comes from connection.identity().id"
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn dispatch_extract_forwarded_for_from_payload_into_context() {
|
|
let registry = OperationRegistry::new();
|
|
let handler = make_handler(|_input, context| async move {
|
|
let forwarded_id = context.forwarded_for.as_ref().map(|i| i.id.clone());
|
|
ResponseEnvelope::ok(
|
|
context.request_id,
|
|
serde_json::json!({ "forwarded_for_id": forwarded_id }),
|
|
)
|
|
});
|
|
registry
|
|
.register(HandlerRegistration::new(
|
|
external_spec("fs/readFile", AccessControl::default()),
|
|
HandlerKind::Once(handler),
|
|
OperationProvenance::Local,
|
|
None,
|
|
None,
|
|
Capabilities::new(),
|
|
))
|
|
.unwrap();
|
|
let registry = Arc::new(registry);
|
|
let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
|
|
let dp = Dispatcher::new(registry, provider);
|
|
let conn = Arc::new(CallConnection::new(stub_connection()));
|
|
|
|
let payload = serde_json::json!({
|
|
"operationId": "/fs/readFile",
|
|
"input": {},
|
|
"forwarded_for": {
|
|
"id": "alice",
|
|
"scopes": ["fs:read"],
|
|
"resources": {}
|
|
},
|
|
});
|
|
let response = dp
|
|
.dispatch_requested(&conn, "req-6".to_string(), payload)
|
|
.await;
|
|
let out = response.result.expect("ok");
|
|
assert_eq!(out["forwarded_for_id"], Value::String("alice".into()));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn dispatch_without_forwarded_for_field_is_none() {
|
|
let registry = OperationRegistry::new();
|
|
let handler = make_handler(|_input, context| async move {
|
|
let present = context.forwarded_for.is_some();
|
|
ResponseEnvelope::ok(
|
|
context.request_id,
|
|
serde_json::json!({ "present": present }),
|
|
)
|
|
});
|
|
registry
|
|
.register(HandlerRegistration::new(
|
|
external_spec("fs/readFile", AccessControl::default()),
|
|
HandlerKind::Once(handler),
|
|
OperationProvenance::Local,
|
|
None,
|
|
None,
|
|
Capabilities::new(),
|
|
))
|
|
.unwrap();
|
|
let registry = Arc::new(registry);
|
|
let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
|
|
let dp = Dispatcher::new(registry, provider);
|
|
let conn = Arc::new(CallConnection::new(stub_connection()));
|
|
|
|
let payload = serde_json::json!({
|
|
"operationId": "/fs/readFile",
|
|
"input": {},
|
|
});
|
|
let response = dp
|
|
.dispatch_requested(&conn, "req-7".to_string(), payload)
|
|
.await;
|
|
let out = response.result.expect("ok");
|
|
assert_eq!(out["present"], Value::Bool(false));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn dispatch_default_access_control_dispatches_to_any_peer() {
|
|
let registry = Arc::new(registry_with(
|
|
"echo/run",
|
|
Visibility::External,
|
|
AccessControl::default(),
|
|
));
|
|
let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
|
|
let dp = Dispatcher::new(registry, provider);
|
|
let conn = Arc::new(CallConnection::new(stub_connection()));
|
|
|
|
let payload = serde_json::json!({
|
|
"operationId": "/echo/run",
|
|
"input": { "msg": "hi" },
|
|
});
|
|
let response = dp
|
|
.dispatch_requested(&conn, "req-8".to_string(), payload)
|
|
.await;
|
|
assert_eq!(response.result, Ok(serde_json::json!({ "msg": "hi" })));
|
|
}
|
|
|
|
#[test]
|
|
fn dispatcher_helper_compiles_with_full_signature() {
|
|
let _dp = dispatcher();
|
|
}
|
|
|
|
// --- non-QUIC (overlay-only) dispatch path ----------------------------
|
|
|
|
fn overlay_only_connection(identity: Identity) -> Arc<CallConnection> {
|
|
Arc::new(CallConnection::new_overlay_only(identity))
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn dispatch_requested_works_with_overlay_only_connection() {
|
|
let registry = Arc::new(registry_with(
|
|
"echo/run",
|
|
Visibility::External,
|
|
AccessControl::default(),
|
|
));
|
|
let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
|
|
let dp = Dispatcher::new(registry, provider);
|
|
let conn = overlay_only_connection(identity_with_scopes("ws-peer", &[]));
|
|
|
|
let payload = serde_json::json!({
|
|
"operationId": "/echo/run",
|
|
"input": { "msg": "hello" },
|
|
});
|
|
let response = dp
|
|
.dispatch_requested(&conn, "ws-req-1".to_string(), payload)
|
|
.await;
|
|
assert_eq!(response.request_id, "ws-req-1");
|
|
assert_eq!(response.result, Ok(serde_json::json!({ "msg": "hello" })));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn dispatch_requested_overlay_only_attaches_peer_keyed_by_stored_identity() {
|
|
let registry = OperationRegistry::new();
|
|
let handler = make_handler(|_input, context| async move {
|
|
let peer_ids = context.env.peer_ids();
|
|
ResponseEnvelope::ok(
|
|
context.request_id,
|
|
serde_json::json!({ "peer_ids": peer_ids }),
|
|
)
|
|
});
|
|
registry
|
|
.register(HandlerRegistration::new(
|
|
external_spec("fs/readFile", AccessControl::default()),
|
|
HandlerKind::Once(handler),
|
|
OperationProvenance::Local,
|
|
None,
|
|
None,
|
|
Capabilities::new(),
|
|
))
|
|
.unwrap();
|
|
let registry = Arc::new(registry);
|
|
let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
|
|
let dp = Dispatcher::new(registry, provider);
|
|
let conn = overlay_only_connection(identity_with_scopes("ws-peer", &[]));
|
|
|
|
let payload = serde_json::json!({
|
|
"operationId": "/fs/readFile",
|
|
"input": {},
|
|
});
|
|
let response = dp
|
|
.dispatch_requested(&conn, "ws-req-2".to_string(), payload)
|
|
.await;
|
|
let out = response.result.expect("ok");
|
|
assert_eq!(out["peer_ids"], serde_json::json!(["ws-peer"]));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn dispatch_requested_overlay_only_unknown_op_returns_not_found() {
|
|
let registry = Arc::new(OperationRegistry::new());
|
|
let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
|
|
let dp = Dispatcher::new(registry, provider);
|
|
let conn = overlay_only_connection(identity_with_scopes("ws-peer", &[]));
|
|
|
|
let payload = serde_json::json!({
|
|
"operationId": "/no/such/op",
|
|
"input": {},
|
|
});
|
|
let response = dp
|
|
.dispatch_requested(&conn, "ws-req-3".to_string(), payload)
|
|
.await;
|
|
match response.result {
|
|
Err(e) => assert_eq!(e.code, "NOT_FOUND"),
|
|
other => panic!("expected NOT_FOUND, got {other:?}"),
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn handle_abort_works_with_overlay_only_connection() {
|
|
let registry = Arc::new(registry_with(
|
|
"echo/run",
|
|
Visibility::External,
|
|
AccessControl::default(),
|
|
));
|
|
let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
|
|
let dp = Dispatcher::new(registry, provider);
|
|
let conn = overlay_only_connection(identity_with_scopes("ws-peer", &[]));
|
|
|
|
let parent_id = "ws-abort-root".to_string();
|
|
let child_id = "ws-abort-child".to_string();
|
|
{
|
|
let mut pending = conn.pending().lock();
|
|
pending.register_call(
|
|
parent_id.clone(),
|
|
Some(Instant::now() + Duration::from_secs(30)),
|
|
None,
|
|
);
|
|
pending.register_call(
|
|
child_id.clone(),
|
|
Some(Instant::now() + Duration::from_secs(30)),
|
|
Some(parent_id.clone()),
|
|
);
|
|
}
|
|
assert!(conn.pending().lock().contains(&parent_id));
|
|
assert!(conn.pending().lock().contains(&child_id));
|
|
|
|
dp.handle_abort(&conn, &parent_id).await;
|
|
|
|
assert!(
|
|
!conn.pending().lock().contains(&parent_id),
|
|
"parent entry removed after abort"
|
|
);
|
|
assert!(
|
|
!conn.pending().lock().contains(&child_id),
|
|
"child aborted by cascade"
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn handle_abort_unknown_request_id_is_noop_for_overlay_only() {
|
|
let registry = Arc::new(OperationRegistry::new());
|
|
let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
|
|
let dp = Dispatcher::new(registry, provider);
|
|
let conn = overlay_only_connection(identity_with_scopes("ws-peer", &[]));
|
|
|
|
dp.handle_abort(&conn, "totally-unknown").await;
|
|
assert!(conn.pending().lock().is_empty());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn overlay_only_full_dispatch_round_trip_returns_response_envelope() {
|
|
let registry = Arc::new(registry_with(
|
|
"echo/run",
|
|
Visibility::External,
|
|
AccessControl::default(),
|
|
));
|
|
let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
|
|
let dp = Dispatcher::new(registry, provider);
|
|
let conn = overlay_only_connection(identity_with_scopes("ws-peer", &[]));
|
|
|
|
let payload = serde_json::json!({
|
|
"operationId": "/echo/run",
|
|
"input": { "v": 42 },
|
|
});
|
|
let request_id = "ws-roundtrip-1".to_string();
|
|
let response = dp
|
|
.dispatch_requested(&conn, request_id.clone(), payload)
|
|
.await;
|
|
assert!(response.result.is_ok());
|
|
let envelope: EventEnvelope = response.into();
|
|
assert_eq!(envelope.r#type, EVENT_RESPONDED);
|
|
assert_eq!(envelope.id, "ws-roundtrip-1");
|
|
assert_eq!(
|
|
envelope.payload.get("output"),
|
|
Some(&serde_json::json!({ "v": 42 }))
|
|
);
|
|
}
|
|
|
|
// --- streaming dispatch branch (ADR-021 §6) ---------------------------
|
|
|
|
fn subscription_spec(name: &str, acl: AccessControl) -> OperationSpec {
|
|
OperationSpec::new(
|
|
name,
|
|
OperationType::Sub,
|
|
Visibility::External,
|
|
serde_json::json!({}),
|
|
serde_json::json!({}),
|
|
vec![],
|
|
acl,
|
|
None,
|
|
)
|
|
}
|
|
|
|
fn encode_frame(envelope: &EventEnvelope) -> Vec<u8> {
|
|
let body = serde_json::to_vec(envelope).expect("serialize envelope");
|
|
let mut buf = (body.len() as u32).to_be_bytes().to_vec();
|
|
buf.extend_from_slice(&body);
|
|
buf
|
|
}
|
|
|
|
async fn read_all_frames(
|
|
reader: &mut (impl tokio::io::AsyncRead + Unpin),
|
|
) -> Vec<EventEnvelope> {
|
|
let mut buf = Vec::new();
|
|
use tokio::io::AsyncReadExt;
|
|
let _ = reader.read_to_end(&mut buf).await;
|
|
let mut frames = Vec::new();
|
|
let mut cursor = std::io::Cursor::new(buf);
|
|
loop {
|
|
let mut len_buf = [0u8; 4];
|
|
match tokio::io::AsyncReadExt::read_exact(&mut cursor, &mut len_buf).await {
|
|
Ok(_) => {}
|
|
Err(_) => break,
|
|
}
|
|
let len = u32::from_be_bytes(len_buf) as usize;
|
|
let mut body = vec![0u8; len];
|
|
if tokio::io::AsyncReadExt::read_exact(&mut cursor, &mut body)
|
|
.await
|
|
.is_err()
|
|
{
|
|
break;
|
|
}
|
|
let envelope: EventEnvelope =
|
|
serde_json::from_slice(&body).expect("deserialize written frame");
|
|
frames.push(envelope);
|
|
}
|
|
frames
|
|
}
|
|
|
|
fn registry_with_subscription(
|
|
name: &str,
|
|
handler: crate::registry::registration::StreamingHandler,
|
|
) -> Arc<OperationRegistry> {
|
|
let registry = OperationRegistry::new();
|
|
registry
|
|
.register(HandlerRegistration::new(
|
|
subscription_spec(name, AccessControl::default()),
|
|
HandlerKind::Stream(handler),
|
|
OperationProvenance::Local,
|
|
None,
|
|
None,
|
|
Capabilities::new(),
|
|
))
|
|
.unwrap();
|
|
Arc::new(registry)
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn dispatch_subscription_returns_stream_result() {
|
|
let handler = make_streaming_handler(|input, ctx| {
|
|
futures::stream::iter(vec![
|
|
ResponseEnvelope::ok(ctx.request_id.clone(), input.clone()),
|
|
ResponseEnvelope::ok(ctx.request_id.clone(), serde_json::json!({"done": true})),
|
|
])
|
|
});
|
|
let registry = registry_with_subscription("events/stream", handler);
|
|
let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
|
|
let dp = Dispatcher::new(registry, provider);
|
|
let conn = Arc::new(CallConnection::new(stub_connection()));
|
|
|
|
let payload = serde_json::json!({
|
|
"operationId": "/events/stream",
|
|
"input": { "v": 1 },
|
|
});
|
|
match dp.dispatch(&conn, "sub-1".to_string(), payload).await {
|
|
DispatchResult::Stream(mut stream) => {
|
|
use futures::stream::StreamExt;
|
|
let first = stream.next().await.expect("first envelope");
|
|
assert_eq!(first.request_id, "sub-1");
|
|
assert_eq!(first.result, Ok(serde_json::json!({ "v": 1 })));
|
|
let second = stream.next().await.expect("second envelope");
|
|
assert_eq!(second.result, Ok(serde_json::json!({ "done": true })));
|
|
assert!(
|
|
stream.next().await.is_none(),
|
|
"stream ends after two values"
|
|
);
|
|
}
|
|
other => panic!("expected Stream, got {other:?}"),
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn dispatch_subscription_clears_deadline_to_none() {
|
|
let handler = make_streaming_handler(|_input, ctx| {
|
|
let deadline = ctx.deadline;
|
|
futures::stream::iter(vec![ResponseEnvelope::ok(
|
|
ctx.request_id.clone(),
|
|
serde_json::json!({ "deadline_is_none": deadline.is_none() }),
|
|
)])
|
|
});
|
|
let registry = registry_with_subscription("events/stream", handler);
|
|
let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
|
|
let dp = Dispatcher::new(registry, provider);
|
|
let conn = Arc::new(CallConnection::new(stub_connection()));
|
|
|
|
let payload = serde_json::json!({
|
|
"operationId": "/events/stream",
|
|
"input": {},
|
|
});
|
|
match dp.dispatch(&conn, "sub-dl".to_string(), payload).await {
|
|
DispatchResult::Stream(mut stream) => {
|
|
use futures::stream::StreamExt;
|
|
let env = stream.next().await.expect("one envelope");
|
|
let out = env.result.expect("ok");
|
|
assert_eq!(out["deadline_is_none"], Value::Bool(true));
|
|
}
|
|
other => panic!("expected Stream, got {other:?}"),
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn dispatch_query_keeps_deadline_some() {
|
|
let registry = OperationRegistry::new();
|
|
let handler = make_handler(|_input, ctx| async move {
|
|
let deadline_is_some = ctx.deadline.is_some();
|
|
ResponseEnvelope::ok(
|
|
ctx.request_id.clone(),
|
|
serde_json::json!({ "deadline_is_some": deadline_is_some }),
|
|
)
|
|
});
|
|
registry
|
|
.register(HandlerRegistration::new(
|
|
external_spec("echo/run", AccessControl::default()),
|
|
HandlerKind::Once(handler),
|
|
OperationProvenance::Local,
|
|
None,
|
|
None,
|
|
Capabilities::new(),
|
|
))
|
|
.unwrap();
|
|
let registry = Arc::new(registry);
|
|
let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
|
|
let dp = Dispatcher::new(registry, provider);
|
|
let conn = Arc::new(CallConnection::new(stub_connection()));
|
|
|
|
let payload = serde_json::json!({
|
|
"operationId": "/echo/run",
|
|
"input": {},
|
|
});
|
|
match dp.dispatch(&conn, "q-1".to_string(), payload).await {
|
|
DispatchResult::Once(env) => {
|
|
let out = env.result.expect("ok");
|
|
assert_eq!(out["deadline_is_some"], Value::Bool(true));
|
|
}
|
|
other => panic!("expected Once, got {other:?}"),
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn handle_stream_subscription_pumps_each_frame_then_completed() {
|
|
let handler = make_streaming_handler(|input, ctx| {
|
|
let first = input.clone();
|
|
let rid = ctx.request_id.clone();
|
|
futures::stream::iter(vec![
|
|
ResponseEnvelope::ok(rid.clone(), first),
|
|
ResponseEnvelope::ok(rid.clone(), serde_json::json!({"n": 2})),
|
|
ResponseEnvelope::ok(rid, serde_json::json!({"n": 3})),
|
|
])
|
|
});
|
|
let registry = registry_with_subscription("events/stream", handler);
|
|
let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
|
|
let dp = Dispatcher::new(registry, provider);
|
|
let conn = Arc::new(CallConnection::new(stub_connection()));
|
|
|
|
let request = EventEnvelope::requested(
|
|
"sub-pump-1",
|
|
serde_json::json!({
|
|
"operationId": "/events/stream",
|
|
"input": { "n": 1 },
|
|
}),
|
|
);
|
|
let recv = tokio::io::BufReader::new(std::io::Cursor::new(encode_frame(&request)));
|
|
let (send, mut sink) = tokio::io::duplex(8 * 1024);
|
|
let stream = crate::core::types::BiStream::from_joined(recv, send);
|
|
|
|
dp.handle_stream(conn, stream).await;
|
|
|
|
let frames = read_all_frames(&mut sink).await;
|
|
assert_eq!(frames.len(), 4, "3 responded + 1 completed");
|
|
for (i, f) in frames[..3].iter().enumerate() {
|
|
assert_eq!(f.r#type, EVENT_RESPONDED, "frame {i} is call.responded");
|
|
assert_eq!(f.id, "sub-pump-1");
|
|
}
|
|
assert_eq!(frames[3].r#type, EVENT_COMPLETED);
|
|
assert_eq!(frames[3].id, "sub-pump-1");
|
|
assert_eq!(frames[3].payload, serde_json::json!({}));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn handle_stream_subscription_error_is_terminal_no_completed() {
|
|
let handler = make_streaming_handler(|_input, ctx| {
|
|
let rid = ctx.request_id.clone();
|
|
futures::stream::iter(vec![
|
|
ResponseEnvelope::ok(rid.clone(), serde_json::json!({"ok": true})),
|
|
ResponseEnvelope::error(rid.clone(), CallError::internal("boom")),
|
|
])
|
|
});
|
|
let registry = registry_with_subscription("events/stream", handler);
|
|
let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
|
|
let dp = Dispatcher::new(registry, provider);
|
|
let conn = Arc::new(CallConnection::new(stub_connection()));
|
|
|
|
let request = EventEnvelope::requested(
|
|
"sub-err-1",
|
|
serde_json::json!({
|
|
"operationId": "/events/stream",
|
|
"input": {},
|
|
}),
|
|
);
|
|
let recv = tokio::io::BufReader::new(std::io::Cursor::new(encode_frame(&request)));
|
|
let (send, mut sink) = tokio::io::duplex(8 * 1024);
|
|
let stream = crate::core::types::BiStream::from_joined(recv, send);
|
|
|
|
dp.handle_stream(conn, stream).await;
|
|
|
|
let frames = read_all_frames(&mut sink).await;
|
|
assert_eq!(frames.len(), 2, "1 responded + 1 error, no completed");
|
|
assert_eq!(frames[0].r#type, EVENT_RESPONDED);
|
|
assert_eq!(frames[1].r#type, EVENT_ERROR);
|
|
assert_eq!(frames[1].id, "sub-err-1");
|
|
assert_eq!(
|
|
frames[1].payload.get("code"),
|
|
Some(&Value::String("INTERNAL".into()))
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn handle_stream_query_dispatch_unchanged_one_frame_no_completed() {
|
|
let registry = Arc::new(registry_with(
|
|
"echo/run",
|
|
Visibility::External,
|
|
AccessControl::default(),
|
|
));
|
|
let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
|
|
let dp = Dispatcher::new(registry, provider);
|
|
let conn = Arc::new(CallConnection::new(stub_connection()));
|
|
|
|
let request = EventEnvelope::requested(
|
|
"q-pump-1",
|
|
serde_json::json!({
|
|
"operationId": "/echo/run",
|
|
"input": { "msg": "hi" },
|
|
}),
|
|
);
|
|
let recv = tokio::io::BufReader::new(std::io::Cursor::new(encode_frame(&request)));
|
|
let (send, mut sink) = tokio::io::duplex(8 * 1024);
|
|
let stream = crate::core::types::BiStream::from_joined(recv, send);
|
|
|
|
dp.handle_stream(conn, stream).await;
|
|
|
|
let frames = read_all_frames(&mut sink).await;
|
|
assert_eq!(frames.len(), 1, "query: exactly one frame, no completed");
|
|
assert_eq!(frames[0].r#type, EVENT_RESPONDED);
|
|
assert_eq!(frames[0].id, "q-pump-1");
|
|
assert_eq!(
|
|
frames[0].payload.get("output"),
|
|
Some(&serde_json::json!({ "msg": "hi" }))
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn handle_stream_subscription_unknown_op_yields_single_error_no_completed() {
|
|
let registry = Arc::new(OperationRegistry::new());
|
|
let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
|
|
let dp = Dispatcher::new(registry, provider);
|
|
let conn = Arc::new(CallConnection::new(stub_connection()));
|
|
|
|
let request = EventEnvelope::requested(
|
|
"sub-missing-1",
|
|
serde_json::json!({
|
|
"operationId": "/no/such/stream",
|
|
"input": {},
|
|
}),
|
|
);
|
|
let recv = tokio::io::BufReader::new(std::io::Cursor::new(encode_frame(&request)));
|
|
let (send, mut sink) = tokio::io::duplex(8 * 1024);
|
|
let stream = crate::core::types::BiStream::from_joined(recv, send);
|
|
|
|
dp.handle_stream(conn, stream).await;
|
|
|
|
let frames = read_all_frames(&mut sink).await;
|
|
assert_eq!(frames.len(), 1, "unknown op: single error, no completed");
|
|
assert_eq!(frames[0].r#type, EVENT_ERROR);
|
|
assert_eq!(frames[0].id, "sub-missing-1");
|
|
assert_eq!(
|
|
frames[0].payload.get("code"),
|
|
Some(&Value::String("NOT_FOUND".into()))
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn handle_stream_aborted_for_streaming_request_drops_stream() {
|
|
use std::sync::atomic::{AtomicBool, Ordering};
|
|
use std::sync::Arc as StdArc;
|
|
|
|
let dropped = StdArc::new(AtomicBool::new(false));
|
|
let dropped_clone = StdArc::clone(&dropped);
|
|
let handler = make_streaming_handler(move |_input, ctx| {
|
|
let rid = ctx.request_id.clone();
|
|
let flag = StdArc::clone(&dropped_clone);
|
|
struct DropGuard(StdArc<AtomicBool>);
|
|
impl Drop for DropGuard {
|
|
fn drop(&mut self) {
|
|
self.0.store(true, Ordering::SeqCst);
|
|
}
|
|
}
|
|
let guard = DropGuard(StdArc::clone(&flag));
|
|
futures::stream::poll_fn(move |_cx| {
|
|
if flag.load(Ordering::SeqCst) {
|
|
return std::task::Poll::Ready(None);
|
|
}
|
|
std::task::Poll::Ready(Some(ResponseEnvelope::ok(
|
|
rid.clone(),
|
|
serde_json::json!({"tick": 1}),
|
|
)))
|
|
})
|
|
.map(move |env| {
|
|
let _keep_guard = &guard;
|
|
env
|
|
})
|
|
});
|
|
let registry = registry_with_subscription("events/stream", handler);
|
|
let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
|
|
let dp = Dispatcher::new(registry, provider);
|
|
let conn = Arc::new(CallConnection::new(stub_connection()));
|
|
|
|
let request = EventEnvelope::requested(
|
|
"sub-abort-1",
|
|
serde_json::json!({
|
|
"operationId": "/events/stream",
|
|
"input": {},
|
|
}),
|
|
);
|
|
let recv = tokio::io::BufReader::new(std::io::Cursor::new(encode_frame(&request)));
|
|
let (send, _sink) = tokio::io::duplex(8 * 1024);
|
|
let stream = crate::core::types::BiStream::from_joined(recv, send);
|
|
|
|
let conn_clone = Arc::clone(&conn);
|
|
let dp_clone = dp.clone();
|
|
let handle = tokio::spawn(async move {
|
|
dp_clone.handle_stream(conn_clone, stream).await;
|
|
});
|
|
|
|
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
|
|
dp.handle_abort(&conn, "sub-abort-1").await;
|
|
assert!(
|
|
!conn.pending().lock().contains("sub-abort-1"),
|
|
"abort removes the pending entry"
|
|
);
|
|
|
|
handle.abort();
|
|
let _ = handle.await;
|
|
assert!(
|
|
dropped.load(Ordering::SeqCst),
|
|
"stream future dropped → Drop guard released handler resources"
|
|
);
|
|
}
|
|
|
|
// --- ADR-046: Pub/sink dispatch branch --------------------------------
|
|
|
|
fn pub_spec(name: &str, acl: AccessControl) -> OperationSpec {
|
|
OperationSpec::new(
|
|
name,
|
|
OperationType::Pub,
|
|
Visibility::External,
|
|
serde_json::json!({}),
|
|
serde_json::json!({}),
|
|
vec![],
|
|
acl,
|
|
None,
|
|
)
|
|
}
|
|
|
|
fn registry_with_pub(
|
|
name: &str,
|
|
handler: crate::registry::registration::SinkHandler,
|
|
) -> Arc<OperationRegistry> {
|
|
let registry = OperationRegistry::new();
|
|
registry
|
|
.register(HandlerRegistration::new(
|
|
pub_spec(name, AccessControl::default()),
|
|
HandlerKind::Sink(handler),
|
|
OperationProvenance::Local,
|
|
None,
|
|
None,
|
|
Capabilities::new(),
|
|
))
|
|
.unwrap();
|
|
Arc::new(registry)
|
|
}
|
|
|
|
fn counting_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 count = 0u32;
|
|
while let Some(item) = stream.next().await {
|
|
match item {
|
|
Ok(_) => count += 1,
|
|
Err(_) => break,
|
|
}
|
|
}
|
|
ResponseEnvelope::ok(ctx.request_id, serde_json::json!({ "count": count }))
|
|
})
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn dispatch_pub_returns_sink_result() {
|
|
let registry = registry_with_pub("fs/upload", counting_sink_handler());
|
|
let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
|
|
let dp = Dispatcher::new(registry, provider);
|
|
let conn = Arc::new(CallConnection::new(stub_connection()));
|
|
|
|
let payload = serde_json::json!({
|
|
"operationId": "/fs/upload",
|
|
"input": { "path": "/x" },
|
|
});
|
|
match dp.dispatch(&conn, "pub-1".to_string(), payload).await {
|
|
DispatchResult::Sink(sink) => {
|
|
use futures::SinkExt;
|
|
let mut tx = sink.chunk_tx;
|
|
tx.send(Ok(serde_json::json!({"chunk": 1}))).await.unwrap();
|
|
tx.send(Ok(serde_json::json!({"chunk": 2}))).await.unwrap();
|
|
drop(tx);
|
|
let response = sink.handler.await;
|
|
let out = response.result.expect("ok");
|
|
assert_eq!(out["count"], serde_json::json!(2));
|
|
}
|
|
other => panic!("expected Sink, got {other:?}"),
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn dispatch_pub_clears_deadline_to_none() {
|
|
let registry = registry_with_pub("fs/upload", counting_sink_handler());
|
|
let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
|
|
let dp = Dispatcher::new(registry, provider);
|
|
let conn = Arc::new(CallConnection::new(stub_connection()));
|
|
|
|
let payload = serde_json::json!({
|
|
"operationId": "/fs/upload",
|
|
"input": {},
|
|
});
|
|
match dp.dispatch(&conn, "pub-dl".to_string(), payload).await {
|
|
DispatchResult::Sink(sink) => {
|
|
drop(sink.chunk_tx);
|
|
let _ = sink.handler.await;
|
|
}
|
|
other => panic!("expected Sink, got {other:?}"),
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn handle_stream_pub_feeds_published_chunks_and_returns_response() {
|
|
let registry = registry_with_pub("fs/upload", counting_sink_handler());
|
|
let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
|
|
let dp = Dispatcher::new(registry, provider);
|
|
let conn = Arc::new(CallConnection::new(stub_connection()));
|
|
|
|
let request = EventEnvelope::requested(
|
|
"pub-pump-1",
|
|
serde_json::json!({
|
|
"operationId": "/fs/upload",
|
|
"input": { "path": "/x" },
|
|
}),
|
|
);
|
|
let published1 = EventEnvelope::published("pub-pump-1", serde_json::json!({"chunk": 1}));
|
|
let published2 = EventEnvelope::published("pub-pump-1", serde_json::json!({"chunk": 2}));
|
|
let completed = EventEnvelope::completed("pub-pump-1");
|
|
|
|
let mut frame_buf = Vec::new();
|
|
frame_buf.extend_from_slice(&encode_frame(&request));
|
|
frame_buf.extend_from_slice(&encode_frame(&published1));
|
|
frame_buf.extend_from_slice(&encode_frame(&published2));
|
|
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, "pub: one response frame, no completed");
|
|
assert_eq!(frames[0].r#type, EVENT_RESPONDED);
|
|
assert_eq!(frames[0].id, "pub-pump-1");
|
|
let output = frames[0].payload.get("output").expect("output field");
|
|
assert_eq!(output["count"], serde_json::json!(2));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn handle_stream_pub_unknown_op_returns_error() {
|
|
let registry = Arc::new(OperationRegistry::new());
|
|
let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
|
|
let dp = Dispatcher::new(registry, provider);
|
|
let conn = Arc::new(CallConnection::new(stub_connection()));
|
|
|
|
let request = EventEnvelope::requested(
|
|
"pub-missing-1",
|
|
serde_json::json!({
|
|
"operationId": "/no/such/pub",
|
|
"input": {},
|
|
}),
|
|
);
|
|
let recv = tokio::io::BufReader::new(std::io::Cursor::new(encode_frame(&request)));
|
|
let (send, mut sink) = tokio::io::duplex(8 * 1024);
|
|
let stream = crate::core::types::BiStream::from_joined(recv, send);
|
|
|
|
dp.handle_stream(conn, stream).await;
|
|
|
|
let frames = read_all_frames(&mut sink).await;
|
|
assert_eq!(frames.len(), 1, "unknown pub op: single error");
|
|
assert_eq!(frames[0].r#type, EVENT_ERROR);
|
|
assert_eq!(frames[0].id, "pub-missing-1");
|
|
assert_eq!(
|
|
frames[0].payload.get("code"),
|
|
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<OperationRegistry> {
|
|
let 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<CallError> }`.
|
|
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<Value> = Vec::new();
|
|
let mut err: Option<CallError> = 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<dyn IdentityProvider> = 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<dyn IdentityProvider> = 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<dyn IdentityProvider> = 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<dyn IdentityProvider> = 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<dyn IdentityProvider> = 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<StdMutex<Vec<Value>>> = 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<dyn IdentityProvider> = 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"
|
|
);
|
|
}
|
|
|
|
// --- Unit 9: abort-cancels-Pub (P-06) ----------------------------------
|
|
|
|
#[tokio::test]
|
|
async fn pump_sink_same_stream_abort_drops_handler_future_no_response_written() {
|
|
use std::sync::atomic::{AtomicBool, Ordering};
|
|
use std::sync::Arc as StdArc;
|
|
|
|
let dropped = StdArc::new(AtomicBool::new(false));
|
|
let dropped_clone = StdArc::clone(&dropped);
|
|
let 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 flag = StdArc::clone(&dropped_clone);
|
|
struct DropGuard(StdArc<AtomicBool>);
|
|
impl Drop for DropGuard {
|
|
fn drop(&mut self) {
|
|
self.0.store(true, Ordering::SeqCst);
|
|
}
|
|
}
|
|
async move {
|
|
let _guard = DropGuard(StdArc::clone(&flag));
|
|
while let Some(item) = stream.next().await {
|
|
if item.is_err() {
|
|
break;
|
|
}
|
|
}
|
|
ResponseEnvelope::ok(ctx.request_id, serde_json::json!({"done": true}))
|
|
}
|
|
})
|
|
};
|
|
let registry = registry_with_pub("fs/upload", handler);
|
|
let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
|
|
let dp = Dispatcher::new(registry, provider);
|
|
let conn = Arc::new(CallConnection::new(stub_connection()));
|
|
|
|
let request = EventEnvelope::requested(
|
|
"pub-abort-ss-1",
|
|
serde_json::json!({
|
|
"operationId": "/fs/upload",
|
|
"input": {},
|
|
}),
|
|
);
|
|
let c1 = EventEnvelope::published("pub-abort-ss-1", serde_json::json!({"chunk": 1}));
|
|
let abort = EventEnvelope::aborted("pub-abort-ss-1");
|
|
|
|
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(&abort));
|
|
|
|
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);
|
|
|
|
dp.handle_stream(conn, stream).await;
|
|
|
|
let frames = read_all_frames(&mut sink).await;
|
|
assert!(
|
|
frames.is_empty(),
|
|
"no response written after same-stream abort (handler was dropped, not awaited)"
|
|
);
|
|
assert!(
|
|
dropped.load(Ordering::SeqCst),
|
|
"handler future was dropped (DropGuard fired)"
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn pump_sink_cross_stream_abort_drops_handler_future_no_response_written() {
|
|
use std::sync::atomic::{AtomicBool, Ordering};
|
|
use std::sync::Arc as StdArc;
|
|
|
|
let dropped = StdArc::new(AtomicBool::new(false));
|
|
let dropped_clone = StdArc::clone(&dropped);
|
|
let 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 flag = StdArc::clone(&dropped_clone);
|
|
struct DropGuard(StdArc<AtomicBool>);
|
|
impl Drop for DropGuard {
|
|
fn drop(&mut self) {
|
|
self.0.store(true, Ordering::SeqCst);
|
|
}
|
|
}
|
|
async move {
|
|
let _guard = DropGuard(StdArc::clone(&flag));
|
|
while let Some(item) = stream.next().await {
|
|
if item.is_err() {
|
|
break;
|
|
}
|
|
}
|
|
ResponseEnvelope::ok(ctx.request_id, serde_json::json!({"done": true}))
|
|
}
|
|
})
|
|
};
|
|
let registry = registry_with_pub("fs/upload", handler);
|
|
let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
|
|
let dp = Dispatcher::new(registry, provider);
|
|
let conn = Arc::new(CallConnection::new(stub_connection()));
|
|
|
|
let request = EventEnvelope::requested(
|
|
"pub-abort-cs-1",
|
|
serde_json::json!({
|
|
"operationId": "/fs/upload",
|
|
"input": {},
|
|
}),
|
|
);
|
|
let c1 = EventEnvelope::published("pub-abort-cs-1", serde_json::json!({"chunk": 1}));
|
|
|
|
let (mut read_half, write_half) = tokio::io::duplex(8 * 1024);
|
|
{
|
|
use tokio::io::AsyncWriteExt;
|
|
read_half.write_all(&encode_frame(&request)).await.unwrap();
|
|
read_half.write_all(&encode_frame(&c1)).await.unwrap();
|
|
}
|
|
|
|
let recv = tokio::io::BufReader::new(write_half);
|
|
let (send, mut sink) = tokio::io::duplex(8 * 1024);
|
|
let stream = crate::core::types::BiStream::from_joined(recv, send);
|
|
|
|
let conn_clone = Arc::clone(&conn);
|
|
let dp_clone = dp.clone();
|
|
let handle = tokio::spawn(async move {
|
|
dp_clone.handle_stream(conn_clone, stream).await;
|
|
});
|
|
|
|
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
|
|
dp.handle_abort(&conn, "pub-abort-cs-1").await;
|
|
drop(read_half);
|
|
|
|
let frames = read_all_frames(&mut sink).await;
|
|
handle.await.unwrap();
|
|
|
|
assert!(
|
|
frames.is_empty(),
|
|
"no response written after cross-stream abort"
|
|
);
|
|
assert!(
|
|
dropped.load(Ordering::SeqCst),
|
|
"handler future was dropped (DropGuard fired) via cross-stream abort signal"
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn pump_sink_abort_cleans_up_in_flight_sink_aborts_map() {
|
|
let registry = registry_with_pub("fs/upload", counting_sink_handler());
|
|
let provider: Arc<dyn IdentityProvider> = Arc::new(StaticIdentityProvider::new());
|
|
let dp = Dispatcher::new(registry, provider);
|
|
let conn = Arc::new(CallConnection::new(stub_connection()));
|
|
|
|
let request = EventEnvelope::requested(
|
|
"pub-abort-cleanup-1",
|
|
serde_json::json!({
|
|
"operationId": "/fs/upload",
|
|
"input": {},
|
|
}),
|
|
);
|
|
let abort = EventEnvelope::aborted("pub-abort-cleanup-1");
|
|
|
|
let mut frame_buf = Vec::new();
|
|
frame_buf.extend_from_slice(&encode_frame(&request));
|
|
frame_buf.extend_from_slice(&encode_frame(&abort));
|
|
|
|
let recv = tokio::io::BufReader::new(std::io::Cursor::new(frame_buf));
|
|
let (send, _sink) = tokio::io::duplex(8 * 1024);
|
|
let stream = crate::core::types::BiStream::from_joined(recv, send);
|
|
|
|
dp.handle_stream(conn, stream).await;
|
|
|
|
assert!(
|
|
dp.in_flight_sink_aborts.lock().is_empty(),
|
|
"in_flight_sink_aborts map is cleaned up after pump_sink exits"
|
|
);
|
|
}
|
|
|
|
// --- review 004 F-04: the full-duplex serving loop ---------------------
|
|
|
|
/// The F-04 probe: two `serve_single_stream` loops over one duplex
|
|
/// — the exact hub↔consumer shape. Each side holds a
|
|
/// `CallConnection` over its own writer; when side A calls
|
|
/// `echo/run`, its `call.requested` crosses the duplex into side
|
|
/// B's serving loop, which dispatches and writes `call.responded`
|
|
/// back, resolving A's pending through the `EVENT_RESPONDED` arm.
|
|
/// This is the direction `read_single_stream_until_closed` used to
|
|
/// drop.
|
|
#[tokio::test]
|
|
async fn serve_single_stream_self_call_resolves_pending() {
|
|
fn echo_registry() -> Arc<crate::registry::registration::OperationRegistry> {
|
|
let registry = crate::registry::registration::OperationRegistry::new();
|
|
registry
|
|
.register(HandlerRegistration::new(
|
|
external_spec("echo/run", AccessControl::default()),
|
|
HandlerKind::Once(make_handler(|input, context| async move {
|
|
ResponseEnvelope::ok(context.request_id, input)
|
|
})),
|
|
OperationProvenance::Local,
|
|
None,
|
|
None,
|
|
Capabilities::new(),
|
|
))
|
|
.unwrap();
|
|
Arc::new(registry)
|
|
}
|
|
|
|
let (client_end, server_end) = tokio::io::duplex(64 * 1024);
|
|
|
|
// Side B: the duplex end wrapped as one BiStream (read+write);
|
|
// `split_single_stream` divides it into the frame writer and
|
|
// reader the serving loop consumes.
|
|
let server_bidi = crate::core::types::BiStream::from_bidi(server_end);
|
|
let (server_writer, server_reader) =
|
|
crate::protocol::connection::split_single_stream(server_bidi);
|
|
let dp_server = Dispatcher::new(echo_registry(), Arc::new(StaticIdentityProvider::new()));
|
|
let server_call_conn = Arc::new(CallConnection::new_single_stream(
|
|
crate::protocol::sink_empty_connection(),
|
|
Arc::clone(&server_writer),
|
|
));
|
|
let server_conn_for_loop = Arc::clone(&server_call_conn);
|
|
let _serve_b = tokio::spawn(async move {
|
|
dp_server
|
|
.serve_single_stream(server_conn_for_loop, server_reader, server_writer)
|
|
.await;
|
|
});
|
|
|
|
// Side A: the same shape on the other duplex end.
|
|
let client_bidi = crate::core::types::BiStream::from_bidi(client_end);
|
|
let (client_writer, client_reader) =
|
|
crate::protocol::connection::split_single_stream(client_bidi);
|
|
let dp_client = Dispatcher::new(echo_registry(), Arc::new(StaticIdentityProvider::new()));
|
|
let client_call_conn = Arc::new(CallConnection::new_single_stream(
|
|
crate::protocol::sink_empty_connection(),
|
|
Arc::clone(&client_writer),
|
|
));
|
|
let client_conn_for_loop = Arc::clone(&client_call_conn);
|
|
let _serve_a = tokio::spawn(async move {
|
|
dp_client
|
|
.serve_single_stream(client_conn_for_loop, client_reader, client_writer)
|
|
.await;
|
|
});
|
|
|
|
// Side A calls `echo/run` — side B serves it.
|
|
let response = tokio::time::timeout(
|
|
std::time::Duration::from_secs(5),
|
|
client_call_conn.call("echo/run", serde_json::json!({ "from": "a" })),
|
|
)
|
|
.await
|
|
.expect("A→B call through serving loops timed out");
|
|
|
|
assert!(
|
|
response.result.is_ok(),
|
|
"A→B call resolves, got {:?}",
|
|
response.result
|
|
);
|
|
assert_eq!(response.result.unwrap(), serde_json::json!({ "from": "a" }));
|
|
|
|
// And B calls back — A serves it (the both-directions gate).
|
|
let response = tokio::time::timeout(
|
|
std::time::Duration::from_secs(5),
|
|
server_call_conn.call("echo/run", serde_json::json!({ "from": "b" })),
|
|
)
|
|
.await
|
|
.expect("B→A call through serving loops timed out");
|
|
|
|
assert!(
|
|
response.result.is_ok(),
|
|
"B→A call resolves, got {:?}",
|
|
response.result
|
|
);
|
|
assert_eq!(response.result.unwrap(), serde_json::json!({ "from": "b" }));
|
|
}
|
|
}
|