glm-5.3-flash
3317ea0963
- open-questions.md: OQ-39 → ADR-052, OQ-40 → ADR-053 (resolutions replace the deferred-until-needed posture) - channel-operations.md: the put contract (initiator allocates + streams, reserved input key, one response, acks-in-result sentence), the control routing (ADR-052), the enumerator aggregation (ADR-053), the connection-owner paragraph's pub-era correction, the ADR table rows - ADR-051 §6: PubTypedOpen → the put translation; ADR-047's 2026-09-06 amendment records its era honestly (the dynamic half is now ADR-053) - review 010: all four findings resolved (with the S-01 sketch correction so the published record stops carrying the wrong-graph reading) - review 011: closed (all seven pins implemented + gates landed) - review 001 C-08: resolved note (odd/even + adopt, implemented) - AGENTS.md: the 'Stub Guard' section — the not_implemented grep is part of cargo test; the registration invariants and the implement-or-stop ruling are release-blocking Verified: cargo test (696+guard), clippy -D warnings, fmt, doc clean.
577 lines
24 KiB
Rust
577 lines
24 KiB
Rust
//! `ChannelRelay` — the in-tree channel relay component (ADR-042 as
|
|
//! amended by ADR-051; review 008 Unit 3a).
|
|
//!
|
|
//! A relay hub imports a producer leg's channel-open op (via
|
|
//! `from_call` discovery — the spec arrives reconstructed WITH the
|
|
//! `channel_open` marker, ADR-047 amendment 3) and re-exposes it on
|
|
//! its own consumer-leg channel-0 fork registry. The open op resolves
|
|
//! on the hub with a hub-allocated `channel_id`; the relay's
|
|
//! establisher (the translate hop, ADR-042 layer 1) opens the channel
|
|
//! on the producer leg and adopts the spoke-allocated id into the
|
|
//! producer-leg `ChannelManager`; the relay's `OpenHandler` (the
|
|
//! byte-forward hop, layer 2) pumps bytes between the two legs' data
|
|
//! planes with `pump_bidi` — the hub never parses chunk framing
|
|
//! (ADR-034/ADR-035: ids are per-connection, no rewrite exists).
|
|
//!
|
|
//! The consumer leg keeps the full open-op wrapper machinery (ACL,
|
|
//! per-identity cap/ledger, establishment bound, teardown-on-failure)
|
|
//! — the relay registers through
|
|
//! [`ChannelCore::register_openable_with_establisher`], never a
|
|
//! parallel authorization path.
|
|
//!
|
|
//! See ADR-042 (the relay contract), ADR-051 (the in-tree design:
|
|
//! §2 what the relay holds, §3 the reason mapping, §4 the
|
|
//! registration seam, §6 the rejection postures).
|
|
|
|
use std::sync::Arc;
|
|
|
|
use serde_json::{Map, Value};
|
|
|
|
use crate::core::auth::AuthContext;
|
|
use crate::protocol::connection::CallConnection;
|
|
use crate::protocol::wire::CallError;
|
|
use crate::registry::registration::OperationRegistry;
|
|
use crate::registry::spec::{OperationSpec, OperationType};
|
|
|
|
use super::manager::ChannelManager;
|
|
use super::operations::{
|
|
ChannelCore, ChannelPlan, Establishment, EstablishmentError, OpenEstablisher, OpenHandler,
|
|
PubOpenHandler, CHANNEL_OPEN_FAILED, RESERVED_REPLY_KEY,
|
|
};
|
|
use super::pump::pump_bidi;
|
|
|
|
use super::reassembly::{MpscRecvStream, MpscSendStream};
|
|
use tokio::io::AsyncWriteExt;
|
|
|
|
/// The producer-leg surface the relay holds (ADR-051 §2): the
|
|
/// `CallConnection` for the translate hop's open call plus the
|
|
/// producer-leg `ChannelManager` for `adopt_channel` and teardown
|
|
/// visibility. Not a `ChannelClient` — the hub's `CallConnection` is
|
|
/// shared by import, relay establisher, and hub ops (three claimants),
|
|
/// so `take_call_connection`'s detach is wrong for a hub. A hub that
|
|
/// dials the spoke with `ChannelClient::from_connection`, imports via
|
|
/// `from_call`, then tears the client down keeps the relay alive on
|
|
/// these `Arc`s.
|
|
#[derive(Clone)]
|
|
pub struct ProducerLeg {
|
|
/// The producer leg's call connection — the hub's channel-0
|
|
/// connection to the spoke, the one `from_call` imported the open
|
|
/// op over.
|
|
pub call: Arc<CallConnection>,
|
|
/// The producer leg's channel manager — the relay's establisher
|
|
/// adopts the spoke-allocated channel id here.
|
|
pub manager: ChannelManager,
|
|
}
|
|
|
|
/// The plan payload the relay's establisher hands its pump handler:
|
|
/// the spoke-leg streams plus the adopted spoke `channel_id`. The
|
|
/// establisher and the handler agree on this concrete type (the
|
|
/// `ChannelPlan` downcast contract, ADR-049 amendment 2); alkcall
|
|
/// never inspects it.
|
|
///
|
|
/// **Adopted-entry ownership (ADR-051 §6):** the plan holds the
|
|
/// producer-leg `ChannelManager` and reclaims the adopted
|
|
/// `spoke_id` entry on drop — including the teardown windows the
|
|
/// pump handler never reaches (the wrapper's establishment bound
|
|
/// expiring after the adopt, the plan dropped before the pump
|
|
/// spawns). Dropping the plan's `spoke_send` also EOFs the spoke
|
|
/// side via the mux pump's implicit-EOF path (REQ-CH-01), so the
|
|
/// spoke handler reclaims through the same cascade the normal path
|
|
/// uses.
|
|
struct RelayPlan {
|
|
spoke_id: u32,
|
|
producer_manager: ChannelManager,
|
|
spoke_send: Option<MpscSendStream>,
|
|
spoke_recv: Option<MpscRecvStream>,
|
|
/// The put path (ADR-047 amendment 4 §4.7): the producer leg's
|
|
/// open-op response, awaited by the relay pub handler after the
|
|
/// consumer EOF (put-acks-in-result — the spoke's response cannot
|
|
/// exist before its byte stream completes, so the establisher
|
|
/// never awaits it).
|
|
pub_reply: Option<tokio::sync::oneshot::Receiver<CallConnectionResponse>>,
|
|
}
|
|
|
|
/// The producer leg's open-op response, carried on the put path's plan.
|
|
type CallConnectionResponse = crate::protocol::wire::ResponseEnvelope;
|
|
|
|
impl Drop for RelayPlan {
|
|
fn drop(&mut self) {
|
|
let _ = self.producer_manager.teardown_channel(self.spoke_id);
|
|
}
|
|
}
|
|
|
|
/// The relay component (ADR-051). Holds the producer-leg surface;
|
|
/// [`ChannelRelay::register_relay_openable`] registers a
|
|
/// from_call-imported marked spec on a consumer-leg channel-0 fork
|
|
/// registry with the translate + byte-forward machinery.
|
|
#[derive(Clone)]
|
|
pub struct ChannelRelay {
|
|
producer_leg: ProducerLeg,
|
|
}
|
|
|
|
impl ChannelRelay {
|
|
/// Construct a relay over a producer leg.
|
|
pub fn new(producer_leg: ProducerLeg) -> Self {
|
|
Self { producer_leg }
|
|
}
|
|
|
|
/// The producer-leg surface this relay translates onto.
|
|
pub fn producer_leg(&self) -> &ProducerLeg {
|
|
&self.producer_leg
|
|
}
|
|
|
|
/// Register a from_call-imported marked spec as a relay openable
|
|
/// on the consumer-leg channel-0 registry (ADR-051 §4 phase 2 —
|
|
/// the fork is the `consumer_registry` argument).
|
|
///
|
|
/// - `consumer_core` is the consumer leg's [`ChannelCore`] (its
|
|
/// channel-0 manager + lifecycle policy) — the full open-op
|
|
/// wrapper machinery (ACL, cap/ledger, establishment bound,
|
|
/// teardown) runs on the consumer leg exactly as for a locally
|
|
/// produced open op.
|
|
/// - `spec` must carry the `channel_open` marker (reconstructed
|
|
/// WITH the marker by discovery, ADR-047 amendment 3). A
|
|
/// `Pub`-typed marked spec composes through the put-forwarding
|
|
/// translation (`register_relay_pub_openable`; ADR-047 amendment 4
|
|
/// §4.7); an
|
|
/// unmarked spec is a
|
|
/// [`RelayError::UnmarkedSpec`] error — branch on the
|
|
/// marker before calling.
|
|
/// - The translate hop calls the producer leg with the spec's own
|
|
/// op name; a hub that re-names the consumer-visible op composes
|
|
/// its own establisher via
|
|
/// [`ChannelCore::register_openable_with_establisher`] instead.
|
|
pub fn register_relay_openable(
|
|
&self,
|
|
consumer_core: &ChannelCore,
|
|
consumer_registry: &OperationRegistry,
|
|
spec: OperationSpec,
|
|
consumer_auth: AuthContext,
|
|
) -> Result<(), RelayError> {
|
|
if spec.channel_open.is_none() {
|
|
return Err(RelayError::UnmarkedSpec {
|
|
message: format!(
|
|
"spec `{}` has no channel_open marker — relay registration is \
|
|
for from_call-imported marked specs (ADR-051 §6)",
|
|
spec.name
|
|
),
|
|
});
|
|
}
|
|
if spec.op_type == OperationType::Pub {
|
|
return self.register_relay_pub_openable(
|
|
consumer_core,
|
|
consumer_registry,
|
|
spec,
|
|
consumer_auth,
|
|
);
|
|
}
|
|
|
|
let producer_call = Arc::clone(&self.producer_leg.call);
|
|
let producer_manager = self.producer_leg.manager.clone();
|
|
let op_name = spec.name.clone();
|
|
let establisher: OpenEstablisher = Arc::new(move |input, per_call_auth| {
|
|
let producer_call = Arc::clone(&producer_call);
|
|
let producer_manager = producer_manager.clone();
|
|
let op_name = op_name.clone();
|
|
Box::pin(async move {
|
|
open_on_producer_leg(
|
|
&producer_call,
|
|
&producer_manager,
|
|
&op_name,
|
|
input,
|
|
&per_call_auth,
|
|
)
|
|
.await
|
|
})
|
|
});
|
|
|
|
let open_handler: OpenHandler = Arc::new(move |_input, plan, channel_conn, _auth| {
|
|
tokio::spawn(async move {
|
|
let Some(plan) = plan else {
|
|
return;
|
|
};
|
|
// The plan is uniquely held here (establisher →
|
|
// wrapper → handler, moved each hop), so `try_unwrap`
|
|
// succeeds; on the phantom-holder case the wrapper's
|
|
// no-accept telemetry fires and the spoke leg EOFs
|
|
// when the plan finally drops (its Drop guard reclaims
|
|
// the adopted entry whenever that drop happens).
|
|
let Ok(relay_plan) = plan.downcast::<RelayPlan>() else {
|
|
tracing::debug!("relay: establisher plan of unexpected type");
|
|
return;
|
|
};
|
|
let Ok(mut relay_plan) = Arc::try_unwrap(relay_plan) else {
|
|
tracing::debug!("relay: establisher plan held elsewhere; skipping pump");
|
|
return;
|
|
};
|
|
let Ok(bidi) = channel_conn.accept_bi().await else {
|
|
return;
|
|
};
|
|
// Take the streams out of the plan; the plan itself
|
|
// stays alive as the adopted-entry drop guard (see
|
|
// `RelayPlan`) until this task's end.
|
|
let Some(spoke_recv) = relay_plan.spoke_recv.take() else {
|
|
return;
|
|
};
|
|
let Some(spoke_send) = relay_plan.spoke_send.take() else {
|
|
return;
|
|
};
|
|
// The byte-forward hop (ADR-042 layer 2, ADR-050):
|
|
// two pumps joined inline (R-02). On completion the
|
|
// pump's send-half drops EOF the spoke leg (the mux
|
|
// pump's implicit-EOF sentinel) and the plan's Drop
|
|
// guard reclaims the adopted producer-leg entry; the
|
|
// wrapper's handler-exit teardown covers the consumer
|
|
// leg.
|
|
let _ = pump_bidi(bidi, spoke_recv, spoke_send).await;
|
|
})
|
|
});
|
|
|
|
consumer_core
|
|
.register_openable_with_establisher(
|
|
spec,
|
|
Some(establisher),
|
|
open_handler,
|
|
consumer_registry,
|
|
consumer_auth,
|
|
None,
|
|
)
|
|
.map_err(|e| RelayError::Registration { message: e })
|
|
}
|
|
|
|
/// Register a from_call-imported **marked `Pub`** spec as a relay
|
|
/// openable — the put-forwarding translation (ADR-047 amendment 4
|
|
/// §4.7). The consumer leg runs the full marked-pub wrapper (adopt
|
|
/// → establish → pub handler awaited inline); this side's compose:
|
|
///
|
|
/// - the establisher allocates the hub's connect-side id on the
|
|
/// producer leg (`open_channel` — an odd id, disjoint from the
|
|
/// spoke's even allocations), sends the open op **in the
|
|
/// background** (the spoke's put response cannot exist before
|
|
/// its byte stream completes; an establisher that awaited it
|
|
/// deadlocks the leg), and plans the spoke-leg streams + the
|
|
/// response's oneshot; the plan's Drop guard reclaims the
|
|
/// producer-leg entry on every handler path, as ADR-051 §6 pins.
|
|
/// - the handler pumps the consumer-leg bytes → the spoke leg
|
|
/// (one direction, pub), EOFs the spoke side on consumer EOF,
|
|
/// then awaits the spoke's single response, strips the
|
|
/// producer's reserved reply key (the consumer reply carries the
|
|
/// hub-allocated consumer-leg id; a spoke `channel_id` merged
|
|
/// through would shadow it — the same reserved-key class the
|
|
/// wrapper refuses), and returns the remainder — the consumer-leg
|
|
/// wrapper merges it with the pinned order (ack fields are none
|
|
/// here; the put acks in the spoke's result, §4.5).
|
|
///
|
|
/// A spoke open-time establishment failure surfaces verbatim in
|
|
/// the final response (put-acks-in-result; the spoke's mapped
|
|
/// `channel:open_failed` rides the response, not the establisher).
|
|
fn register_relay_pub_openable(
|
|
&self,
|
|
consumer_core: &ChannelCore,
|
|
consumer_registry: &OperationRegistry,
|
|
spec: OperationSpec,
|
|
consumer_auth: AuthContext,
|
|
) -> Result<(), RelayError> {
|
|
let producer_call = Arc::clone(&self.producer_leg.call);
|
|
let producer_manager = self.producer_leg.manager.clone();
|
|
let op_name = spec.name.clone();
|
|
|
|
let establisher: OpenEstablisher = Arc::new(move |input, per_call_auth| {
|
|
let producer_call = Arc::clone(&producer_call);
|
|
let producer_manager = producer_manager.clone();
|
|
let op_name = op_name.clone();
|
|
Box::pin(async move {
|
|
open_pub_on_producer_leg(
|
|
&producer_call,
|
|
&producer_manager,
|
|
&op_name,
|
|
input,
|
|
&per_call_auth,
|
|
)
|
|
.await
|
|
})
|
|
});
|
|
|
|
let pub_open_handler: PubOpenHandler =
|
|
Arc::new(move |_input, plan, channel_conn, _auth| {
|
|
Box::pin(async move { pub_relay_forward(channel_conn, plan).await })
|
|
});
|
|
|
|
consumer_core
|
|
.register_pub_openable(
|
|
spec,
|
|
Some(establisher),
|
|
pub_open_handler,
|
|
consumer_registry,
|
|
consumer_auth,
|
|
None,
|
|
)
|
|
.map_err(|e| RelayError::Registration { message: e })
|
|
}
|
|
}
|
|
|
|
/// The translate hop (ADR-051 §1 layer 1, §2): call the producer leg's
|
|
/// open op with the forwarded payload, adopt the returned spoke
|
|
/// `channel_id` into the producer-leg manager, and return
|
|
/// `Establishment::new(plan)` whose plan carries the spoke-leg
|
|
/// streams. The spoke reply's `channel_id` is stripped (the consumer
|
|
/// reply carries the hub-allocated id); the reply's other fields ride
|
|
/// `with_reply_fields` (`bound` flows end-to-end, per-hop truthful).
|
|
async fn open_on_producer_leg(
|
|
producer_call: &Arc<CallConnection>,
|
|
producer_manager: &ChannelManager,
|
|
op_name: &str,
|
|
input: Value,
|
|
per_call_auth: &AuthContext,
|
|
) -> Result<Establishment, EstablishmentError> {
|
|
let payload = build_relay_payload(op_name, input, per_call_auth);
|
|
let response = producer_call.call_with_payload(payload).await;
|
|
let out = match response.result {
|
|
Ok(out) => out,
|
|
Err(spoke_error) => return Err(map_spoke_error(&spoke_error)),
|
|
};
|
|
|
|
let spoke_id = out
|
|
.get("channel_id")
|
|
.and_then(|v| v.as_u64())
|
|
.ok_or_else(|| EstablishmentError::HandlerError {
|
|
message: format!(
|
|
"spoke open reply for `{op_name}` missing `channel_id` — malformed responder"
|
|
),
|
|
})? as u32;
|
|
|
|
// Adopt the spoke-allocated id into the producer-leg manager so
|
|
// the spoke leg's mux/demux route this channel — the same shape
|
|
// `ChannelClient::open_channel` performs, minus the client. The
|
|
// producer leg is the connect side of the hub↔spoke channels
|
|
// connection, so the spoke's ids are even and disjoint from the
|
|
// hub's odd connect-side allocations.
|
|
let (spoke_send, spoke_recv) = producer_manager
|
|
.adopt_channel(spoke_id, "alk/relay", None)
|
|
.await
|
|
.map_err(|e| EstablishmentError::ResourceShortage {
|
|
message: format!(
|
|
"spoke channel {spoke_id} adopt failed on the hub's producer leg: {e}"
|
|
),
|
|
})?;
|
|
|
|
let mut reply_fields = Map::new();
|
|
if let Some(obj) = out.as_object() {
|
|
for (key, value) in obj {
|
|
if key != RESERVED_REPLY_KEY {
|
|
reply_fields.insert(key.clone(), value.clone());
|
|
}
|
|
}
|
|
}
|
|
|
|
let establishment = Establishment::new(Arc::new(RelayPlan {
|
|
spoke_id,
|
|
producer_manager: producer_manager.clone(),
|
|
spoke_send: Some(spoke_send),
|
|
spoke_recv: Some(spoke_recv),
|
|
pub_reply: None,
|
|
}) as super::operations::ChannelPlan);
|
|
Ok(if reply_fields.is_empty() {
|
|
establishment
|
|
} else {
|
|
establishment.with_reply_fields(reply_fields)
|
|
})
|
|
}
|
|
|
|
/// Map a spoke failure into the establisher vocabulary (ADR-051 §3 —
|
|
/// the reason table; the spoke's code and message are preserved in the
|
|
/// mapped variant's `message`, never discarded — a consumer debugging
|
|
/// a FORBIDDEN-through-relay sees the spoke's grant failure).
|
|
///
|
|
/// The `timeout` row is the one non-1:1 case: the consumer-visible
|
|
/// reason is `dial_failed` because establishment timeouts are
|
|
/// wrapper-generated (ADR-049 §3) and not an establisher vocabulary
|
|
/// word; the spoke's message carries the truth.
|
|
fn map_spoke_error(spoke_error: &CallError) -> EstablishmentError {
|
|
if spoke_error.code == CHANNEL_OPEN_FAILED {
|
|
let reason = spoke_error
|
|
.details
|
|
.as_ref()
|
|
.and_then(|d| d.get("reason"))
|
|
.and_then(|r| r.as_str())
|
|
.unwrap_or("");
|
|
let detail = spoke_error
|
|
.details
|
|
.as_ref()
|
|
.and_then(|d| d.get("message"))
|
|
.and_then(|m| m.as_str())
|
|
.unwrap_or(spoke_error.message.as_str());
|
|
let message = format!("spoke leg: {reason}: {detail}");
|
|
return match reason {
|
|
"dial_failed" | "timeout" => EstablishmentError::DialFailed { message },
|
|
"unknown_resource" => EstablishmentError::UnknownResource { message },
|
|
"resource_shortage" => EstablishmentError::ResourceShortage { message },
|
|
_ => EstablishmentError::HandlerError { message },
|
|
};
|
|
}
|
|
// `NOT_FOUND` (spoke predates the op), `FORBIDDEN` (hub lacks the
|
|
// spoke grant), and every other call-error class.
|
|
EstablishmentError::HandlerError {
|
|
message: format!("spoke leg: {} ({})", spoke_error.code, spoke_error.message),
|
|
}
|
|
}
|
|
|
|
/// The put path's translate hop (ADR-047 amendment 4 §4.7): allocate
|
|
/// the hub's connect-side channel id on the producer leg, send the
|
|
/// spoke's open op **in the background** with the reserved input key
|
|
/// set, and plan the spoke-leg streams + the response oneshot. The
|
|
/// establisher returns promptly; the spoke's response arrives when the
|
|
/// spoke's put completes (or fails) — put-acks-in-result.
|
|
async fn open_pub_on_producer_leg(
|
|
producer_call: &Arc<CallConnection>,
|
|
producer_manager: &ChannelManager,
|
|
op_name: &str,
|
|
input: Value,
|
|
per_call_auth: &AuthContext,
|
|
) -> Result<Establishment, EstablishmentError> {
|
|
let (spoke_id, spoke_send, spoke_recv) = producer_manager
|
|
.open_channel(
|
|
"alk/relay",
|
|
// The opener ledger is inert on the hub's connect side
|
|
// (the connect side does not enforce the per-identity cap,
|
|
// ADR-041) — the record is observability only.
|
|
"hub-relay",
|
|
None,
|
|
)
|
|
.await
|
|
.map_err(|e| EstablishmentError::ResourceShortage {
|
|
message: format!("producer leg: channel allocation failed: {e}"),
|
|
})?;
|
|
|
|
let mut hop_input = input.clone();
|
|
if let Some(obj) = hop_input.as_object_mut() {
|
|
obj.insert(
|
|
crate::registry::spec::RESERVED_OPEN_INPUT_KEY.to_string(),
|
|
serde_json::json!(spoke_id),
|
|
);
|
|
}
|
|
let payload = build_relay_payload(op_name, hop_input, per_call_auth);
|
|
|
|
let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
|
|
let producer_call_task = Arc::clone(producer_call);
|
|
tokio::spawn(async move {
|
|
let _ = reply_tx.send(producer_call_task.call_with_payload(payload).await);
|
|
});
|
|
|
|
Ok(Establishment::new(Arc::new(RelayPlan {
|
|
spoke_id,
|
|
producer_manager: producer_manager.clone(),
|
|
spoke_send: Some(spoke_send),
|
|
spoke_recv: Some(spoke_recv),
|
|
pub_reply: Some(reply_rx),
|
|
}) as ChannelPlan))
|
|
}
|
|
|
|
/// The relay pub handler (ADR-047 amendment 4 §4.7): pump the
|
|
/// consumer-leg bytes → the spoke leg (one direction, pub — the
|
|
/// shutdown-on-completion contract, ADR-050's), then await the spoke's
|
|
/// single response and strip its reserved reply key. The plan's Drop
|
|
/// guard reclaims the producer-leg entry on every path.
|
|
async fn pub_relay_forward(
|
|
channel_conn: crate::core::types::Connection,
|
|
plan: Option<ChannelPlan>,
|
|
) -> Result<Value, CallError> {
|
|
let Some(plan) = plan else {
|
|
return Err(CallError::internal(
|
|
"relay: pub establisher produced no plan",
|
|
));
|
|
};
|
|
let Ok(relay_plan_arc) = plan.downcast::<RelayPlan>() else {
|
|
return Err(CallError::internal(
|
|
"relay: establisher plan of unexpected type",
|
|
));
|
|
};
|
|
let Ok(mut relay_plan) = Arc::try_unwrap(relay_plan_arc) else {
|
|
return Err(CallError::internal(
|
|
"relay: establisher plan held elsewhere; skipping pump",
|
|
));
|
|
};
|
|
let Some(mut spoke_send) = relay_plan.spoke_send.take() else {
|
|
return Err(CallError::internal(
|
|
"relay: pub plan missing the spoke send half",
|
|
));
|
|
};
|
|
relay_plan.spoke_recv.take();
|
|
|
|
let Ok(bidi) = channel_conn.accept_bi().await else {
|
|
return Err(CallError::internal(
|
|
"relay: consumer leg data plane never opened",
|
|
));
|
|
};
|
|
let (mut read_half, _write_half) = tokio::io::split(bidi);
|
|
// The one-direction pump: consumer bytes → spoke; shutdown on
|
|
// consumer EOF (the pump contract, ADR-050).
|
|
let _copied = tokio::io::copy(&mut read_half, &mut spoke_send)
|
|
.await
|
|
.unwrap_or(0);
|
|
let _ = spoke_send.shutdown().await;
|
|
drop(spoke_send);
|
|
|
|
match relay_plan.pub_reply.take() {
|
|
Some(rx) => match rx.await {
|
|
Ok(response) => match response.result {
|
|
Ok(mut out) => {
|
|
if let Some(obj) = out.as_object_mut() {
|
|
obj.remove(RESERVED_REPLY_KEY);
|
|
}
|
|
Ok(out)
|
|
}
|
|
Err(e) => Err(e),
|
|
},
|
|
Err(_) => Err(CallError::internal(
|
|
"relay: producer-leg open resolved without a response",
|
|
)),
|
|
},
|
|
None => Err(CallError::internal(
|
|
"relay: pub plan missing the response receiver",
|
|
)),
|
|
}
|
|
}
|
|
|
|
/// Build the `call.requested` payload for the translate hop: the hub
|
|
/// as caller, the consumer as `forwarded_for` (ADR-026 §3, ADR-032 §3
|
|
/// — the same shape `build_forwarded_payload` produces for the plain
|
|
/// forwarding stubs). `forwarded_for` is metadata, never consulted by
|
|
/// any `AccessControl::check`; it is omitted when the consumer's call
|
|
/// carried no resolved identity.
|
|
fn build_relay_payload(op_name: &str, input: Value, per_call_auth: &AuthContext) -> Value {
|
|
let mut payload = Map::new();
|
|
payload.insert(
|
|
"operationId".to_string(),
|
|
Value::String(op_name.to_string()),
|
|
);
|
|
payload.insert("input".to_string(), input);
|
|
if let Some(originator) = &per_call_auth.identity {
|
|
if let Ok(value) = serde_json::to_value(originator) {
|
|
payload.insert("forwarded_for".to_string(), value);
|
|
}
|
|
}
|
|
Value::Object(payload)
|
|
}
|
|
|
|
/// The assembly-time errors `ChannelRelay::register_relay_openable`
|
|
/// produces (ADR-051 §6 — the rejection postures are loud, never
|
|
/// silent stubs; the `PubTypedOpen` variant was removed by ADR-047
|
|
/// amendment 4 §4.7 — pub-typed marked specs translate now, never
|
|
/// reject-and-strand).
|
|
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
|
|
pub enum RelayError {
|
|
/// The spec carries no `channel_open` marker — the relay only
|
|
/// registers marked specs.
|
|
#[error("unmarked spec: {message}")]
|
|
UnmarkedSpec { message: String },
|
|
/// The consumer-leg registration itself failed (schema compile,
|
|
/// handler-kind mismatch).
|
|
#[error("registration failed: {message}")]
|
|
Registration { message: String },
|
|
}
|
|
|
|
#[cfg(test)]
|
|
#[path = "relay_tests.rs"]
|
|
mod tests;
|