fix(review 005 Unit 1): concurrent serving loops + stub-exercising gates (G-01, G-02)
- Split dispatch() into dispatch_start() (sync prefix) + spawned invocation: both single-stream loops (serve_single_stream and the accept-side run_loop_single_stream) spawn Once invocations, Sub pumps, and sink response writers; only the Pub sink start stays inline (chunk_tx must register before the next call.published). Inline dispatch deadlocked same-connection nested composition: the read loop awaited the parent handler, which awaited a nested call whose response only the same read loop could resolve (resolved only via the 30s sweeper). Spawned handles tracked + aborted at loop exit; in_flight_sinks behind an Arc<parking_lot::Mutex> with guards dropped before awaits. - run_loop_single_stream gains the pending-resolution arms (RESPONDED/COMPLETED/ERROR): the accept side previously served only and had no loop resolving its own outbound pendings in single-stream mode — the latent accept-side imported-op composition hazard is mechanized shut. - Write-failure in the spawned Once path warns instead of closing the loop (matches the Sink arm; dying transport still surfaces via ConnectionClosed on the next read). - G-02 gate: hub_handler_composes_peer_announced_op_via_nested_composition — announce -> consumer calls hub/compose -> hub's serving loop wire-dispatches it -> handler composes via ctx.env -> forwarding stub's nested call crosses back to the consumer. The F-05 gate bypassed this path entirely. - Interleaved-directions gate: outbound_call_resolves_while_inbound_ subscription_is_being_served — consumer serves a live Sub while a wire-dispatched hub handler issues an outbound call on the same connection. - Both gates verified load-bearing: run against the pre-fix loop each reproduces the G-01 hang (no progress, bounded-timeout failure); post-fix both resolve in <0.2s, no sweeper evictions. Verification: cargo test 583 / --all-features 600, clippy (all-targets, all-features, wasm32) clean, fmt clean, doc clean. Refs docs/reviews/005-...md (G-01, G-02; Units 2-3 open).
This commit is contained in:
@@ -2,7 +2,8 @@
|
||||
|
||||
## Status
|
||||
|
||||
Verified, open for remediation.
|
||||
Unit 1 (G-01, G-02) remediated and verified; Units 2–3 open for
|
||||
remediation. See Remediation log.
|
||||
|
||||
## Scope
|
||||
|
||||
@@ -88,6 +89,9 @@ Same scale as review 004:
|
||||
|
||||
## G-01 [major] — Same-connection nested composition deadlocks the serving loop; the forwarding stub's nested call resolves only via the 30s sweeper
|
||||
|
||||
**Status: REMEDIATED (Unit 1)** — see Remediation log; both gates
|
||||
added and verified load-bearing against the pre-fix loop.
|
||||
|
||||
**ADR drift:** ADR-022 amendment (2026-09-03) §`op/register`: the
|
||||
announced op is *"invocable via nested composition (`env.invoke`)"*;
|
||||
the amendment's whole point is that the hub's handlers compose
|
||||
@@ -184,6 +188,9 @@ a Sub is being served) passes without sweeper intervention.
|
||||
|
||||
## G-02 [major] — The F-05 acceptance gate bypasses the forwarding stub it claims to prove
|
||||
|
||||
**Status: REMEDIATED (Unit 1)** — the stub-exercising gate and the
|
||||
interleaved-directions gate are added; see Remediation log.
|
||||
|
||||
**Verified:** YES. `op_register_announce_then_hub_call_routes_back_to_consumer`
|
||||
(`src/channels/client.rs:1412-1543`) asserts the announce resolves and
|
||||
the overlay holds the op — then calls
|
||||
@@ -368,7 +375,8 @@ either way.
|
||||
Sequenced by dependency. All units are alkcall work; Unit 4 (alkhttp
|
||||
wiring) stays downstream and should **not** start before Unit 1 —
|
||||
alkhttp's serving consumers would compose over the same connection
|
||||
and hit G-01 immediately.
|
||||
and hit G-01 immediately. (Unit 1 landed — see Remediation log;
|
||||
Units 2–3 remain.)
|
||||
|
||||
## Unit 1 — Concurrent serving loop + a stub-exercising gate (G-01, G-02)
|
||||
|
||||
@@ -407,6 +415,140 @@ and hit G-01 immediately.
|
||||
|
||||
---
|
||||
|
||||
# Remediation log
|
||||
|
||||
## Unit 1 — Concurrent serving loop + stub-exercising gates (G-01, G-02) — LANDED
|
||||
|
||||
**Fix shape.** `dispatch()` was split into a synchronous start half and
|
||||
an awaited invocation:
|
||||
|
||||
- `Dispatcher::dispatch_start`
|
||||
(`src/protocol/dispatch.rs`) runs the sync prefix (identity
|
||||
resolution, root context, op-type branch) and returns a
|
||||
`StartedDispatch`: `Once` (the invocation as a boxed future), `Stream`
|
||||
(the `ResponseStream`, returned synchronously — `invoke_streaming`
|
||||
is a sync call), or `Sink` (started **inline**, unchanged). `dispatch()`
|
||||
is now a thin wrapper (`Once` = await the boxed future) and keeps its
|
||||
shape for `dispatch_requested`/`handle_stream`/the gateway.
|
||||
- The Sink start stays inline **by design**: its `chunk_tx` must be in
|
||||
`in_flight_sinks` before the next `call.published` frame can be
|
||||
routed; the loop insert cannot race the feed. Pub handlers that
|
||||
block forever inside their sink future are a separate (pre-existing,
|
||||
unreported) shape — the read loop no longer waits on any handler
|
||||
except for the few instructions of the sink start itself.
|
||||
- Both single-stream loops' `EVENT_REQUESTED` arms spawn the Once
|
||||
invocation (`spawn_once_dispatch`), the Sub pump
|
||||
(`spawn_stream_pump`), and the sink's response writer
|
||||
(`spawn_sink_response_writer`); handles are tracked in a
|
||||
`spawned` list and aborted at loop exit (teardown: sink-map clear →
|
||||
spawn aborts → `fail_all` → sweeper abort). `in_flight_sinks` moved
|
||||
behind `Arc<parking_lot::Mutex>` because the ABORTED/PUBLISHED/ERROR
|
||||
arms must not hold the guard across the `chunk_tx.send().await`
|
||||
(non-`Send` guard across an await — the reason the pre-fix loop
|
||||
couldn't just be `tokio::spawn`ed piecemeal). Guard-dropping
|
||||
(`let entry = ...remove()` before the await) keeps lock discipline:
|
||||
no lock is held across an await anywhere in the loops.
|
||||
- **`run_loop_single_stream` (the accept side) got the
|
||||
pending-resolution arms** (`EVENT_RESPONDED`/`COMPLETED`/`ERROR` →
|
||||
the outbound pending map). Previously it served only — an accept-side
|
||||
nested-composing handler (e.g. a `from_call` imported-op stub riding
|
||||
the same connection) had no loop resolving its response frames at
|
||||
all. The G-01 accept-side latent hazard is mechanized shut, not just
|
||||
unblocked: both single-stream loops are now the same shape
|
||||
(dispatch-spawn + pending-resolution), the full-duplex loop
|
||||
`serve_single_stream` composes them and is unchanged in its arm
|
||||
semantics (frame-arm equivalence preserved per the non-findings
|
||||
audit).
|
||||
- Write-failure semantics changed from `break` (close the loop) to
|
||||
`warn` (keep reading) in the spawned Once path, matching the Sink
|
||||
arm: a dying transport surfaces on the next read as
|
||||
`ConnectionClosed`; a transient frame-write failure no longer tears
|
||||
down every in-flight request on the connection.
|
||||
|
||||
**Gates (G-02, productized from the removed probe):**
|
||||
|
||||
- `hub_handler_composes_peer_announced_op_via_nested_composition`
|
||||
(`src/channels/client.rs`): announce → consumer calls `hub/compose`
|
||||
→ the hub's serving loop wire-dispatches it → the handler resolves
|
||||
`consumer/exec` via `ctx.env` nested composition → the forwarding
|
||||
stub's nested `call.requested` crosses back to the consumer. This is
|
||||
the stub path the F-05 gate bypassed. Two wiring facts surfaced (both
|
||||
recorded as spec-consistent, both were silent before): (a) the hub's
|
||||
channel-0 `Connection` must carry the peer identity —
|
||||
`compose_root_env` attaches the connection overlay keyed by
|
||||
`identity.id` (ADR-030 §5), and a deployment that skips identity
|
||||
resolution silently gets no peer overlay (the test sets it via the
|
||||
`AuthContext` the adapter passes to `install_channel_zero`);
|
||||
(b) composition reachability is declared on the composing handler's
|
||||
registration (`scoped_env: ScopedPeerEnv::new(["consumer/exec"])`) —
|
||||
the empty `ScopedPeerEnv` is deny-by-default in
|
||||
`PeerCompositeEnv::invoke_with_policy`, so a wire-dispatched handler
|
||||
with no `scoped_env` composes nothing (the reachability gate, not the
|
||||
overlay, was what the probe's first draft tripped over).
|
||||
- `outbound_call_resolves_while_inbound_subscription_is_being_served`
|
||||
(the optional interleaved-directions phase on the F-04 shape): the
|
||||
consumer subscribes to the hub (Sub served by the consumer's loop),
|
||||
then calls `hub/interleave` over the wire — its handler issues an
|
||||
outbound hub→consumer call on the same connection while the Sub is
|
||||
live — and asserts the call resolves and the Sub is still live
|
||||
after. Both gates bounded at 5s (vs. the 30s sweeper), so a
|
||||
regression fails fast, not via sweeper eviction.
|
||||
|
||||
**Load-bearing verification (both gates, empirical):** each gate was
|
||||
run against the pre-fix loop (`git stash push
|
||||
src/protocol/dispatch.rs`) and reproduced the G-01 hang:
|
||||
|
||||
```
|
||||
hub_handler_composes…: panicked "nested composition through the
|
||||
forwarding stub timed out (G-01 shape): Elapsed(())" — 5.01s, no progress
|
||||
outbound_call_resolves…: panicked "interleaved outbound call starved
|
||||
while Sub served (G-01 shape): Elapsed(())" — 5.06s, no progress
|
||||
```
|
||||
|
||||
With the fix, both resolve (0.01s / 0.11s wall including connection
|
||||
setup). No sweeper evictions (bounded timeouts ≪ 30s).
|
||||
|
||||
**Consequences audited against the fix:**
|
||||
|
||||
- Same-connection nested composition of peer-announced ops works for
|
||||
wire-dispatched parents (the flagship ADR-022 amendment flow) — the
|
||||
gate proves it end-to-end.
|
||||
- The accept-side imported-op composition hazard (G-01's second
|
||||
consequence, latent pre-`f84d214`) is mechanized shut by the
|
||||
pending-resolution arms in `run_loop_single_stream` — not merely
|
||||
unblocked. There is no dedicated gate for this shape yet (the
|
||||
accept-side import + nested-compose e2e would be a further gate;
|
||||
noted as residual work, not a regression).
|
||||
- Head-of-line blocking is gone: spawned arms proceed concurrently;
|
||||
the loop never awaits a handler.
|
||||
- Frame ordering: per-request ordering is preserved by the
|
||||
`SharedFrameWriter` (each `write_frame` is atomic under the mutex);
|
||||
responses for two requests may now interleave *at frame
|
||||
granularity*, which single-stream mode always permitted (both
|
||||
directions' frames are multiplexed by design; correlation is by id).
|
||||
|
||||
**Verification (post-fix):**
|
||||
|
||||
```
|
||||
cargo test → 583 passed, 0 failed
|
||||
cargo test --all-features → 600 passed, 0 failed
|
||||
cargo clippy --all-targets -- -D warnings → clean
|
||||
cargo clippy --all-features --all-targets -- -D warnings → clean
|
||||
cargo fmt --check → clean (fmt applied)
|
||||
cargo clippy --target wasm32-unknown-unknown -- -D warnings → clean
|
||||
cargo doc --no-deps → clean
|
||||
```
|
||||
|
||||
**Residual (not blocking Unit 1):** the sink-start-inline shape means a
|
||||
`Pub` op whose registration itself blocks (ACL/schema compile are sync
|
||||
and fast; `resolve_sink_handler` runs the handler's ACL path) still
|
||||
holds the loop briefly — bounded by registry work, not handler work. A
|
||||
malicious `AccessControl::check` implementation could stall the loop;
|
||||
that is a deployment-provided trait object, the same trust boundary as
|
||||
`IdentityProvider`, and unchanged from the pre-existing shape.
|
||||
|
||||
---
|
||||
|
||||
## Verification log (this pass)
|
||||
|
||||
- All gates in Baseline verification reproduced at tree `f84d214`
|
||||
|
||||
+437
-1
@@ -271,7 +271,7 @@ mod tests {
|
||||
use crate::protocol::dispatch::Dispatcher;
|
||||
use crate::registry::context::OperationContext;
|
||||
use crate::registry::registration::{
|
||||
make_handler, HandlerKind, HandlerRegistration, OperationProvenance,
|
||||
make_handler, make_streaming_handler, HandlerKind, HandlerRegistration, OperationProvenance,
|
||||
};
|
||||
use crate::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
|
||||
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
|
||||
@@ -1542,6 +1542,442 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
/// Interleaved-directions gate (review 005 G-01, third consequence):
|
||||
/// the consumer's outbound call resolves **while** its serving loop
|
||||
/// is mid-pump on a long-lived inbound Sub to the same peer. Under
|
||||
/// the pre-G-01 loop the Sub pump held the read loop inline and the
|
||||
/// outbound call's `call.responded` queued unresolved until the
|
||||
/// sweeper; sequential-direction gates (hub→consumer, *then*
|
||||
/// consumer→hub) could never catch it. The consumer also issues the
|
||||
/// outbound call *from inside* a wire-dispatched handler, the
|
||||
/// tightest interleaving.
|
||||
#[tokio::test]
|
||||
async fn outbound_call_resolves_while_inbound_subscription_is_being_served() {
|
||||
// Consumer side: serves `consumer/echo` (Once) and
|
||||
// `consumer/tick` (Sub, unbounded — held open until aborted).
|
||||
let consumer_registry = crate::registry::registration::OperationRegistry::new();
|
||||
consumer_registry
|
||||
.register(HandlerRegistration::new(
|
||||
external_query_spec("consumer/echo"),
|
||||
HandlerKind::Once(make_handler(|input, ctx| async move {
|
||||
ResponseEnvelope::ok(ctx.request_id, input)
|
||||
})),
|
||||
OperationProvenance::Local,
|
||||
None,
|
||||
None,
|
||||
crate::core::types::Capabilities::new(),
|
||||
))
|
||||
.unwrap();
|
||||
consumer_registry
|
||||
.register(HandlerRegistration::new(
|
||||
OperationSpec::new(
|
||||
"consumer/tick",
|
||||
OperationType::Sub,
|
||||
Visibility::External,
|
||||
serde_json::json!({}),
|
||||
serde_json::json!({}),
|
||||
vec![],
|
||||
AccessControl::default(),
|
||||
None,
|
||||
),
|
||||
HandlerKind::Stream(make_streaming_handler(|_input, ctx: OperationContext| {
|
||||
let request_id = ctx.request_id.clone();
|
||||
futures::stream::unfold((0u32, request_id), |(n, request_id)| async move {
|
||||
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
|
||||
Some((
|
||||
ResponseEnvelope::ok(
|
||||
request_id.clone(),
|
||||
serde_json::json!({ "tick": n }),
|
||||
),
|
||||
(n + 1, request_id),
|
||||
))
|
||||
})
|
||||
})),
|
||||
OperationProvenance::Local,
|
||||
None,
|
||||
None,
|
||||
crate::core::types::Capabilities::new(),
|
||||
))
|
||||
.unwrap();
|
||||
let consumer_registry = Arc::new(consumer_registry);
|
||||
|
||||
// Handle channel: the install hook yields the accept side's
|
||||
// channel-0 `CallConnection` so the test driver can subscribe
|
||||
// (the accept side initiates the Sub) and so the
|
||||
// `hub/interleave` handler can reach it.
|
||||
let (accept_conn_tx, mut accept_conn_rx) =
|
||||
tokio::sync::mpsc::channel::<Arc<CallConnection>>(1);
|
||||
|
||||
// Accept side (hub): serves `accept/echo` and
|
||||
// `hub/interleave` — the latter, when wire-dispatched from the
|
||||
// consumer, issues an outbound call on the *same* connection.
|
||||
let accept_registry = crate::registry::registration::OperationRegistry::new();
|
||||
accept_registry
|
||||
.register(HandlerRegistration::new(
|
||||
external_query_spec("accept/echo"),
|
||||
HandlerKind::Once(make_handler(|input, ctx| async move {
|
||||
ResponseEnvelope::ok(ctx.request_id, input)
|
||||
})),
|
||||
OperationProvenance::Local,
|
||||
None,
|
||||
None,
|
||||
crate::core::types::Capabilities::new(),
|
||||
))
|
||||
.unwrap();
|
||||
let accept_conn_for_handler =
|
||||
Arc::new(tokio::sync::Mutex::new(None::<Arc<CallConnection>>));
|
||||
let accept_conn_slot = Arc::clone(&accept_conn_for_handler);
|
||||
accept_registry
|
||||
.register(HandlerRegistration::new(
|
||||
external_query_spec("hub/interleave"),
|
||||
HandlerKind::Once(make_handler(move |input, ctx| {
|
||||
let conn_slot = Arc::clone(&accept_conn_for_handler);
|
||||
async move {
|
||||
let conn = conn_slot
|
||||
.lock()
|
||||
.await
|
||||
.clone()
|
||||
.expect("hub call connection set");
|
||||
let response = conn
|
||||
.call("consumer/echo", serde_json::json!({ "via": "interleave" }))
|
||||
.await;
|
||||
let echo = response.result.expect("outbound call resolves");
|
||||
ResponseEnvelope::ok(
|
||||
ctx.request_id,
|
||||
serde_json::json!({
|
||||
"input": input,
|
||||
"outbound": echo,
|
||||
}),
|
||||
)
|
||||
}
|
||||
})),
|
||||
OperationProvenance::Local,
|
||||
None,
|
||||
None,
|
||||
crate::core::types::Capabilities::new(),
|
||||
))
|
||||
.unwrap();
|
||||
let accept_registry = Arc::new(accept_registry);
|
||||
|
||||
let install_hook: crate::channels::adapter::InstallChannelZero =
|
||||
Arc::new(move |_manager, channel0_conn, _auth| {
|
||||
let accept_registry = Arc::clone(&accept_registry);
|
||||
let accept_conn_tx = accept_conn_tx.clone();
|
||||
let accept_conn_slot = Arc::clone(&accept_conn_slot);
|
||||
tokio::spawn(async move {
|
||||
let channel0_bidi = match channel0_conn.accept_bi().await {
|
||||
Ok(s) => s,
|
||||
Err(_) => return,
|
||||
};
|
||||
let (writer, reader) = split_single_stream(channel0_bidi);
|
||||
let call_connection = Arc::new(CallConnection::new_single_stream(
|
||||
channel0_conn,
|
||||
Arc::clone(&writer),
|
||||
));
|
||||
accept_conn_slot
|
||||
.lock()
|
||||
.await
|
||||
.replace(Arc::clone(&call_connection));
|
||||
let _ = accept_conn_tx.send(Arc::clone(&call_connection)).await;
|
||||
let dp = Dispatcher::new(
|
||||
accept_registry,
|
||||
Arc::new(NoopIdProvider) as Arc<dyn IdentityProvider>,
|
||||
);
|
||||
dp.serve_single_stream(call_connection, reader, writer)
|
||||
.await;
|
||||
})
|
||||
});
|
||||
|
||||
let (client_end, server_end) = tokio::io::duplex(64 * 1024);
|
||||
let client_conn =
|
||||
Connection::from_bidi(client_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
|
||||
let server_conn =
|
||||
Connection::from_bidi(server_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
|
||||
|
||||
let adapter = ChannelsAdapter::new(install_hook, Arc::new(NoCap));
|
||||
let auth = AuthContext::anonymous(b"alk/channels");
|
||||
let _server_handle = tokio::spawn(async move {
|
||||
let _ = crate::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await;
|
||||
});
|
||||
|
||||
let client = ChannelClient::from_connection_with_serving(
|
||||
client_conn,
|
||||
Some(ServingConfig {
|
||||
registry: Arc::clone(&consumer_registry),
|
||||
identity_provider: Arc::new(NoopIdProvider),
|
||||
}),
|
||||
)
|
||||
.await
|
||||
.expect("channel client init");
|
||||
|
||||
let accept_conn = accept_conn_rx
|
||||
.recv()
|
||||
.await
|
||||
.expect("accept side channel-0 connection handle");
|
||||
|
||||
// Phase 1: the consumer subscribes to the hub's `consumer/tick`
|
||||
// — the consumer's serving loop pumps the Sub inline (pre-fix)
|
||||
// or in a spawned task (post-fix). The stream stays open.
|
||||
use futures::stream::StreamExt;
|
||||
let mut ticks = tokio::time::timeout(
|
||||
std::time::Duration::from_secs(5),
|
||||
accept_conn.subscribe("consumer/tick", serde_json::json!({})),
|
||||
)
|
||||
.await
|
||||
.expect("subscribe timed out");
|
||||
let first = tokio::time::timeout(std::time::Duration::from_secs(5), ticks.next())
|
||||
.await
|
||||
.expect("first tick timed out")
|
||||
.expect("first tick item");
|
||||
assert!(
|
||||
first.result.is_ok(),
|
||||
"first tick ok, got {:?}",
|
||||
first.result
|
||||
);
|
||||
|
||||
// Phase 2: with the Sub live, the consumer calls the hub's
|
||||
// `hub/interleave` — the wire-dispatched handler issues an
|
||||
// outbound hub→consumer call on the same connection while the
|
||||
// consumer's serving loop is serving the Sub. Pre-fix, the
|
||||
// consumer's serving loop held the Sub pump inline and this
|
||||
// call starved (30s sweeper); post-fix it resolves.
|
||||
let response = tokio::time::timeout(
|
||||
std::time::Duration::from_secs(5),
|
||||
client.call_open_op(
|
||||
"hub/interleave",
|
||||
serde_json::json!({ "when": "while-sub-live" }),
|
||||
),
|
||||
)
|
||||
.await
|
||||
.expect("interleaved outbound call starved while Sub served (G-01 shape)");
|
||||
assert!(
|
||||
response.result.is_ok(),
|
||||
"interleaved call resolves, got {:?}",
|
||||
response.result
|
||||
);
|
||||
let out = response.result.unwrap();
|
||||
assert_eq!(
|
||||
out.get("outbound"),
|
||||
Some(&serde_json::json!({ "via": "interleave" })),
|
||||
"the nested outbound hub→consumer call resolved through the serving loop"
|
||||
);
|
||||
|
||||
// The Sub is still live after the interleaved call.
|
||||
let second = tokio::time::timeout(std::time::Duration::from_secs(5), ticks.next())
|
||||
.await
|
||||
.expect("second tick timed out")
|
||||
.expect("second tick item");
|
||||
assert!(
|
||||
second.result.is_ok(),
|
||||
"sub still live after interleaved call, got {:?}",
|
||||
second.result
|
||||
);
|
||||
}
|
||||
|
||||
/// G-02 gate (review 005): the forwarding stub is exercised
|
||||
/// through **nested composition from a wire-dispatched hub
|
||||
/// handler** — the exact path the direct wire call in
|
||||
/// `op_register_announce_then_hub_call_routes_back_to_consumer`
|
||||
/// bypasses, and the path G-01 deadlocked (the inline dispatch
|
||||
/// blocked the read loop that had to resolve the stub's nested
|
||||
/// call; it resolved only via the 30s sweeper). The hub registry
|
||||
/// serves `hub/compose`, whose handler invokes the announced op
|
||||
/// through `context.env` — the `PeerCompositeEnv` →
|
||||
/// `OverlayOperationEnv` resolution a real hub handler produces.
|
||||
/// Resolving quickly (bounded well under the sweeper interval)
|
||||
/// proves the serving loop resolves same-connection nested
|
||||
/// composition live.
|
||||
#[tokio::test]
|
||||
async fn hub_handler_composes_peer_announced_op_via_nested_composition() {
|
||||
// Consumer side: serves `consumer/exec` locally; announces it.
|
||||
let consumer_registry = crate::registry::registration::OperationRegistry::new();
|
||||
consumer_registry
|
||||
.register(HandlerRegistration::new(
|
||||
external_query_spec("consumer/exec"),
|
||||
HandlerKind::Once(make_handler(|input, ctx| async move {
|
||||
ResponseEnvelope::ok(
|
||||
ctx.request_id,
|
||||
serde_json::json!({ "ran_on": "consumer", "input": input }),
|
||||
)
|
||||
})),
|
||||
OperationProvenance::Local,
|
||||
None,
|
||||
None,
|
||||
crate::core::types::Capabilities::new(),
|
||||
))
|
||||
.unwrap();
|
||||
let consumer_registry = Arc::new(consumer_registry);
|
||||
|
||||
// Accept side (hub): serves `op/register` + `hub/compose` on a
|
||||
// fork. `hub/compose` composes the announced op via
|
||||
// `context.env` — the nested-composition shape ADR-022's
|
||||
// amendment promises. The stub is not invoked by the assertion
|
||||
// below directly; it is resolved through the connection
|
||||
// overlay `compose_root_env` attaches. The channel-0
|
||||
// connection carries the peer's identity (the assembly layer
|
||||
// resolves it from the accepted connection's `AuthContext`) —
|
||||
// `compose_root_env` attaches the connection overlay keyed by
|
||||
// that identity (ADR-030 §5); an identity-less connection gets
|
||||
// no overlay.
|
||||
let (accept_conn_tx, mut accept_conn_rx) =
|
||||
tokio::sync::mpsc::channel::<Arc<CallConnection>>(1);
|
||||
let accept_registry_for_hook =
|
||||
Arc::new(crate::registry::registration::OperationRegistry::new());
|
||||
let accept_tx_for_hook = accept_conn_tx.clone();
|
||||
let install_hook: crate::channels::adapter::InstallChannelZero =
|
||||
Arc::new(move |_manager, channel0_conn, auth| {
|
||||
if let Some(identity) = auth.identity.clone() {
|
||||
let _ = channel0_conn.set_identity(identity);
|
||||
}
|
||||
let accept_registry = Arc::clone(&accept_registry_for_hook);
|
||||
let accept_conn_tx = accept_tx_for_hook.clone();
|
||||
tokio::spawn(async move {
|
||||
let channel0_bidi = match channel0_conn.accept_bi().await {
|
||||
Ok(s) => s,
|
||||
Err(_) => return,
|
||||
};
|
||||
let (writer, reader) = split_single_stream(channel0_bidi);
|
||||
let call_connection = Arc::new(CallConnection::new_single_stream(
|
||||
channel0_conn,
|
||||
Arc::clone(&writer),
|
||||
));
|
||||
let _ = accept_conn_tx.send(Arc::clone(&call_connection)).await;
|
||||
|
||||
let reg = accept_registry.fork();
|
||||
reg.register(HandlerRegistration::new(
|
||||
crate::registry::op_register::op_register_spec(AccessControl::default()),
|
||||
HandlerKind::Once(crate::registry::op_register::op_register_handler(
|
||||
Arc::clone(&call_connection),
|
||||
)),
|
||||
OperationProvenance::Local,
|
||||
None,
|
||||
None,
|
||||
crate::core::types::Capabilities::new(),
|
||||
))
|
||||
.unwrap();
|
||||
reg.register(HandlerRegistration::new(
|
||||
external_query_spec("hub/compose"),
|
||||
HandlerKind::Once(make_handler(|input, ctx| async move {
|
||||
let name = input
|
||||
.get("op")
|
||||
.and_then(|v| v.as_str())
|
||||
.unwrap_or("")
|
||||
.to_string();
|
||||
let (namespace, operation) = match name.split_once('/') {
|
||||
Some(parts) => parts,
|
||||
None => {
|
||||
return ResponseEnvelope::error(
|
||||
ctx.request_id,
|
||||
crate::protocol::wire::CallError::invalid_input(
|
||||
"hub/compose input missing `op` as `ns/op`",
|
||||
),
|
||||
)
|
||||
}
|
||||
};
|
||||
let response = ctx.env.invoke(namespace, operation, input, &ctx).await;
|
||||
ResponseEnvelope {
|
||||
request_id: ctx.request_id,
|
||||
result: response.result,
|
||||
}
|
||||
})),
|
||||
OperationProvenance::Local,
|
||||
None,
|
||||
Some(crate::registry::context::ScopedPeerEnv::new([
|
||||
"consumer/exec",
|
||||
])),
|
||||
crate::core::types::Capabilities::new(),
|
||||
))
|
||||
.unwrap();
|
||||
let dp = Dispatcher::new(
|
||||
Arc::new(reg),
|
||||
Arc::new(NoopIdProvider) as Arc<dyn IdentityProvider>,
|
||||
);
|
||||
dp.serve_single_stream(call_connection, reader, writer)
|
||||
.await;
|
||||
})
|
||||
});
|
||||
|
||||
let (client_end, server_end) = tokio::io::duplex(64 * 1024);
|
||||
let client_conn =
|
||||
Connection::from_bidi(client_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
|
||||
let server_conn =
|
||||
Connection::from_bidi(server_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
|
||||
|
||||
let adapter = ChannelsAdapter::new(install_hook, Arc::new(NoCap));
|
||||
let auth = AuthContext {
|
||||
identity: Some(crate::core::auth::Identity {
|
||||
id: "consumer-peer".to_string(),
|
||||
scopes: vec![],
|
||||
resources: Default::default(),
|
||||
}),
|
||||
..AuthContext::anonymous(b"alk/channels")
|
||||
};
|
||||
let _server_handle = tokio::spawn(async move {
|
||||
let _ = crate::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await;
|
||||
});
|
||||
|
||||
let client = ChannelClient::from_connection_with_serving(
|
||||
client_conn,
|
||||
Some(ServingConfig {
|
||||
registry: Arc::clone(&consumer_registry),
|
||||
identity_provider: Arc::new(NoopIdProvider),
|
||||
}),
|
||||
)
|
||||
.await
|
||||
.expect("channel client init");
|
||||
|
||||
let _accept_conn = accept_conn_rx
|
||||
.recv()
|
||||
.await
|
||||
.expect("accept side channel-0 connection handle");
|
||||
|
||||
// Consumer announces `consumer/exec` over channel 0.
|
||||
let announce = tokio::time::timeout(
|
||||
std::time::Duration::from_secs(10),
|
||||
client.call_open_op(
|
||||
crate::registry::op_register::OP_REGISTER_NAME,
|
||||
crate::registry::op_register::OpRegisterRequest {
|
||||
spec: external_query_spec("consumer/exec"),
|
||||
replace: false,
|
||||
}
|
||||
.to_json(),
|
||||
),
|
||||
)
|
||||
.await
|
||||
.expect("op/register announce timed out");
|
||||
assert!(
|
||||
announce.result.is_ok(),
|
||||
"announce ok, got {:?}",
|
||||
announce.result
|
||||
);
|
||||
|
||||
// A wire-dispatched hub handler composes the announced op via
|
||||
// nested composition: the consumer calls `hub/compose` over the
|
||||
// wire, the hub's serving loop dispatches it, and the handler
|
||||
// resolves `consumer/exec` through the forwarding stub. Under
|
||||
// G-01 this resolved only at 30s via the sweeper; the bounded
|
||||
// timeout proves live resolution.
|
||||
let response = tokio::time::timeout(
|
||||
std::time::Duration::from_secs(5),
|
||||
client.call_open_op(
|
||||
"hub/compose",
|
||||
serde_json::json!({ "op": "consumer/exec", "task": "nested" }),
|
||||
),
|
||||
)
|
||||
.await
|
||||
.expect("nested composition through the forwarding stub timed out (G-01 shape)");
|
||||
|
||||
assert!(
|
||||
response.result.is_ok(),
|
||||
"nested composition resolves through the forwarding stub, got {:?}",
|
||||
response.result
|
||||
);
|
||||
assert_eq!(
|
||||
response.result.unwrap(),
|
||||
serde_json::json!({ "ran_on": "consumer", "input": { "op": "consumer/exec", "task": "nested" } })
|
||||
);
|
||||
}
|
||||
|
||||
// --- review 004 Unit 2 acceptance gate (F-02/F-06 fork) ----------------
|
||||
|
||||
/// F-02/F-06 gate: the open op is registered on a **fork** of the
|
||||
|
||||
+265
-154
@@ -114,6 +114,18 @@ impl std::fmt::Debug for DispatchResult {
|
||||
}
|
||||
}
|
||||
|
||||
/// The started half of a dispatched `call.requested` event, from
|
||||
/// [`Dispatcher::dispatch_start`]. `Once`/`Stream` are returned ready
|
||||
/// for the caller to spawn (review 005 G-01); `Sink` was started
|
||||
/// inline (its `chunk_tx` must be registered before the next
|
||||
/// `call.published` frame is read) and carries the started handler
|
||||
/// future.
|
||||
pub enum StartedDispatch {
|
||||
Once(Pin<Box<dyn Future<Output = ResponseEnvelope> + Send>>),
|
||||
Stream(ResponseStream),
|
||||
Sink(SinkDispatch),
|
||||
}
|
||||
|
||||
/// 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
|
||||
@@ -299,6 +311,39 @@ impl Dispatcher {
|
||||
request_id: String,
|
||||
payload: Value,
|
||||
) -> DispatchResult {
|
||||
match self.dispatch_start(connection, request_id, payload) {
|
||||
StartedDispatch::Once(invoke) => DispatchResult::Once(invoke.await),
|
||||
StartedDispatch::Stream(stream) => DispatchResult::Stream(stream),
|
||||
StartedDispatch::Sink(sink) => DispatchResult::Sink(sink),
|
||||
}
|
||||
}
|
||||
|
||||
/// Start dispatching a `call.requested` event without awaiting the
|
||||
/// invocation (review 005 G-01). The synchronous prefix of
|
||||
/// [`dispatch`](Self::dispatch) — identity resolution, root-context
|
||||
/// construction, op-type branch — is split from the invocation:
|
||||
///
|
||||
/// - `Query`/`Mutation` return the invocation as a boxed future
|
||||
/// (`StartedDispatch::Once`) for the caller to spawn. Awaiting it
|
||||
/// inline is what deadlocked the serving loops: a handler that
|
||||
/// nested-composes over the same connection blocks the one read
|
||||
/// loop that must resolve the nested call's response frame.
|
||||
/// - `Sub` returns the [`ResponseStream`] synchronously
|
||||
/// (`invoke_streaming` is a sync call) for the caller to pump in
|
||||
/// a spawned task — an inline pump starves the same loop for the
|
||||
/// stream's whole lifetime.
|
||||
/// - `Pub` runs the full sink start inline (`StartedDispatch::Sink`):
|
||||
/// ACL failure resolves to a ready `Once` future (the response
|
||||
/// envelope is already built), success carries the `chunk_tx` the
|
||||
/// loop must register in `in_flight_sinks` *before* the next
|
||||
/// `call.published` frame can be read — the insert cannot race
|
||||
/// the feed, so the sink start stays synchronous by design.
|
||||
pub(crate) fn dispatch_start(
|
||||
&self,
|
||||
connection: &Arc<CallConnection>,
|
||||
request_id: String,
|
||||
payload: Value,
|
||||
) -> StartedDispatch {
|
||||
let operation_id = payload
|
||||
.get("operationId")
|
||||
.and_then(|v| v.as_str())
|
||||
@@ -320,7 +365,7 @@ impl Dispatcher {
|
||||
.unwrap_or(OperationType::Query);
|
||||
|
||||
let mut context = self.build_root_context(
|
||||
request_id.clone(),
|
||||
request_id,
|
||||
&operation_name,
|
||||
identity,
|
||||
forwarded_for,
|
||||
@@ -329,40 +374,125 @@ impl Dispatcher {
|
||||
|
||||
match op_type {
|
||||
OperationType::Query | OperationType::Mutation => {
|
||||
let envelope = self.registry.invoke(&operation_name, input, context).await;
|
||||
DispatchResult::Once(envelope)
|
||||
let registry = Arc::clone(&self.registry);
|
||||
let name = operation_name;
|
||||
StartedDispatch::Once(Box::pin(async move {
|
||||
registry.invoke(&name, input, context).await
|
||||
}))
|
||||
}
|
||||
OperationType::Sub => {
|
||||
context.deadline = None;
|
||||
let stream = self
|
||||
.registry
|
||||
.invoke_streaming(&operation_name, input, context);
|
||||
DispatchResult::Stream(stream)
|
||||
StartedDispatch::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,
|
||||
})
|
||||
match self
|
||||
.registry
|
||||
.resolve_sink_handler(&operation_name, &input, &context)
|
||||
{
|
||||
Ok(sink_handler) => {
|
||||
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);
|
||||
StartedDispatch::Sink(SinkDispatch {
|
||||
handler: Box::pin(handler),
|
||||
chunk_tx,
|
||||
publish_validator,
|
||||
})
|
||||
}
|
||||
Err(envelope) => StartedDispatch::Once(Box::pin(std::future::ready(envelope))),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Spawn the Once invocation to write its single response frame on
|
||||
/// completion. The write failure is a warning, not a loop exit: the
|
||||
/// loop keeps reading (a dying transport surfaces on the next read
|
||||
/// as `ConnectionClosed`), matching the Sink arm's behavior.
|
||||
fn spawn_once_dispatch(
|
||||
writer: &Arc<super::connection::SharedFrameWriter>,
|
||||
request_id: String,
|
||||
invoke: Pin<Box<dyn Future<Output = ResponseEnvelope> + Send>>,
|
||||
) -> JoinHandle<()> {
|
||||
let writer = Arc::clone(writer);
|
||||
tokio::spawn(async move {
|
||||
let response = invoke.await;
|
||||
let event: EventEnvelope = response.into();
|
||||
if let Err(err) = writer.write_frame(&event).await {
|
||||
warn!(
|
||||
error = %err,
|
||||
request_id = %request_id,
|
||||
"serving loop: failed to write Once response frame"
|
||||
);
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
/// Spawn a subscription's [`ResponseStream`] pump: each
|
||||
/// [`ResponseEnvelope`] becomes a `call.responded` / `call.error`
|
||||
/// frame; on natural end, a `call.completed` frame. The shared
|
||||
/// writer serializes frames so the pump's frames do not interleave
|
||||
/// with concurrent calls' frames.
|
||||
fn spawn_stream_pump(
|
||||
writer: &Arc<super::connection::SharedFrameWriter>,
|
||||
request_id: String,
|
||||
mut stream: ResponseStream,
|
||||
) -> JoinHandle<()> {
|
||||
let writer = Arc::clone(writer);
|
||||
tokio::spawn(async move {
|
||||
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,
|
||||
request_id = %request_id,
|
||||
"serving loop: 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,
|
||||
request_id = %request_id,
|
||||
"serving loop: failed to write call.completed"
|
||||
);
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
/// Spawn a Pub's handler task: awaits the handler future and writes
|
||||
/// the single `call.responded` / `call.error` frame.
|
||||
fn spawn_sink_response_writer(
|
||||
writer: &Arc<super::connection::SharedFrameWriter>,
|
||||
request_id: String,
|
||||
handler: Pin<Box<dyn Future<Output = ResponseEnvelope> + Send>>,
|
||||
) -> JoinHandle<()> {
|
||||
let writer = Arc::clone(writer);
|
||||
tokio::spawn(async move {
|
||||
let response = handler.await;
|
||||
let event: EventEnvelope = response.into();
|
||||
if let Err(err) = writer.write_frame(&event).await {
|
||||
warn!(
|
||||
error = %err,
|
||||
request_id = %request_id,
|
||||
"serving loop: failed to write sink response frame"
|
||||
);
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
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(());
|
||||
@@ -716,14 +846,24 @@ impl Dispatcher {
|
||||
/// `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.
|
||||
/// other requests continue to be dispatched. Once invocations and
|
||||
/// Sub pumps are spawned (review 005 G-01 — inline dispatch
|
||||
/// deadlocked same-connection nested composition); only the sink
|
||||
/// start runs inline.
|
||||
///
|
||||
/// This is the accept side's serving loop, so the loop also carries
|
||||
/// the pending-resolution arms the connect side's
|
||||
/// `serve_single_stream` composes: a nested-composing handler
|
||||
/// (e.g. a `from_call` imported-op stub riding this connection)
|
||||
/// blocks its own task, not the read loop, and the loop resolves
|
||||
/// the stub's response frames here. Without these arms the accept
|
||||
/// side could dispatch inbound requests but never resolve its own
|
||||
/// outbound pendings in single-stream mode.
|
||||
///
|
||||
/// 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).
|
||||
/// pending requests are failed with `connection closed`, in-flight
|
||||
/// sinks' `chunk_tx` are dropped (the handler's `PublishStream`
|
||||
/// sees EOF), and spawned Once/Stream/Sink tasks are aborted.
|
||||
pub async fn run_loop_single_stream(
|
||||
self,
|
||||
connection: Arc<CallConnection>,
|
||||
@@ -749,7 +889,10 @@ impl Dispatcher {
|
||||
});
|
||||
|
||||
let mut reader = FrameFramedReader::new(reader);
|
||||
let mut in_flight_sinks: HashMap<String, InFlightSink> = HashMap::new();
|
||||
let in_flight_sinks: Arc<ParkingLotMutex<HashMap<String, InFlightSink>>> =
|
||||
Arc::new(ParkingLotMutex::new(HashMap::new()));
|
||||
let spawned: Arc<ParkingLotMutex<Vec<JoinHandle<()>>>> =
|
||||
Arc::new(ParkingLotMutex::new(Vec::new()));
|
||||
|
||||
loop {
|
||||
let envelope = match reader.read_frame().await {
|
||||
@@ -764,46 +907,29 @@ impl Dispatcher {
|
||||
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;
|
||||
}
|
||||
match self.dispatch_start(&connection, request_id.clone(), envelope.payload) {
|
||||
StartedDispatch::Once(invoke) => {
|
||||
let handle =
|
||||
Self::spawn_once_dispatch(&writer, request_id.clone(), invoke);
|
||||
spawned.lock().push(handle);
|
||||
}
|
||||
DispatchResult::Stream(stream) => {
|
||||
self.pump_stream_single_stream(&writer, &request_id, stream)
|
||||
.await;
|
||||
StartedDispatch::Stream(stream) => {
|
||||
let handle =
|
||||
Self::spawn_stream_pump(&writer, request_id.clone(), stream);
|
||||
spawned.lock().push(handle);
|
||||
}
|
||||
DispatchResult::Sink(sink) => {
|
||||
StartedDispatch::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(
|
||||
let handle = Self::spawn_sink_response_writer(
|
||||
&writer,
|
||||
request_id.clone(),
|
||||
handler,
|
||||
);
|
||||
in_flight_sinks.lock().insert(
|
||||
request_id.clone(),
|
||||
InFlightSink {
|
||||
chunk_tx,
|
||||
@@ -814,9 +940,25 @@ impl Dispatcher {
|
||||
}
|
||||
}
|
||||
}
|
||||
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.lock().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) {
|
||||
let entry = in_flight_sinks.lock().remove(&request_id);
|
||||
if let Some(mut entry) = entry {
|
||||
entry.handler_handle.abort();
|
||||
let _ = entry
|
||||
.chunk_tx
|
||||
@@ -833,7 +975,8 @@ impl Dispatcher {
|
||||
.get("input")
|
||||
.cloned()
|
||||
.unwrap_or(Value::Null);
|
||||
if let Some(mut entry) = in_flight_sinks.remove(&request_id) {
|
||||
let entry = in_flight_sinks.lock().remove(&request_id);
|
||||
if let Some(mut entry) = entry {
|
||||
let validated = match &entry.publish_validator {
|
||||
Some(validator) => {
|
||||
if validator.is_valid(&chunk) {
|
||||
@@ -851,7 +994,7 @@ impl Dispatcher {
|
||||
let keep = validated.is_ok();
|
||||
let _ = entry.chunk_tx.send(validated).await;
|
||||
if keep {
|
||||
in_flight_sinks.insert(request_id, entry);
|
||||
in_flight_sinks.lock().insert(request_id, entry);
|
||||
}
|
||||
} else {
|
||||
debug!(
|
||||
@@ -860,36 +1003,33 @@ impl Dispatcher {
|
||||
);
|
||||
}
|
||||
}
|
||||
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 call_error: CallError = serde_json::from_value(envelope.payload)
|
||||
.unwrap_or_else(|_| {
|
||||
CallError::internal("publish error from initiator (malformed)")
|
||||
});
|
||||
let entry = in_flight_sinks.lock().remove(&request_id);
|
||||
if let Some(mut entry) = entry {
|
||||
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"
|
||||
);
|
||||
pending.lock().handle_error(&request_id, call_error);
|
||||
}
|
||||
}
|
||||
other => {
|
||||
debug!(
|
||||
event_type = %other,
|
||||
id = %envelope.id,
|
||||
"single-stream: ignoring non-requested/non-published/non-aborted/non-completed event"
|
||||
"single-stream: ignoring unknown event type"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
in_flight_sinks.clear();
|
||||
in_flight_sinks.lock().clear();
|
||||
for handle in spawned.lock().drain(..) {
|
||||
handle.abort();
|
||||
}
|
||||
|
||||
let failed = pending
|
||||
.lock()
|
||||
@@ -904,35 +1044,6 @@ impl Dispatcher {
|
||||
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
|
||||
@@ -941,9 +1052,16 @@ impl Dispatcher {
|
||||
/// 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.requested` → start the dispatch and spawn the work (the
|
||||
/// serving half — what the accept side's `run_loop_single_stream`
|
||||
/// does). Once invocations and Sub pumps run as spawned tasks;
|
||||
/// only the Pub sink start runs inline (its `chunk_tx` must be
|
||||
/// registered before the next `call.published` frame is read).
|
||||
/// Inline dispatch was the review 005 G-01 defect: a handler that
|
||||
/// nested-composes over the same connection blocked the one read
|
||||
/// loop that must resolve the nested call's response frame. The
|
||||
/// `SharedFrameWriter` serializes frames, so spawned arms' frames
|
||||
/// do not interleave mid-frame.
|
||||
/// - `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
|
||||
@@ -966,8 +1084,9 @@ impl Dispatcher {
|
||||
/// 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.
|
||||
/// closed`), in-flight inbound sinks are dropped, and the spawned
|
||||
/// Once/Stream/Sink tasks are aborted (their response frames cannot
|
||||
/// reach a closed transport anyway).
|
||||
pub async fn serve_single_stream(
|
||||
self,
|
||||
connection: Arc<CallConnection>,
|
||||
@@ -993,7 +1112,10 @@ impl Dispatcher {
|
||||
});
|
||||
|
||||
let mut reader = FrameFramedReader::new(reader);
|
||||
let mut in_flight_sinks: HashMap<String, InFlightSink> = HashMap::new();
|
||||
let in_flight_sinks: Arc<ParkingLotMutex<HashMap<String, InFlightSink>>> =
|
||||
Arc::new(ParkingLotMutex::new(HashMap::new()));
|
||||
let spawned: Arc<ParkingLotMutex<Vec<JoinHandle<()>>>> =
|
||||
Arc::new(ParkingLotMutex::new(Vec::new()));
|
||||
|
||||
loop {
|
||||
let envelope = match reader.read_frame().await {
|
||||
@@ -1008,46 +1130,29 @@ impl Dispatcher {
|
||||
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;
|
||||
}
|
||||
match self.dispatch_start(&connection, request_id.clone(), envelope.payload) {
|
||||
StartedDispatch::Once(invoke) => {
|
||||
let handle =
|
||||
Self::spawn_once_dispatch(&writer, request_id.clone(), invoke);
|
||||
spawned.lock().push(handle);
|
||||
}
|
||||
DispatchResult::Stream(stream) => {
|
||||
self.pump_stream_single_stream(&writer, &request_id, stream)
|
||||
.await;
|
||||
StartedDispatch::Stream(stream) => {
|
||||
let handle =
|
||||
Self::spawn_stream_pump(&writer, request_id.clone(), stream);
|
||||
spawned.lock().push(handle);
|
||||
}
|
||||
DispatchResult::Sink(sink) => {
|
||||
StartedDispatch::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(
|
||||
let handle = Self::spawn_sink_response_writer(
|
||||
&writer,
|
||||
request_id.clone(),
|
||||
handler,
|
||||
);
|
||||
in_flight_sinks.lock().insert(
|
||||
request_id.clone(),
|
||||
InFlightSink {
|
||||
chunk_tx,
|
||||
@@ -1069,13 +1174,14 @@ impl Dispatcher {
|
||||
}
|
||||
EVENT_COMPLETED => {
|
||||
let request_id = envelope.id.clone();
|
||||
if in_flight_sinks.remove(&request_id).is_none() {
|
||||
if in_flight_sinks.lock().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) {
|
||||
let entry = in_flight_sinks.lock().remove(&request_id);
|
||||
if let Some(mut entry) = entry {
|
||||
entry.handler_handle.abort();
|
||||
let _ = entry
|
||||
.chunk_tx
|
||||
@@ -1092,7 +1198,8 @@ impl Dispatcher {
|
||||
.get("input")
|
||||
.cloned()
|
||||
.unwrap_or(Value::Null);
|
||||
if let Some(mut entry) = in_flight_sinks.remove(&request_id) {
|
||||
let entry = in_flight_sinks.lock().remove(&request_id);
|
||||
if let Some(mut entry) = entry {
|
||||
let validated = match &entry.publish_validator {
|
||||
Some(validator) => {
|
||||
if validator.is_valid(&chunk) {
|
||||
@@ -1110,7 +1217,7 @@ impl Dispatcher {
|
||||
let keep = validated.is_ok();
|
||||
let _ = entry.chunk_tx.send(validated).await;
|
||||
if keep {
|
||||
in_flight_sinks.insert(request_id, entry);
|
||||
in_flight_sinks.lock().insert(request_id, entry);
|
||||
}
|
||||
} else {
|
||||
debug!(
|
||||
@@ -1125,7 +1232,8 @@ impl Dispatcher {
|
||||
.unwrap_or_else(|_| {
|
||||
CallError::internal("publish error from initiator (malformed)")
|
||||
});
|
||||
if let Some(mut entry) = in_flight_sinks.remove(&request_id) {
|
||||
let entry = in_flight_sinks.lock().remove(&request_id);
|
||||
if let Some(mut entry) = entry {
|
||||
let _ = entry.chunk_tx.send(Err(call_error)).await;
|
||||
} else {
|
||||
pending.lock().handle_error(&request_id, call_error);
|
||||
@@ -1141,7 +1249,10 @@ impl Dispatcher {
|
||||
}
|
||||
}
|
||||
|
||||
in_flight_sinks.clear();
|
||||
in_flight_sinks.lock().clear();
|
||||
for handle in spawned.lock().drain(..) {
|
||||
handle.abort();
|
||||
}
|
||||
|
||||
let failed = pending
|
||||
.lock()
|
||||
|
||||
Reference in New Issue
Block a user