alkdev/alktunnels
6
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docs: remove em dashes from Rust doc comments

Browse Source 
Rewrote each em-dash construction in doc and inline comments with
varied prose (colons, semicolons, parentheses, split sentences);
no code or identifiers changed. The one string-literal em dash
(UDP oversize error, local/mod.rs) became '; '.
No string-literal em dashes remain except none were found beyond it.

Verification:
- grep -rn '—' src/ --include='*.rs' -> no output
- cargo test: all ok (125 tests)
- cargo clippy --all-targets: 0 warnings
- cargo fmt --check: clean
This commit is contained in:
glm-5.3-flash committed 2026-11-03 13:48:51 +00:00
1 parent 92c3cc5123
commit 6b20924f68
7 files changed
+308 -307
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/consumer.rs
+47 -47
View File
@@ -1,4 +1,4 @@
//! The consumer half: `TunnelSession` — the typed client for tunnel
//! The consumer half: `TunnelSession`, the typed client for tunnel
//! channels (ADR-005). Forward path (`open`), reverse path (`adopt`
//! and `open_reverse_channel`), substrate-shaped data planes
//! (`stream_halves`, `take_halves`, `send_datagram`, `recv_datagram`),
@@ -6,7 +6,7 @@
//! (`close`, `join`, `Drop`).
//!
//! One session = one channel = one tunnel (the channel ID is the flow
//! key, OQ-TN-02). The session owns the adopted channel entry —
//! key, OQ-TN-02). The session owns the adopted channel entry;
//! nothing upstream awaits the adopter's pump (the W3 gap); the
//! session closes it structurally. Role follows the resource, not the
//! connection (OQ-TN-03): `-L` and `-R` consumers share this one API.
@@ -44,7 +44,7 @@ use crate::wire::{frame_datagram, DatagramReader};
///
/// Fields are `Option` so `take_halves`/`pump_against` can move the
/// data plane and pump out of the `Drop`-impl type; an emptied session
/// drops as a no-op (teardown is idempotent — an already-reaped entry
/// drops as a no-op (teardown is idempotent; an already-reaped entry
/// reaps `Err(UnknownChannel)`, ignored).
pub struct TunnelSession {
pub channel_id: u32,
@@ -54,7 +54,7 @@ pub struct TunnelSession {
}
enum DataPlane {
/// Raw pass-through (stream substrates: `tcp`, `unix`) — the
/// Raw pass-through (stream substrates: `tcp`, `unix`): the
/// halves ARE the tunnel (ADR-003); zero tunnel-level framing.
Stream(StreamHalves),
/// The mandatory `[len: u16 BE]` codec over the halves (`udp`,
@@ -71,7 +71,7 @@ struct DatagramHalves {
read: ReadHalf<BiStream>,
write: WriteHalf<BiStream>,
reader: DatagramReader,
/// Datagrams decoded from a chunk beyond the one returned — a
/// Datagrams decoded from a chunk beyond the one returned; a
/// single chunk may batch several frames (the codec decodes all
/// completions per feed); they queue here for the next
/// `recv_datagram`.
@@ -84,7 +84,7 @@ impl TunnelSession {
/// (`ChannelClient::open_channel(OP_TUNNEL_OPEN, params,
/// TUNNEL_ALPN)`), adopts the returned channel ID, splits the
/// channel `BiStream`, and presents the substrate-shaped data
/// plane. A failed open never yields a session (typed errors —
/// plane. A failed open never yields a session (typed errors,
/// ADR-049 §4; no phantom session).
pub async fn open(
client: &ChannelClient,
@@ -108,22 +108,22 @@ impl TunnelSession {
}
/// Open a tunnel channel toward a CALLER-NAMED target (the direct
/// op — dynamic-target egress, ADR-007; the ssh `direct-tcpip` /
/// op: dynamic-target egress, ADR-007; the ssh `direct-tcpip` /
/// SOCKS5 CONNECT shape). A separate constructor from [`open`](Self::open),
/// not a params enum: `tunnel:direct` and `tunnel:open` are two
/// scopes — two capabilities — and the caller states which one it
/// scopes (two capabilities), and the caller states which one it
/// holds.
///
/// Serializes `{substrate, target}`, calls
/// `open_channel(OP_TUNNEL_DIRECT, …, TUNNEL_ALPN)` (plain
/// `open_channel` — dial establishers never bind, so no reply
/// `open_channel`; dial establishers never bind, so no reply
/// fields are expected), and presents the identical session type,
/// data planes, and teardown API as [`open`](Self::open).
///
/// Error surface: the base open errors minus `unknown_resource`
/// (no registry lookup); a malformed target is the schema /
/// `invalid_input`-class failure (registry-side); an old producer
/// rejects the unknown op loudly (`NOT_FOUND` — the SSH posture).
/// rejects the unknown op loudly (`NOT_FOUND`; the SSH posture).
pub async fn open_direct(
client: &ChannelClient,
target: SubstrateAddr,
@@ -150,13 +150,13 @@ impl TunnelSession {
/// Adopt a producer-allocated channel ID (the reverse path, `-R`):
/// install the data plane from the adopted halves. Early arrivals
/// are parked by the manager and drained on adopt (the adoption
/// race — ADR-047 §5). An adopt failure is `AdoptFailed`-class on
/// race, ADR-047 §5). An adopt failure is `AdoptFailed`-class on
/// the typed error surface, not an establishment error.
///
/// The substrate shapes the data plane (raw halves vs the UDP
/// codec) and rode the open call that produced the `channel_id`;
/// `adopt` takes it so the session presents the right variant (the
/// task sketch omitted it — a datagram session cannot exist without
/// task sketch omitted it: a datagram session cannot exist without
/// it; documented in the task notes).
pub async fn adopt(
manager: &ChannelManager,
@@ -204,7 +204,7 @@ impl TunnelSession {
}
/// Borrow the raw halves (stream substrates). The halves ARE the
/// tunnel — raw pass-through (ADR-003). For owned access, use
/// tunnel: raw pass-through (ADR-003). For owned access, use
/// [`TunnelSession::take_halves`]. `None` for datagram sessions
/// (drive those with `send_datagram`/`recv_datagram`) or after the
/// halves were taken.
@@ -226,11 +226,11 @@ impl TunnelSession {
/// Take the raw halves (either substrate): consumes the session's
/// data plane into owned boxed halves. The session stays alive as
/// the channel-entry owner — reap with `close`/`join`/`Drop`
/// the channel-entry owner; reap with `close`/`join`/`Drop`
/// (`join` on a pump-less session completes immediately, reaps
/// only, `(0, 0, reaped)`). The halves may be taken once; later
/// data-plane calls fail with [`TunnelIoError::ChannelTaken`].
/// Datagram halves carry the codec framing (ADR-003) — raw bytes
/// Datagram halves carry the codec framing (ADR-003): raw bytes
/// on the wire are `[len: u16 BE][payload]` frames.
pub fn take_halves(mut self) -> TakenHalves {
let manager = self.manager.take();
@@ -261,7 +261,7 @@ impl TunnelSession {
/// UDP: send one datagram (frame → write → flush; ADR-003's
/// mandatory framing). `Oversize` above 65535 (the u16 length
/// field would wrap). Empty payloads are legal (`len=0` — the
/// field would wrap). Empty payloads are legal (`len=0`, the
/// F-2 layering).
pub async fn send_datagram(&mut self, payload: &[u8]) -> Result<(), TunnelIoError> {
match self.data.as_mut() {
@@ -278,7 +278,7 @@ impl TunnelSession {
/// UDP: datagrams decoded from a chunk beyond the one returned to
/// the caller, queued for the next [`recv_datagram`](Self::recv_datagram)
/// (the `DatagramHalves.pending` backlog — the diagnostic the
/// (the `DatagramHalves.pending` backlog, the diagnostic the
/// wave-2 fuzz harness asserts against, `docs/research/fuzzing.md`
/// §3.3). `None` for stream sessions or after the halves were
/// taken (the [`stream_halves`](Self::stream_halves) Option
@@ -292,8 +292,8 @@ impl TunnelSession {
}
/// UDP: receive the next datagram (boundary-preserving, the
/// incremental decode path — the only consumer-visible decode
/// path). `Some(bytes)` per datagram — possibly empty (`len=0`
/// incremental decode path; the only consumer-visible decode
/// path). `Some(bytes)` per datagram, possibly empty (`len=0`
/// legal); `None` only on stream EOF (the channels-level sentinel;
/// the codec never collides with it, F-2). A stream end
/// mid-datagram is [`TunnelIoError::TruncatedDatagram`]
@@ -321,14 +321,14 @@ impl TunnelSession {
/// Spawn `pump_bidi(channel_halves, accepted_read, accepted_write)`
/// against an accepted local connection and hold the returned
/// handle (the reverse-flow use; ADR-005). Builder-style: returns
/// the session, which now owns the pump — `close` aborts it,
/// the session, which now owns the pump: `close` aborts it,
/// `join` awaits it, `Drop` aborts it. Consumes the data plane
/// (the pump owns the channel halves from here).
///
/// For UDP sessions the accepted halves must present the same
/// framing as the channel side (ADR-003): hand the FRAMED adapter
/// (the `local` feature's framing shape, or an assembly
/// equivalent), never a raw UDP socket — the pump copies raw
/// equivalent), never a raw UDP socket; the pump copies raw
/// bytes and cannot know which side carries the `[len: u16 BE]`
/// framing.
pub async fn pump_against<S>(mut self, accepted: S) -> Self
@@ -370,7 +370,7 @@ impl TunnelSession {
.unwrap_or(false)
}
/// Await pump completion (both directions finished — half-close
/// Await pump completion (both directions finished; half-close
/// semantics fall out of `pump_bidi`, W4), THEN reap the adopted
/// channel. Returns the `(u64, u64)` copy counts for
/// observability plus the reap result. **Pump-less sessions**
@@ -401,7 +401,7 @@ impl TunnelSession {
impl Drop for TunnelSession {
/// Abort the pump and sync-reap the channel (`teardown_channel` is
/// sync — best-effort; `Drop` cannot await). Dropping without
/// sync, best-effort; `Drop` cannot await). Dropping without
/// close/join never leaks the channel entry (ADR-005).
fn drop(&mut self) {
if let Some(pump) = self.pump.take() {
@@ -422,29 +422,29 @@ pub struct TakenHalves {
pub write: Box<dyn AsyncWrite + Send + Unpin>,
}
/// The listen-flow consumer session (ADR-008 §4) — the sibling of
/// [`TunnelSession`]: it OPENS a listen tunnel (`channels/tunnel/sub`
/// — the establisher shape is the producer's per-registration choice,
/// The listen-flow consumer session (ADR-008 §4), the sibling of
/// [`TunnelSession`]: it OPENS a listen tunnel (`channels/tunnel/sub`:
/// the establisher shape is the producer's per-registration choice,
/// invisible here), reads the bind-first reply fields (`bound` /
/// `listen`) via the lenient [`crate::params::listen_reply_fields`]
/// projection, and owns the listen channel's teardown (ADR-005's
/// ownership posture extended to the listen flow).
///
/// **The held stream (the one new thing):** a listen session's channel
/// stream has NO data-plane role — its only job is to EOF when the
/// stream has NO data-plane role; its only job is to EOF when the
/// session drops. The struct holds the channel's write half
/// (`MpscSendStream`, the mux sender) IDLE and never pumps it; the
/// drop → EOF ordering is structural (the mux sender drops → the mux
/// pump writes the zero-length EOF sentinel → the far accept
/// dispatcher's idle-stream watch fires and tears the far listener
/// down — ADR-008 §1's contract). The stream is deliberately not
/// down, per ADR-008 §1's contract). The stream is deliberately not
/// exposed: pumping on a listen channel is meaningless (the far side's
/// dispatcher ignores it too).
///
/// **Forwarded sessions are NOT owned by the listener.** They arrive
/// as ordinary adopted-side [`TunnelSession`]s the assembly correlates
/// via the [`listen`](Self::listen) token and owns (the correlation
/// registry is assembly-side bookkeeping — consumer.md's posture).
/// registry is assembly-side bookkeeping, consumer.md's posture).
/// Listener teardown does NOT cascade to already-adopted forwarded
/// sessions; the far listener dying stops NEW accepts only.
///
@@ -458,22 +458,22 @@ pub struct TakenHalves {
/// ```
///
/// Fields are `Option` so `close` can move them out of the `Drop`-impl
/// type; an emptied listener drops as a no-op (teardown is idempotent —
/// type; an emptied listener drops as a no-op (teardown is idempotent:
/// an already-reaped entry reaps `Err(UnknownChannel)`, ignored).
pub struct TunnelListener {
/// The adopted listen channel's ID (one session = one channel —
/// The adopted listen channel's ID (one session = one channel;
/// OQ-TN-02's resolution; observability + correlation).
pub channel_id: u32,
manager: Option<ChannelManager>,
/// The `bound` reply field (the real OS-chosen address) — `None`
/// The `bound` reply field (the real OS-chosen address); `None`
/// for accept-first/dial establishers or old producers (the Option
/// posture). Informational surface, NOT addressing.
bound: Option<SubstrateAddr>,
/// The `listen` reply field (the producer-minted correlation
/// token) — `None` for establishers that mint nothing (the Option
/// token); `None` for establishers that mint nothing (the Option
/// posture).
token: Option<u32>,
/// The listen channel's write half, held IDLE — the structural
/// The listen channel's write half, held IDLE, the structural
/// teardown signal (see the type docs). Never pumped, never
/// exposed.
write: Option<MpscSendStream>,
@@ -482,18 +482,18 @@ pub struct TunnelListener {
impl TunnelListener {
/// Open a listen tunnel (the listen flow, ADR-008 §4). Calls the
/// open op through [`ChannelClient::open_channel_with_reply`] (the
/// U-2 surface, alkcall 0.8.0) — the full reply arrives, the
/// U-2 surface, alkcall 0.8.0); the full reply arrives, the
/// bind-first reply fields are projected via
/// [`crate::params::listen_reply_fields`], and the already-adopted
/// channel entry is held (adopt-free construction — the client
/// channel entry is held (adopt-free construction: the client
/// adopted it). A failed open never yields a listener (typed
/// errors — ADR-049 §4; no phantom session).
/// errors, ADR-049 §4; no phantom session).
///
/// `bound()`/`listen()` surface `None` when the establisher
/// supplied nothing (accept-first/dial producers, or old producers
/// — the additive posture); `bound` is informational surface and
/// supplied nothing (accept-first/dial producers, or old producers;
/// the additive posture); `bound` is informational surface and
/// MUST NOT be treated as a dial target (the producer owns the
/// backing — ADR-001's invariant).
/// backing, ADR-001's invariant).
pub async fn open(
client: &ChannelClient,
params: TunnelParams,
@@ -516,7 +516,7 @@ impl TunnelListener {
})
}
/// The `bound` reply field — the real OS-chosen bound address
/// The `bound` reply field: the real OS-chosen bound address
/// (port `0` resolved), present only when the establisher supplied
/// it (bind-first producers). reply#1-class fidelity (BND.ADDR).
/// Informational surface, never addressing.
@@ -524,7 +524,7 @@ impl TunnelListener {
self.bound.as_ref()
}
/// The `listen` reply field — the correlation token (ADR-008
/// The `listen` reply field: the correlation token (ADR-008
/// Amendment 1) the producer minted for this listen tunnel. The
/// assembly keys its forwarded-op correlation map on it; the
/// listener does not own the receipt queue (forwarded sessions
@@ -534,7 +534,7 @@ impl TunnelListener {
}
/// Tear the listen tunnel down: shut the held write half (the
/// graceful EOF sentinel — the far accept dispatcher's teardown
/// graceful EOF sentinel, the far accept dispatcher's teardown
/// signal), then reap the adopted channel entry
/// (`teardown_channel`). Returns whether the entry existed. The
/// far-side teardown cascades (EOF → accept-loop cancel → far
@@ -554,8 +554,8 @@ impl TunnelListener {
impl Drop for TunnelListener {
/// Drop the held write half (the mux pump writes the implicit-EOF
/// chunk — the far dispatcher's teardown signal) and sync-reap the
/// adopted channel entry (`teardown_channel` is sync —
/// chunk, the far dispatcher's teardown signal) and sync-reap the
/// adopted channel entry (`teardown_channel` is sync,
/// best-effort; `Drop` cannot await). Dropping without close never
/// leaks the channel entry (ADR-005's Drop posture, listen-
/// flavored).
@@ -568,7 +568,7 @@ impl Drop for TunnelListener {
}
/// Read from the chunk stream until one complete datagram is
/// assembled (the incremental decode path — ADR-003). `Ok(None)` on
/// assembled (the incremental decode path, ADR-003). `Ok(None)` on
/// stream EOF; mid-datagram EOF is `TruncatedDatagram` (fail-loud).
async fn read_one_datagram<R: AsyncRead + Unpin>(
reader: &mut DatagramReader,
@@ -593,7 +593,7 @@ async fn read_one_datagram<R: AsyncRead + Unpin>(
/// Call the producer's open op on channel 0 and extract the allocated
/// `channel_id` (the reverse path's first step; the POC's shape). The
/// optional `auth_token` rides the payload — the hub-forwarding path
/// optional `auth_token` rides the payload, the hub-forwarding path
/// (precedence: token > `ServingConfig.identity` > transport identity,
/// CF-005). The assembly layer then constructs the session with
/// [`TunnelSession::adopt`].
src/error.rs
+10 -9
View File
@@ -20,7 +20,7 @@ pub enum TunnelError {
/// An I/O failure on the session's data plane.
#[error("io: {0}")]
Io(#[from] std::io::Error),
/// A reply field failed its typed projection — the wire carried a
/// A reply field failed its typed projection: the wire carried a
/// malformed shape (e.g. [`crate::params::parse_listen_reply`]
/// found `bound` not substrate-shaped, a non-positive `listen`,
/// or a reply with no `channel_id` at all). A misbehaving
@@ -30,7 +30,7 @@ pub enum TunnelError {
}
/// A failed tunnel open or adopt. A failure never yields a session
/// (no phantom session, mirroring the no-phantom-channel property —
/// (no phantom session, mirroring the no-phantom-channel property of
/// ADR-005). Branch on [`ChannelOpenError::establishment_reason`]
/// through [`TunnelOpenError::open_ref`] for `channel:open_failed`'s
/// reason (`dial_failed` / `unknown_resource` / `resource_shortage` /
@@ -38,7 +38,7 @@ pub enum TunnelError {
/// (`FORBIDDEN`, `channel:too_many_channels`).
#[derive(Debug, thiserror::Error)]
pub enum TunnelOpenError {
/// The open op failed, or the reply/adopt failed — the typed
/// The open op failed, or the reply/adopt failed. The typed
/// [`ChannelOpenError`] carries the wire [`CallError`] verbatim
/// (adopt failures surface as `AdoptFailed`).
#[error("tunnel open failed: {0}")]
@@ -61,11 +61,12 @@ pub enum TunnelIoError {
/// The underlying stream failed.
#[error("io: {0}")]
Io(#[from] std::io::Error),
/// The datagram codec rejected the operation (e.g. `Oversize` —
/// The datagram codec rejected the operation (e.g. `Oversize`:
/// a >65535-byte send, rejected at frame time per ADR-003).
#[error("codec: {0}")]
Codec(#[from] DatagramCodecError),
/// The stream ended mid-datagram — declared length not satisfied
/// The stream ended mid-datagram; the declared length was not
/// satisfied
/// (a clean peer never does this; the fail-loud posture, OQ-TN-13
/// / ADR-003: never a silent partial datagram).
#[error("stream ended mid-datagram (truncated)")]
@@ -74,20 +75,20 @@ pub enum TunnelIoError {
/// plane (e.g. `send_datagram` on a stream tunnel).
#[error("wrong substrate for this operation")]
WrongSubstrate,
/// The session's channel halves were already taken — by
/// The session's channel halves were already taken: by
/// `take_halves` (caller-driven) or `pump_against` (pump-owned);
/// no data plane remains on the session.
#[error("channel halves already taken from this session")]
ChannelTaken,
}
/// The reverse-path open call failed (`open_reverse_channel` — the
/// The reverse-path open call failed (`open_reverse_channel`; the
/// assembly layer calls it before [`crate::consumer::TunnelSession::
/// adopt`]; the hub's call surface is a `CallConnection`, so the
/// adopt`], and the hub's call surface is a `CallConnection`, so the
/// two-step is the honest API).
#[derive(Debug, thiserror::Error)]
pub enum ReverseOpenError {
/// The open op failed — the wire [`CallError`]; branch on
/// The open op failed (the wire [`CallError`]); branch on
/// `details.reason` of `channel:open_failed` (ADR-049 §4).
#[error("open call failed: {0:?}")]
Call(CallError),
src/lib.rs
+3 -3
View File
@@ -1,4 +1,4 @@
//! alktunnels — arbitrary bidirectional tunnels over alkcall channels.
//! alktunnels: arbitrary bidirectional tunnels over alkcall channels.
//!
//! TCP, UDP, unix-socket, and other stream/datagram substrates, in the
//! `ssh -L` / `ssh -D` / `ssh -R` sense. A producer/consumer protocol
@@ -6,14 +6,14 @@
//! is the sibling precedent): the producer half registers the
//! `alk/tunnel` open op and pumps bytes between the channel and the
//! substrate; the consumer half is the typed session
//! ([`TunnelSession`]) that opens tunnel channels and owns teardown —
//! ([`TunnelSession`]) that opens tunnel channels and owns teardown,
//! plus [`TunnelListener`] (the listen-flow session, ADR-008) for the
//! bind-first listen shape.
//!
//! A tunnel is a resource, not an address: the open-op params identify
//! a produced resource + substrate ([`TunnelParams`], ADR-001); the
//! producer owns the backing. The protocol layer is substrate-agnostic
//! by construction — socket/platform I/O is confined to the
//! by construction: socket/platform I/O is confined to the
//! feature-gated `local` backend (the default crate is wasm-clean).
//!
//! - Wire format: `docs/architecture/wire.md` (+ BAST at `bast.md`)
src/local/mod.rs
+60 -60
View File
@@ -1,39 +1,39 @@
//! The `local` feature module: real-socket halves functions the
//! assembly layer injects as `DialFn`/`AcceptFn` closures (ADR-004 —
//! assembly layer injects as `DialFn`/`AcceptFn` closures (ADR-004:
//! "produce boxed halves for a resource" is a function, not a trait).
//! Non-wasm by design: this module IS the platform boundary — the
//! Non-wasm by design: this module IS the platform boundary; the
//! default crate stays wasm-clean (convention 16: `local`-gated code
//! is never imported from the shared/producer/consumer modules).
//!
//! Substrate coverage (overview.md §Feature Gates):
//! - **TCP** — [`dial_tcp`] (5s connect timeout per the POCs) and the
//! - **TCP**: [`dial_tcp`] (5s connect timeout per the POCs) and the
//! [`TcpListenerHalves`] accept half for listen producers.
//! - **UDP** — [`connect_udp`]: a connected `UdpSocket` wrapped in the
//! FRAMED adapter ([`UdpHalf`], ADR-003 — the codec composes at this
//! - **UDP**: [`connect_udp`]: a connected `UdpSocket` wrapped in the
//! FRAMED adapter ([`UdpHalf`], ADR-003; the codec composes at this
//! boundary: the channel side speaks the `[len: u16 BE]` frames, the
//! target side sees payload-only datagrams; the pump never sees UDP
//! specifics). Truncation fails loud (OQ-TN-13's resolved posture).
//! - **Unix** — [`dial_unix`], in v1 per OQ-TN-14 (same halves shape
//! as TCP). **Stdio is NOT here** — a spawned process's stdio is
//! - **Unix**: [`dial_unix`], in v1 per OQ-TN-14 (same halves shape
//! as TCP). **Stdio is NOT here**: a spawned process's stdio is
//! alktty's pipe mode (exit codes + signals are tty semantics);
//! remote command execution composes via alktty, not here.
//! - **Bind-first** (shape 3, ADR-008) — [`bind_tcp_first`] /
//! - **Bind-first** (shape 3, ADR-008): [`bind_tcp_first`] /
//! [`bind_unix_first`] (the real-socket [`BindFn`] sources: the
//! `bound` reply field's data — the REAL OS-chosen address, port 0
//! `bound` reply field's data, the REAL OS-chosen address, port 0
//! resolved), [`TcpListenerHalves::accept_pair`] /
//! [`UnixListenerHalves::accept_pair`] (one accept per call yielding
//! `(peer, halves)` — the [`BindFirstAcceptLoop`] contract), and the
//! `(peer, halves)`; the [`BindFirstAcceptLoop`] contract), and the
//! assembly convenience closures [`local_bind_first()`] +
//! [`local_bindfirst_accept_loop()`]. UDP bind-first is NOT here —
//! [`local_bindfirst_accept_loop()`]. UDP bind-first is NOT here:
//! a UDP socket has no accept; the per-peer associate shape is a
//! separate design decision (the bind closure fails loud for udp).
//!
//! Error mapping: [`TunnelEstablishError`] variants (the POC's
//! `Into<EstablishmentError>` path) — dial refusal → `dial_failed`,
//! `Into<EstablishmentError>` path); dial refusal → `dial_failed`,
//! bind/resource exhaustion → `resource_shortage`, handler-internal
//! → `handler_error`.
//!
//! `TargetHandle` re-used from producer.rs — the halves type lives
//! `TargetHandle` re-used from producer.rs; the halves type lives
//! there; `local/` depends on it, never vice versa (convention 16).
use std::sync::Arc;
@@ -52,7 +52,7 @@ const DIAL_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
// --- TCP ----------------------------------------------------------------
/// Dial a TCP target (`host:port`): `TcpStream::connect` bounded by
/// [`DIAL_TIMEOUT`], `into_split` → boxed halves (raw pass-through —
/// [`DIAL_TIMEOUT`], `into_split` → boxed halves (raw pass-through;
/// the halves ARE the tunnel, ADR-003).
pub async fn dial_tcp(target: &str) -> Result<TargetHandle, TunnelEstablishError> {
let stream = tokio::time::timeout(DIAL_TIMEOUT, tokio::net::TcpStream::connect(target))
@@ -67,11 +67,11 @@ pub async fn dial_tcp(target: &str) -> Result<TargetHandle, TunnelEstablishError
}
/// Bind a TCP listener (the listen shape's assembly-side half).
/// The binding decision belongs to the caller (OQ-TN-04) — this is a
/// The binding decision belongs to the caller (OQ-TN-04); this is a
/// `local`-feature convenience, not protocol behavior.
///
/// **Bind posture (review 001 U-2):** the caller-supplied address is
/// deliberate — a wildcard bind (`0.0.0.0` or `::`; note `::1` is
/// deliberate: a wildcard bind (`0.0.0.0` or `::`; note `::1` is
/// loopback, not a wildcard) is the caller's explicit choice and
/// exposes the listener per the host's firewall. There is no implicit
/// wildcard default here.
@@ -102,7 +102,7 @@ impl TcpListenerHalves {
/// The accept loop: accept connections until the listener fails,
/// pushing each accepted handle into the queue. Returns when the
/// listener is closed (accept errors end the loop — the assembly
/// listener is closed (accept errors end the loop; the assembly
/// layer aborts this task on teardown).
pub async fn accept_loop(&self, queue: AcceptQueue) -> Result<(), TunnelEstablishError> {
loop {
@@ -124,16 +124,16 @@ impl TcpListenerHalves {
}
}
/// ONE accept as a `(peer, halves)` pair — the BIND-FIRST shape's
/// ONE accept as a `(peer, halves)` pair, the BIND-FIRST shape's
/// substrate half (ADR-008 §1): the accept dispatcher invokes the
/// injected [`BindFirstAcceptLoop`] closure PER accept; this method
/// is what the closure calls. The peer address rides `accept()`'s
/// return value (the forwarded op's `peer` carrier — reply#2-class
/// fidelity, ADR-008 §3).
/// return value (the forwarded op's `peer` carrier,
/// reply#2-class fidelity, ADR-008 §3).
///
/// This is deliberately NOT the [`AcceptQueue`]-feeding
/// [`TcpListenerHalves::accept_loop`] (the accept-FIRST shape's
/// contract — the establisher pops pre-accepted handles): the
/// contract, the establisher pops pre-accepted handles): the
/// bind-first dispatcher owns the accept pacing, so one accept per
/// closure call with the peer attached is the honest shape.
pub async fn accept_pair(&self) -> Result<(SubstrateAddr, TargetHandle), TunnelEstablishError> {
@@ -160,16 +160,16 @@ impl TcpListenerHalves {
/// Connect a UDP "tunnel" to a target: resolve the target, bind an
/// ephemeral socket on the MATCHING address family (never an implicit
/// wildcard mismatch — a v6-only target dials from a v6-bound socket;
/// wildcard mismatch: a v6-only target dials from a v6-bound socket;
/// review 001 U-2's posture), `connect(target)` (the
/// associated-socket posture — one peer), and wrap it in the FRAMED
/// associated-socket posture, one peer), and wrap it in the FRAMED
/// adapter (ADR-003): `UdpHalf` composes the `[len: u16 BE]` codec at
/// this boundary — the channel side speaks framed bytes, the target
/// this boundary: the channel side speaks framed bytes, the target
/// side sees payload-only datagrams; the pump never sees UDP
/// specifics.
pub async fn connect_udp(target: &str) -> Result<TargetHandle, TunnelEstablishError> {
use std::net::ToSocketAddrs;
// Same posture as tokio's own `UdpSocket::connect(target: &str)` —
// Same posture as tokio's own `UdpSocket::connect(target: &str)`;
// it resolves inline too; numeric targets (the registry's shape)
// make this a non-blocking parse.
let resolved = target
@@ -200,9 +200,9 @@ pub async fn connect_udp(target: &str) -> Result<TargetHandle, TunnelEstablishEr
/// The FRAMED UDP adapter (ADR-003's substrate-boundary codec): the
/// pump-facing face speaks the tunnel's `[len: u16 BE][payload]`
/// channel framing; the socket face speaks whole raw datagrams.
/// Orientation (review 001 §Re-validation — the fix sketch's caution):
/// Orientation (review 001 §Re-validation, the fix sketch's caution):
/// `UdpHalf`'s caller is the pump, and the CHANNEL side is the framed
/// side —
/// side:
///
/// - `poll_write` is fed CHANNEL bytes (already-framed frames, from
/// the session's codec) → it **de-frames** (an incremental
@@ -210,13 +210,13 @@ pub async fn connect_udp(target: &str) -> Result<TargetHandle, TunnelEstablishEr
/// as one raw datagram. A frame may straddle successive poll_write
/// buffers (the pump's 8 KiB copy chunks split large frames); the
/// decoder carries the partial state until the frame completes.
/// `len=0` decodes to a zero-payload datagram — a real empty
/// `len=0` decodes to a zero-payload datagram: a real empty
/// datagram, NOT EOF (the F-2 fix end-to-end).
/// - `poll_read` yields bytes destined for the CHANNEL (the session's
/// codec decodes there) → it **frames** ([`frame_datagram`]) each
/// raw target datagram and hands the frame bytes to the pump. A
/// frame larger than the caller's buffer is staged and handed out
/// across successive reads — a partial frame is never misread as a
/// across successive reads; a partial frame is never misread as a
/// shorter one.
///
/// The target sees payload-only bytes; an empty datagram crosses the
@@ -224,10 +224,10 @@ pub async fn connect_udp(target: &str) -> Result<TargetHandle, TunnelEstablishEr
///
/// **Truncation never happens silently** (OQ-TN-13's fail-loud
/// posture, ADR-003): the socket-facing recv uses a full 65535-byte
/// scratch (`try_recv` truncates at the buffer size — a smaller
/// scratch (`try_recv` truncates at the buffer size; a smaller
/// scratch would corrupt the framing before any size check runs), and
/// a datagram beyond [`MAX_DATAGRAM_LEN`] (impossible for a
/// well-formed peer — the u16 frame bound) errors loudly instead of
/// well-formed peer, the u16 frame bound) errors loudly instead of
/// truncating.
///
/// The scratch is allocated lazily on the first recv (review 001 U-3:
@@ -237,7 +237,7 @@ pub struct UdpHalf {
sock: Arc<tokio::net::UdpSocket>,
kind: UdpHalfKind,
/// Read-half state: the recv scratch (lazily grown to the full
/// 65535 on first use — `try_recv` truncates silently at the
/// 65535 on first use; `try_recv` truncates silently at the
/// buffer size, so the full-size read is correctness-load-bearing)
/// plus the frame-staging state for handing a >caller-buffer
/// frame out across reads.
@@ -270,13 +270,13 @@ struct UdpReadStage {
struct UdpWriteDeFrame {
reader: DatagramReader,
/// Decoded datagrams awaiting send (FIFO). poll_write accepts all
/// offered bytes into the de-framer — the pump's copy window
/// offered bytes into the de-framer; the pump's copy window
/// cannot be held closed for later top-up (a full window is
/// re-offered unsatisfied until drained); backpressure is the
/// accepted-byte high-water mark instead.
queue: std::collections::VecDeque<bytes::Bytes>,
/// The datagram mid-send (the socket returned WouldBlock under
/// it — retried first on every poll).
/// it, retried first on every poll).
sending: Option<bytes::Bytes>,
/// Accepted-byte accounting for the high-water mark (the bytes
/// accepted but not yet sent on the socket).
@@ -341,14 +341,14 @@ impl AsyncRead for UdpHalf {
// Hand out staged frame bytes first (a frame larger than the
// caller's buffer was split across reads; the frame's length
// prefix already crossed, so the pump's de-framer on the far
// side reassembles — no shorter-datagram misread is possible).
// side reassembles; no shorter-datagram misread is possible).
if !this.stage.pending.is_empty() {
let n = this.stage.pending.len().min(buf.remaining());
buf.put_slice(&this.stage.pending.split_to(n));
return Poll::Ready(Ok(()));
}
// Recv one raw datagram (full-size scratch — try_recv
// Recv one raw datagram (full-size scratch; try_recv
// truncates silently at the buffer size, so the scratch must
// hold the largest possible datagram before any size check).
if this.stage.recv_scratch.is_none() {
@@ -370,7 +370,7 @@ impl AsyncRead for UdpHalf {
std::io::ErrorKind::InvalidData,
format!(
"udp datagram {len} bytes exceeds the codec frame bound {max} \
(truncation fails loud — OQ-TN-13)"
(truncation fails loud; OQ-TN-13)"
),
)
},
@@ -380,7 +380,7 @@ impl AsyncRead for UdpHalf {
} else {
// The frame exceeds the caller's buffer: hand out
// the first slice, stage the rest (the length
// prefix rode along — the far-side de-framer
// prefix rode along; the far-side de-framer
// reassembles across reads).
let n = buf.remaining();
buf.put_slice(&framed[..n]);
@@ -416,7 +416,7 @@ impl tokio::io::AsyncWrite for UdpHalf {
// Backpressure: if too many accepted bytes are still queued
// for the socket, try to drain first; if the queue is still
// over the line, return Pending (the pump's copy re-offers
// its window unsatisfied — safe, nothing is lost).
// its window unsatisfied; safe, nothing is lost).
if this.deframe.accepted > WRITE_QUEUE_HIGH_WATER {
match this.pump_queue_to_socket(cx) {
Poll::Ready(Ok(_)) => {}
@@ -432,15 +432,15 @@ impl tokio::io::AsyncWrite for UdpHalf {
// Accept ALL offered bytes into the incremental de-framer and
// report them all consumed. The pump's copy window cannot be
// held closed for a later top-up (a full window is re-offered
// unsatisfied until drained — a partial-frame tail could sit
// unsatisfied until drained; a partial-frame tail could sit
// forever in a full window), and reporting fewer than all
// bytes risks the copy's `WriteZero` on an empty re-offer.
// The DatagramReader's partial-frame state is bounded by
// construction (the u16 cap — review 001 N-2), so accepting
// construction (the u16 cap, review 001 N-2), so accepting
// everything is bounded memory; socket-side backpressure is
// the accepted-byte high-water mark above.
// The de-framer's feed is infallible (the oversize bound is
// enforced at frame time — see the DatagramCodecError docs):
// enforced at frame time; see the DatagramCodecError docs):
// accept everything, bounded by the u16 cap (review 001 N-2).
for dg in this.deframe.reader.feed(buf) {
this.deframe.accepted += dg.len() + 2;
@@ -462,7 +462,7 @@ impl tokio::io::AsyncWrite for UdpHalf {
) -> std::task::Poll<std::io::Result<()>> {
// The send pump: tokio's copy polls flush after draining its
// window, so the queued datagrams drain here whenever the
// pump cycles — and a flush-resolved Ready(Ok(())) tells the
// pump cycles, and a flush-resolved Ready(Ok(())) tells the
// copy the bytes are on the wire. UDP sends are all-or-nothing
// per datagram; a WouldBlock keeps the datagram queued (the
// send waker is registered) and the queue persists until it
@@ -507,7 +507,7 @@ impl UdpHalf {
self.deframe.accepted -= dg.len() + 2;
}
// WouldBlock (EAGAIN) and ENOBUFS both mean "the socket
// cannot take this datagram now" — the unresolved-
// cannot take this datagram now": the unresolved-
// neighbor + shrunken-SNDBUF state reports ENOBUFS(105);
// the stash + waker retry is the same behavior for both.
Err(e)
@@ -530,7 +530,7 @@ impl UdpHalf {
// --- Unix ---------------------------------------------------------------
/// Dial a unix-socket target (OQ-TN-14 — ships in v1, the same halves
/// Dial a unix-socket target (OQ-TN-14, ships in v1, the same halves
/// shape as TCP): `UnixStream::connect` bounded by [`DIAL_TIMEOUT`],
/// `into_split` → boxed halves (raw pass-through, ADR-003).
pub async fn dial_unix(path: &str) -> Result<TargetHandle, TunnelEstablishError> {
@@ -546,10 +546,10 @@ pub async fn dial_unix(path: &str) -> Result<TargetHandle, TunnelEstablishError>
}
/// Bind a unix-socket listener (the listen shape's assembly-side half,
/// OQ-TN-14's dial parity — same wrapper shape as
/// OQ-TN-14's dial parity; same wrapper shape as
/// [`TcpListenerHalves`]). The binding decision belongs to the caller
/// (OQ-TN-04); a stale socket file at `path` fails loud (the
/// caller unlinks first — `bind` never silently reuses).
/// caller unlinks first; `bind` never silently reuses).
pub fn bind_unix(path: &str) -> Result<UnixListenerHalves, TunnelEstablishError> {
let listener = tokio::net::UnixListener::bind(path)
.map_err(|e| TunnelEstablishError::ResourceShortage(format!("bind `{path}`: {e}")))?;
@@ -562,7 +562,7 @@ pub fn bind_unix(path: &str) -> Result<UnixListenerHalves, TunnelEstablishError>
/// accept loop ([`UnixListenerHalves::accept_loop`]) is the
/// accept-FIRST shape's feed ([`AcceptQueue`]); the per-accept
/// [`UnixListenerHalves::accept_pair`] is the bind-FIRST shape's
/// (one accept = one `(peer, halves)` pair — the peer is the
/// (one accept = one `(peer, halves)` pair; the peer is the
/// connecting path, `accept()`'s return value).
pub struct UnixListenerHalves {
listener: Arc<tokio::net::UnixListener>,
@@ -586,7 +586,7 @@ impl UnixListenerHalves {
/// The accept-FIRST shape's loop: accept until the listener fails,
/// pushing each accepted handle into the queue (the assembly layer
/// aborts this task on teardown). Unix peer addresses are not
/// reliably readable from an accepted stream — the queue contract
/// reliably readable from an accepted stream; the queue contract
/// carries halves only, same as the TCP shape's loop.
pub async fn accept_loop(&self, queue: AcceptQueue) -> Result<(), TunnelEstablishError> {
loop {
@@ -608,11 +608,11 @@ impl UnixListenerHalves {
}
}
/// ONE accept as a `(peer, halves)` pair — the BIND-FIRST shape's
/// ONE accept as a `(peer, halves)` pair: the BIND-FIRST shape's
/// substrate half for unix (OQ-TN-14 parity: the bind path is the
/// same shape as [`TcpListenerHalves::accept_pair`]). The peer is
/// the ACCEPTED side's bound path ([`UnixStream::peer_addr`]);
/// an unnamed peer (an autobind client) fails loud — the
/// an unnamed peer (an autobind client) fails loud; the
/// forwarded op's `peer` param is required on the wire, so an
/// addressless peer cannot be represented (ADR-008 §3).
pub async fn accept_pair(&self) -> Result<(SubstrateAddr, TargetHandle), TunnelEstablishError> {
@@ -645,9 +645,9 @@ impl UnixListenerHalves {
/// Bind a unix-socket listener for the BIND-FIRST shape (the
/// real-socket [`BindFn`] source, OQ-TN-14's dial/bind parity): the
/// returned [`BoundListener::bound`] is the REAL bound socket path
/// (the `bound` reply field's data — reply#1-class fidelity; the
/// (the `bound` reply field's data, reply#1-class fidelity; the
/// forwarded op's `peer` shape matches). The opaque `listener` half is
/// the [`UnixListenerHalves`] wrapper — the assembly's
/// the [`UnixListenerHalves`] wrapper; the assembly's
/// [`BindFirstAcceptLoop`] closure downcasts to it and calls
/// [`UnixListenerHalves::accept_pair`] per accept.
pub async fn bind_unix_first(path: &str) -> Result<BoundListener, TunnelEstablishError> {
@@ -662,7 +662,7 @@ pub async fn bind_unix_first(path: &str) -> Result<BoundListener, TunnelEstablis
}
/// The local dial closure (the ADR-004 injection point): one
/// [`DialFn`]-shaped function covering tcp/udp/unix — the assembly
/// [`DialFn`]-shaped function covering tcp/udp/unix; the assembly
/// layer injects this (closed over nothing substrate-specific beyond
/// this module).
pub fn local_dial() -> DialFn {
@@ -681,13 +681,13 @@ pub fn local_dial() -> DialFn {
// --- Bind-first (shape 3, ADR-008 §1) ------------------------------------
/// Bind a TCP listener for the BIND-FIRST shape (the real-socket
/// [`BindFn`] source — shape 3's backing, ADR-008 §1): `port 0` = the
/// [`BindFn`] source; shape 3's backing, ADR-008 §1): `port 0` = the
/// OS chooses, and the returned [`BoundListener::bound`] is the REAL
/// resolved address (`local_addr()` — the `bound` reply field's data,
/// resolved address (`local_addr()`: the `bound` reply field's data,
/// reply#1-class fidelity). The caller-supplied address is deliberate
/// ([`bind_tcp`]'s wildcard-bind posture applies unchanged).
///
/// The opaque `listener` half is the [`TcpListenerHalves`] wrapper —
/// The opaque `listener` half is the [`TcpListenerHalves`] wrapper;
/// the assembly's [`BindFirstAcceptLoop`] closure downcasts to it and
/// calls [`TcpListenerHalves::accept_pair`] per accept.
pub async fn bind_tcp_first(addr: &str) -> Result<BoundListener, TunnelEstablishError> {
@@ -704,7 +704,7 @@ pub async fn bind_tcp_first(addr: &str) -> Result<BoundListener, TunnelEstablish
/// The TCP bind closure + accept-pair loop, bundled (the assembly
/// convenience): [`local_bind_first()`] renders the [`BindFn`] shape-3
/// expects (the dispatch substrate picks tcp/unix; udp fails loud —
/// expects (the dispatch substrate picks tcp/unix; udp fails loud:
/// a UDP socket has no accept, so bind-first is not defined there),
/// [`local_bindfirst_accept_loop()`] renders the [`BindFirstAcceptLoop`]
/// that downcasts the plan's opaque listener to [`TcpListenerHalves`] /
@@ -730,12 +730,12 @@ pub fn local_bind_first() -> BindFn {
/// [`local_bind_first()`]): downcasts the plan's opaque listener to
/// the concrete wrapper the bind closure produced and accepts ONE
/// connection as a `(peer, halves)` pair per invocation (the
/// dispatcher's per-accept contract — one closure call = one accept;
/// dispatcher's per-accept contract: one closure call = one accept;
/// the loop lives in the protocol's dispatcher).
///
/// The downcast is the assembly side of the type-erased injection
/// pattern: the protocol crate never knows the listener's type, so the
/// closure must — a listener not produced by this module's bind
/// closure must; a listener not produced by this module's bind
/// closures (or a mismatched substrate/backing pairing in the
/// assembly's registry) fails loud as `handler_error`.
pub fn local_bindfirst_accept_loop() -> BindFirstAcceptLoop {
src/params.rs
+50 -50
View File
@@ -1,18 +1,18 @@
//! Open-op params: `TunnelParams {resource, substrate}` — the
//! Open-op params: `TunnelParams {resource, substrate}`, the
//! wire-stable carrier of "what to open" (ADR-001), plus the open-op
//! `OperationSpec` builder, the scope/ALPN/op-id constants, and the
//! typed establishment-reason helper (ADR-049 §4 surface).
//!
//! The direct op's params ([`TunnelDirectParams`] + [`SubstrateAddr`],
//! ADR-007) and its spec builder live here too: `{substrate, target}`
//! — the caller names the dial target at session time (the ssh
//! ADR-007) and its spec builder live here too: `{substrate, target}`;
//! the caller names the dial target at session time (the ssh
//! `direct-tcpip` / SOCKS5 CONNECT shape), scope-gated by
//! `tunnel:direct` (a separate grant from `tunnel:open`, never implied
//! by it).
//!
//! The forwarded op's params ([`TunnelForwardedParams`], ADR-008 §3
//! as amended) and its spec builder live here as well:
//! `{listen, substrate, peer}` — the accept-as-open ingress op, keyed
//! `{listen, substrate, peer}`, the accept-as-open ingress op, keyed
//! by the producer-minted correlation token (Amendment 1), scope-gated
//! by `tunnel:forwarded` (a separate grant again). The bind-first
//! listen open's reply projection ([`TunnelListenReply`] +
@@ -21,11 +21,11 @@
//! contributes.
//!
//! The params layout is a one-way door once a consumer exists
//! (ADR-001): `{resource, substrate}` only — the producer owns the
//! (ADR-001): `{resource, substrate}` only; the producer owns the
//! backing; no URL-style addressing. `deny_unknown_fields` is the
//! loud-rejection posture (unknown fields fail, never silently
//! ignored); an older producer rejecting a newer substrate value is
//! the SSH "unknown channel type" posture — loud, not silent.
//! the SSH "unknown channel type" posture: loud, not silent.
pub use alkcall::channels::client::ChannelOpenError;
use alkcall::registry::spec::{
@@ -46,25 +46,25 @@ pub const TUNNEL_ALPN: &str = "alk/tunnel";
/// The scope gating tunnel opens (ADR-006). Stable once published.
pub const TUNNEL_OPEN_SCOPE: &str = "tunnel:open";
/// The `channels/tunnel/direct` operation id (the direct op —
/// The `channels/tunnel/direct` operation id (the direct op:
/// dynamic-target egress, ADR-007; the flavor-form op-id convention,
/// ADR-002 Amendment 1).
pub const OP_TUNNEL_DIRECT: &str = "channels/tunnel/direct";
/// The scope gating the direct op (ADR-007 §3 / ADR-006 Amendment 1).
/// A separate grant from [`TUNNEL_OPEN_SCOPE`] — **never implied by
/// A separate grant from [`TUNNEL_OPEN_SCOPE`], **never implied by
/// it**: "dial `postgres-primary`" and "dial anything from my
/// network" are different capabilities. Stable once published.
pub const TUNNEL_DIRECT_SCOPE: &str = "tunnel:direct";
/// The `channels/tunnel/forwarded` operation id (the forwarded op —
/// The `channels/tunnel/forwarded` operation id (the forwarded op:
/// accept-as-open ingress, ADR-008 §3; the flavor-form op-id
/// convention, ADR-002 Amendment 1).
pub const OP_TUNNEL_FORWARDED: &str = "channels/tunnel/forwarded";
/// The scope gating the forwarded op (ADR-008 §3 / ADR-006 Amendment
/// 1). A separate grant from [`TUNNEL_OPEN_SCOPE`] and
/// [`TUNNEL_DIRECT_SCOPE`] — **never implied by either**: receiving
/// [`TUNNEL_DIRECT_SCOPE`], **never implied by either**: receiving
/// producer-initiated forwarded connections (ingress) is a different
/// capability from dialing egress. Stable once published.
pub const TUNNEL_FORWARDED_SCOPE: &str = "tunnel:forwarded";
@@ -75,7 +75,7 @@ pub const TUNNEL_FORWARDED_SCOPE: &str = "tunnel:forwarded";
#[serde(deny_unknown_fields)]
pub struct TunnelParams {
/// The produced resource identifier: the producer's stable name
/// for the tunnel target — NOT an address (the producer's
/// for the tunnel target, NOT an address (the producer's
/// registry maps it to its backing, OQ-TN-11).
pub resource: String,
/// The extensible substrate discriminator; selects the data-plane
@@ -84,7 +84,7 @@ pub struct TunnelParams {
}
/// The substrate discriminator (ADR-001). `Unix` ships in v1 (the
/// `local` feature implements it — OQ-TN-14); a new substrate is a
/// `local` feature implements it; OQ-TN-14); a new substrate is a
/// new value, not a format change.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
@@ -95,12 +95,12 @@ pub enum Substrate {
}
/// The substrate-shaped address object (ADR-007 §2): keyed by the
/// sibling `substrate` field of the params — `tcp`/`udp` →
/// sibling `substrate` field of the params: `tcp`/`udp` →
/// [`SubstrateAddr::HostPort`] (domain-form hosts stay unresolved on
/// the wire — producer-side resolution, the ssh `-D` semantic),
/// the wire: producer-side resolution, the ssh `-D` semantic),
/// `unix` → [`SubstrateAddr::Path`]. Untagged: the shapes are
/// mutually exclusive in practice and self-describing. Also the
/// `peer` value shape of the forwarded op (ADR-008 §3) — defined
/// `peer` value shape of the forwarded op (ADR-008 §3), defined
/// once here.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(untagged)]
@@ -114,7 +114,7 @@ pub enum SubstrateAddr {
/// shape-vs-substrate correspondence is pinned by the spec's
/// `input_schema` (per-substrate `oneOf`, validated registry-side
/// before the establisher) and re-checked by serde here (the
/// backstop — a mismatch parses as the wrong variant or fails).
/// backstop: a mismatch parses as the wrong variant or fails).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct TunnelDirectParams {
@@ -128,7 +128,7 @@ pub struct TunnelDirectParams {
impl TunnelDirectParams {
/// Render the target to the backing-string form the `DialFn` /
/// registry backings use: `"host:port"` for tcp/udp, the path for
/// unix. The direct establisher calls this before dialing — the
/// unix. The direct establisher calls this before dialing; the
/// dial closure stays substrate-agnostic (ADR-007 §4).
pub fn render_target(&self) -> String {
match &self.target {
@@ -179,11 +179,11 @@ pub fn tunnel_open_spec() -> OperationSpec {
/// The direct-op spec (`channels/tunnel/direct`, ADR-007): `Sub`-typed,
/// external, channel-open marker on the `alk/tunnel` ALPN, scope-gated
/// by `tunnel:direct` (a separate grant from `tunnel:open` — never
/// by `tunnel:direct` (a separate grant from `tunnel:open`, never
/// implied by it, ADR-006 Amendment 1), input/output schemas per
/// wire.md §The Direct Op. The target sub-schema is per-substrate —
/// wire.md §The Direct Op. The target sub-schema is per-substrate:
/// `if`/`then` conditioned on the sibling `substrate` value
/// (`tcp`/`udp` → `{host, port}`, `unix` → `{path}`) — so a malformed
/// (`tcp`/`udp` → `{host, port}`, `unix` → `{path}`), so a malformed
/// target fails the registry's schema validation (the typed
/// `INVALID_INPUT` class) before the establisher runs.
pub fn tunnel_direct_spec() -> OperationSpec {
@@ -272,7 +272,7 @@ pub fn tunnel_direct_spec() -> OperationSpec {
/// The reply arrives as `{channel_id, bound?, listen?}` (the
/// wrapper-reserved `channel_id` key plus the establisher's additive
/// fields); unknown extra reply fields must be tolerated (the additive
/// posture — old consumers ignore unknown fields). Parse the full
/// posture: old consumers ignore unknown fields). Parse the full
/// reply with [`parse_listen_reply`], never with `deny_unknown_fields`
/// deserialization of the whole reply. Both fields are `None` for
/// accept-first/dial opens (the establisher supplied nothing).
@@ -280,33 +280,33 @@ pub fn tunnel_direct_spec() -> OperationSpec {
#[serde(deny_unknown_fields)]
pub struct TunnelListenReply {
/// The real OS-chosen bound address (port `0` resolved),
/// substrate-shaped — present only when the establisher supplied
/// substrate-shaped, present only when the establisher supplied
/// it. Informational surface, NOT addressing: a consumer MUST NOT
/// treat `bound` as a dial target (the producer owns the backing —
/// treat `bound` as a dial target (the producer owns the backing,
/// ADR-001's invariant); it exists for RFC-1928-reply-class
/// fidelity (BND.ADDR) and observability.
#[serde(skip_serializing_if = "Option::is_none")]
pub bound: Option<SubstrateAddr>,
/// The correlation token (ADR-008 Amendment 1): the u32 the
/// establisher minted — the key the forwarded op's `listen` param
/// establisher minted, the key the forwarded op's `listen` param
/// carries. Present only when the establisher supplied it.
#[serde(skip_serializing_if = "Option::is_none")]
pub listen: Option<u32>,
}
/// Project the bind-first listen open's reply (ADR-008 §2 as amended)
/// out of the FULL success output [`Value`] — the shape
/// out of the FULL success output [`Value`], the shape
/// `ChannelClient::open_channel_with_reply` returns. Extracts the
/// optional `bound` + `listen` fields into [`TunnelListenReply`].
///
/// - Unknown extra reply fields are **tolerated** — the reply viewer
/// - Unknown extra reply fields are **tolerated**; the reply viewer
/// must not deny unknown fields (the additive posture; the wire is
/// plain JSON reply fields, and the projection is a typed view).
/// - `channel_id` must be present and a positive integer (the
/// wrapper-owned key — its absence means the caller is not looking
/// wrapper-owned key: its absence means the caller is not looking
/// at an open reply at all; fail loud).
/// - A present-but-malformed `bound` (not substrate-shaped) or
/// `listen` (not a positive u32) is a loud error — a misbehaving
/// `listen` (not a positive u32) is a loud error: a misbehaving
/// producer is never silently coerced ([`TunnelError::
/// MalformedReply`]).
pub fn parse_listen_reply(reply: &Value) -> Result<TunnelListenReply, TunnelError> {
@@ -341,7 +341,7 @@ pub fn parse_listen_reply(reply: &Value) -> Result<TunnelListenReply, TunnelErro
}
/// Extract the bind-first listen open's reply fields (ADR-008 §2 as
/// amended) from an open-op success reply — the **lenient, infallible
/// amended) from an open-op success reply, the **lenient, infallible
/// projection** the typed consumer surface reads: both fields
/// [`Option`], absent for dial/accept-first producers, unknown extra
/// reply fields tolerated (the additive posture).
@@ -350,9 +350,9 @@ pub fn parse_listen_reply(reply: &Value) -> Result<TunnelListenReply, TunnelErro
/// projects a malformed reply to both-`None`: the session type
/// presents `Option` fields and never refuses construction over an
/// informational field (`bound`/`listen` are informational surface,
/// NOT addressing — ADR-008 §2's invariant; leniency here is safe by
/// NOT addressing, per ADR-008 §2's invariant; leniency here is safe by
/// construction). A malformed shape is logged loud (`tracing::error!`)
/// and surfaces as `None` — the session's caller cannot distinguish
/// and surfaces as `None`; the session's caller cannot distinguish
/// "producer supplied nothing" from "producer misbehaved"; use
/// [`parse_listen_reply`] directly when that distinction matters (the
/// fail-loud posture for consumers that must reject misbehaving
@@ -377,13 +377,13 @@ pub fn listen_reply_fields(reply: &Value) -> TunnelListenReply {
/// The forwarded-op input params (ADR-008 §3 as amended). Wire-stable:
/// all three fields required; renames are a breaking wire change.
/// `deny_unknown_fields` is the loud-rejection posture (ADR-001's
/// precedent — unknown fields fail, never silently ignored).
/// precedent: unknown fields fail, never silently ignored).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct TunnelForwardedParams {
/// The correlation token (ADR-008 Amendment 1): the u32 the
/// bind-first establisher minted and returned in the listen
/// open's `listen` reply field — NOT the listen channel's ID
/// open's `listen` reply field, NOT the listen channel's ID
/// (that key is superseded: the serving-side handler never sees
/// its own channel ID, and generic relays don't rewrite params).
/// Positive by wire contract (schema `minimum: 1`; the serde pin
@@ -394,7 +394,7 @@ pub struct TunnelForwardedParams {
/// data-plane framing (ADR-003).
pub substrate: Substrate,
/// The accepted connection's remote address, keyed by the sibling
/// `substrate` field ([`SubstrateAddr`]). Required in v1 — a
/// `substrate` field ([`SubstrateAddr`]). Required in v1: a
/// producer that cannot determine the peer does not serve a
/// bind-first listener over that substrate (ADR-008 §3).
pub peer: SubstrateAddr,
@@ -403,7 +403,7 @@ pub struct TunnelForwardedParams {
/// The serde-level pin behind `listen`'s schema `minimum: 1`: the
/// correlation token is positive by contract (the establisher seeds
/// the mint counter non-zero), so a `listen: 0` on the wire is a
/// malformed open — rejected at parse time, never silently accepted.
/// malformed open, rejected at parse time, never silently accepted.
fn positive_u32<'de, D>(deserializer: D) -> Result<u32, D::Error>
where
D: serde::Deserializer<'de>,
@@ -417,20 +417,20 @@ where
/// The forwarded-op spec (`channels/tunnel/forwarded`, ADR-008 §3 as
/// amended): `Sub`-typed, external, channel-open marker on the
/// `alk/tunnel` ALPN, scope-gated by `tunnel:forwarded` (ingress — a
/// `alk/tunnel` ALPN, scope-gated by `tunnel:forwarded` (ingress: a
/// separate grant from `tunnel:open` and `tunnel:direct`, ADR-006
/// Amendment 1), input/output schemas per wire.md §The Forwarded Op.
/// All three params required; the `peer` sub-schema is per-substrate —
/// All three params required; the `peer` sub-schema is per-substrate:
/// `if`/`then` conditioned on the sibling `substrate` value (`tcp`/
/// `udp` → `{host, port}`, `unix` → `{path}`), so a malformed peer
/// fails the registry's schema validation (the typed
/// `INVALID_INPUT` class) before the opener-side handler runs.
///
/// **Serving registration** (ADR-008 §3 as amended by Amendment 2):
/// establisher-less base shape — `register_openable`, the data plane
/// establisher-less base shape: `register_openable`, the data plane
/// starts immediately (handler = correlate by `listen` + accept the
/// `BiStream` + pump; the correlation registry is assembly-side
/// bookkeeping) — or the recommended establisher variant,
/// bookkeeping), or the recommended establisher variant,
/// `register_tunnel_forwarded_openable_with_establisher` (the
/// injected correlation gate; a stale/missing token fails the open as
/// a typed `channel:open_failed`). The spec itself is unchanged: the
@@ -604,7 +604,7 @@ mod tests {
#[test]
fn serde_rejects_unknown_substrate_value() {
// The serde layer (the enum deserializer), not the JSON Schema —
// The serde layer (the enum deserializer), not the JSON Schema:
// the registry's real schema gate is pinned end-to-end by the
// integration probes (`schema_rejects_unknown_substrate_loudly`,
// `malformed_target_is_schema_class_and_leaves_no_channel`, the
@@ -709,7 +709,7 @@ mod tests {
});
assert!(serde_json::from_value::<TunnelDirectParams>(v).is_err());
// The untagged target is lenient about extra target fields
// (untagged variants carry no deny_unknown_fields) — a
// (untagged variants carry no deny_unknown_fields); a
// cross-shape target parses as whichever variant fits. The
// SCHEMA is the gate that rejects shape-vs-substrate
// mismatches registry-side (additionalProperties: false per
@@ -932,7 +932,7 @@ mod tests {
#[test]
fn listen_reply_bare_projection() {
// The accept-first/dial shape: the establisher contributed
// nothing — both fields None.
// nothing; both fields are None.
let reply = json!({"channel_id": 7});
let projected = parse_listen_reply(&reply).unwrap();
assert_eq!(
@@ -946,7 +946,7 @@ mod tests {
#[test]
fn listen_reply_tolerates_unknown_extra_fields() {
// The additive posture — the reply viewer must NOT deny
// The additive posture: the reply viewer must NOT deny
// unknown fields (future additive reply fields keep old
// consumers working).
let reply = json!({
@@ -969,9 +969,9 @@ mod tests {
#[test]
fn listen_reply_rejects_malformed_shapes() {
// No channel_id at all — not an open reply.
// No channel_id at all: not an open reply.
assert!(parse_listen_reply(&json!({"bound": {"path": "/p"}})).is_err());
// channel_id = 0 (the wrapper never allocates 0 — channel 0 is
// channel_id = 0 (the wrapper never allocates 0; channel 0 is
// the call plane).
assert!(parse_listen_reply(&json!({"channel_id": 0})).is_err());
// Malformed bound (not substrate-shaped).
@@ -1015,9 +1015,9 @@ mod tests {
#[test]
fn forwarded_spec_peer_schema_gates_shape_per_substrate() {
// The per-substrate peer sub-schema: the same if/then
// structure the direct op pins its target with — structural
// assertions per tunnels/params' precedent (no jsonschema dep
// here).
// structure the direct op pins its target with, asserted
// structurally per tunnels/params' precedent (no jsonschema
// dep here).
let schema = tunnel_forwarded_spec().input_schema;
let allof = schema["allOf"].as_array().expect("allOf branches");
assert_eq!(allof.len(), 2, "tcp/udp branch + unix branch");
@@ -1056,7 +1056,7 @@ mod tests {
// fails loud at the deserializer. `deny_unknown_fields` holds:
// unknown top-level params reject; the untagged peer is
// lenient about extra peer fields (same documented asymmetry
// as the direct op's target — the SCHEMA is the shape-vs-
// as the direct op's target: the SCHEMA is the shape-vs-
// substrate gate).
let cross_shape = json!({
"listen": 1,
@@ -1109,7 +1109,7 @@ mod tests {
#[test]
fn listen_reply_fields_tolerates_unknown_extra_fields() {
// The additive posture — the lenient projection must NOT gate
// The additive posture: the lenient projection must NOT gate
// unknown reply fields.
let reply = json!({
"channel_id": 7,
src/producer.rs
+125 -125
View File
@@ -1,4 +1,4 @@
//! The producer half: the `channels/tunnel/sub` open op — its
//! The producer half: the `channels/tunnel/sub` open op, its
//! establisher (dial shape), its substrate-agnostic pump handler, and
//! its registration (`register_tunnel_openable`) per `producer.md` /
//! ADR-004.
@@ -6,11 +6,11 @@
//! The establisher is the open op's awaited, bounded establishment
//! phase (alkcall ADR-049): parse params semantically, look the
//! resource up in the producer's registry, dial via the injected
//! `DialFn` (ADR-004 — "produce boxed halves for a resource" is a
//! `DialFn` (ADR-004: "produce boxed halves for a resource" is a
//! function, not a trait), and return the handle in
//! `Establishment::new(plan)` (R-01 — typed-opaque `ChannelPlan`; its
//! `Establishment::new(plan)` (R-01: typed-opaque `ChannelPlan`; its
//! `Send + Sync` bound is why `TargetHandle`'s halves carry `+ Sync`).
//! The dial closure closes over substrate code — the protocol crate
//! The dial closure closes over substrate code: the protocol crate
//! never sees a socket type.
//!
//! The pump handler downcasts the plan and awaits
@@ -19,13 +19,13 @@
//! return = teardown-at-birth). It cannot know what the halves came
//! from and must not (AGENTS.md convention 7).
//!
//! The listen establisher variant (the producer-side listener — the
//! The listen establisher variant (the producer-side listener, the
//! `-R` far-side listener shape) is [`listen_establisher`] +
//! [`AcceptQueue`] below (the `tunnels/producer-listen` shape). The
//! direct op's establisher ([`direct_establisher`] — dynamic-target
//! direct op's establisher ([`direct_establisher`], dynamic-target
//! egress, ADR-007) is the dial shape minus the registry lookup.
//!
//! The bind-first shape's plan payload ([`BindFirstPlan`] — the opaque
//! The bind-first shape's plan payload ([`BindFirstPlan`]: the opaque
//! listener + the correlation token, ADR-008 as amended) is defined
//! here; the establisher and accept dispatcher that consume it land
//! with `tunnels/bindfirst-producer` (the wire-stable types pin
@@ -53,7 +53,7 @@ pub use crate::params::{
TUNNEL_OPEN_SCOPE,
};
/// The dialed target handle — the plan payload (typed-opaque
/// The dialed target handle: the plan payload (typed-opaque
/// `ChannelPlan`; establisher and handler agree on the concrete
/// type). The boxed halves carry `+ Sync` to satisfy the plan's
/// `Send + Sync` bound (POC F-1, documented upstream).
@@ -75,29 +75,29 @@ impl std::fmt::Debug for TargetHandle {
/// the assembly's listener handle + the correlation token + the listen
/// tunnel's substrate.
///
/// **The listener is NOT `TargetHandle` halves** — a listener is
/// **The listener is NOT `TargetHandle` halves**: a listener is
/// accepted FROM, not pumped. The plan carries whatever the assembly's
/// `BindFn` produced, typed-opaque ([`Any`]; the same
/// injection pattern as `DialFn`/`AcceptFn` — the accept dispatcher
/// injection pattern as `DialFn`/`AcceptFn`; the accept dispatcher
/// downcasts it to the assembly's listener type via a type-erased
/// accept closure; the ADR's "listener handle in the plan" is
/// assembly-typed, and the protocol sees only this opaque pair).
///
/// **The correlation token (ADR-008 Amendment 1):** minted by the
/// bind-first establisher — a fresh u32 per open, unique per producer
/// bind-first establisher: a fresh u32 per open, unique per producer
/// (an `AtomicU32` counter seeded non-zero, e.g. 1, inside the
/// establisher's closure state; wraps are acceptable — u32, and a wrap
/// establisher's closure state; wraps are acceptable: u32, and a wrap
/// collision requires 2^32 concurrent opens on one producer). Carried
/// twice: in the plan (the accept dispatcher stamps it into every
/// forwarded open it opens for this listener) and as the additive
/// `listen` reply field (the opener correlates its waiting session on
/// it). It is NOT the listen channel's ID — the serving-side handler
/// it). It is NOT the listen channel's ID; the serving-side handler
/// never sees its own channel ID, and generic hub relays forward
/// params untouched (Amendment 1's rationale). Per-open fresh,
/// per-producer unique; retired with the listener session.
pub struct BindFirstPlan {
/// The assembly-typed listener handle (e.g. a `TcpListener`
/// wrapper the `BindFn` produced). Opaque to the protocol — the
/// wrapper the `BindFn` produced). Opaque to the protocol: the
/// accept dispatcher downcasts via a type-erased accept closure.
pub listener: Arc<dyn Any + Send + Sync>,
/// The correlation token (see above): minted fresh per open,
@@ -106,11 +106,11 @@ pub struct BindFirstPlan {
pub token: u32,
/// The listen tunnel's substrate (ADR-008 §3), stamped here at
/// establisher time (the establisher already parsed the params
/// successfully — one parse, one source of truth). The accept
/// successfully: one parse, one source of truth). The accept
/// dispatcher reads it from the plan it already downcast; it must
/// NOT re-parse wire input (a silent `Tcp` fallback on a parse
/// failure would mismatch a UDP listen tunnel's data-plane
/// framing — review 002 M-1).
/// framing, review 002 M-1).
pub substrate: Substrate,
}
@@ -158,9 +158,9 @@ impl From<TunnelEstablishError> for EstablishmentError {
/// The producer's registry of produced resources:
/// `(resource, substrate)` → backing. An assembly-owned construct
/// (OQ-TN-11 — the collision domain is the producer's registry; two
/// (OQ-TN-11: the collision domain is the producer's registry; two
/// producers on one connection may expose the same resource name
/// independently). The backing is opaque to the protocol — the
/// independently). The backing is opaque to the protocol: the
/// `DialFn` closes over whatever shape the assembly layer chose.
#[derive(Clone, Default)]
pub struct ResourceRegistry {
@@ -173,7 +173,7 @@ impl ResourceRegistry {
}
/// Produce a resource: map `(resource, substrate)` to its backing
/// (a local port, a socket path, an in-process address — the
/// (a local port, a socket path, an in-process address; the
/// dial closure's business, not the wire's).
pub async fn register(&self, resource: &str, substrate: Substrate, backing: &str) {
self.targets
@@ -208,7 +208,7 @@ pub type DialFn = Arc<
>;
/// The witness capturing the establisher's per-call opener identity
/// (CF-006 probe — the establisher sees the END CALLER overlaid onto
/// (CF-006 probe: the establisher sees the END CALLER overlaid onto
/// the install-time context, not a synthetic install-time value).
pub type IdentityWitness = Arc<tokio::sync::Mutex<Option<String>>>;
@@ -256,26 +256,26 @@ fn parse_params(input: &Value) -> Result<TunnelParams, TunnelEstablishError> {
.map_err(|e| TunnelEstablishError::HandlerError(format!("invalid params: {e}")))
}
/// The direct establisher (ADR-007 §4 — the dial shape minus the
/// The direct establisher (ADR-007 §4, the dial shape minus the
/// registry lookup): parse `TunnelDirectParams`, render the target to
/// the same backing-string form the registry backings use
/// (`"host:port"` for tcp/udp, the path for unix), and call the
/// injected [`DialFn`] — the SAME dial function the base op uses, so
/// injected [`DialFn`]; the SAME dial function the base op uses, so
/// a producer that registered `dial_tcp` for the base op registers
/// the direct op with zero new substrate code (the UDP framed-adapter
/// wrap, ADR-003, happens inside the dial closure — the establisher
/// stays substrate-agnostic). No registry lookup — `unknown_resource`
/// wrap, ADR-003, happens inside the dial closure; the establisher
/// stays substrate-agnostic). No registry lookup, so `unknown_resource`
/// can never fire.
///
/// The per-call `auth` carries the opener identity (CF-006); record it
/// into `identity_witness` when provided (the direct op is the
/// arbitrary-egress capability — identity visibility matters more,
/// arbitrary-egress capability: identity visibility matters more,
/// not less).
///
/// Error mapping (ADR-007 §4's table minus the registry row): the
/// serde-parse failure of a schema-valid-shaped target maps
/// [`TunnelEstablishError::HandlerError`] (the schema gate should
/// have caught it — the parse is the backstop); dial failures pass
/// have caught it; the parse is the backstop); dial failures pass
/// through the `DialFn`'s own typed errors unchanged (`dial_failed` /
/// `resource_shortage` / `handler_error`).
pub fn direct_establisher(dial: DialFn) -> OpenEstablisher {
@@ -307,17 +307,17 @@ pub fn direct_establisher_with_witness(
})
}
/// The listen establisher (shape 2 — producer.md §The Establisher):
/// The listen establisher (shape 2, producer.md §The Establisher):
/// the same open op, same params, same typed errors; the establisher
/// pops the next accepted connection from the injected [`AcceptFn`]
/// instead of dialing. The resource lookup follows the same registry
/// miss mapping (`unknown_resource`); the accept closure's failures
/// map at the closure (empty queue → `resource_shortage`, closed
/// listener → `dial_failed` — producer.md's table). The pump handler
/// listener → `dial_failed`; producer.md's table). The pump handler
/// is UNCHANGED: the listen variant is plan-flow with a different
/// halves source.
///
/// The registry check is a namespace gate only — the listen producer's
/// The registry check is a namespace gate only: the listen producer's
/// "backing" is its accept queue, not a dial target. A resource the
/// producer does not produce is still `unknown_resource` (same as
/// dial); an openable resource with nothing yet accepted is
@@ -344,9 +344,9 @@ pub fn listen_establisher(registry: ResourceRegistry, accept: AcceptFn) -> OpenE
}
/// The accepted-connection source the assembly layer injects for a
/// LISTEN producer (shape 2 — the `-R` far-side listener; producer.md
/// LISTEN producer (shape 2, the `-R` far-side listener; producer.md
/// §The Establisher): pops one accepted connection's halves. The
/// assembly layer owns the listener + its accept loop — the protocol
/// assembly layer owns the listener + its accept loop; the protocol
/// crate never binds (OQ-TN-04). The queue contract lives in
/// [`AcceptQueue`]; the closure pops from it (or from whatever the
/// assembly chose).
@@ -360,17 +360,17 @@ pub type AcceptFn = Arc<
/// listen establisher (the protocol-side queue contract). The accept
/// loop feeds ([`AcceptQueue::push`]); the establisher pops during the
/// open op's establishment phase ([`AcceptQueue::pop`]). `pop` is
/// bounded by the establisher's own deadline (ADR-049 §2) — but an
/// bounded by the establisher's own deadline (ADR-049 §2), but an
/// empty queue is NOT inherently an error (a late accept is
/// legitimate); the mapping to `resource_shortage` is the ASSEMBLY
/// layer's posture (its accept closure knows the listener's bound —
/// layer's posture (its accept closure knows the listener's bound:
/// a drained listener or an accept budget). The wrapper's
/// establishment timeout is the backstop above that.
///
/// `pop` returns `None` when the queue is closed-and-empty (the
/// listener is gone — the assembly layer calls [`AcceptQueue::close`]
/// listener is gone; the assembly layer calls [`AcceptQueue::close`]
/// on teardown). FIFO order: every popped handle was accepted BEFORE
/// that pop began (always-before-take ordering — push enqueues under
/// that pop began (always-before-take ordering: push enqueues under
/// the same lock that pops dequeue under, and `pop` re-checks the
/// queue after every wake).
#[derive(Clone, Default)]
@@ -414,7 +414,7 @@ impl AcceptQueue {
/// is created BEFORE the state check on every iteration. A `push`
/// landing in the thread-preemption window between the check and
/// the future's first poll is captured by tokio's notify-waiters
/// counter comparison at that first poll — the future resolves
/// counter comparison at that first poll; the future resolves
/// immediately; a later `push` finds the waiter registered and
/// wakes it. A future created after the check (the naive shape)
/// loses that push, parking the pop until the next one.
@@ -443,10 +443,10 @@ impl AcceptQueue {
/// The pump handler: accept the channel's yield-once `BiStream`,
/// downcast the plan to `TargetHandle`, and await `pump_bidi` inline
/// (R-02 — the returned `JoinHandle` tracks the data-plane lifetime;
/// (R-02: the returned `JoinHandle` tracks the data-plane lifetime;
/// an early return is teardown-at-birth, and the birth-teardown
/// telemetry diagnoses the spawn-and-forget shape at runtime). The
/// handler is substrate-agnostic by construction — it sees boxed
/// handler is substrate-agnostic by construction: it sees boxed
/// halves, never a socket type.
pub fn make_tunnel_pump_handler() -> OpenHandler {
Arc::new(
@@ -505,7 +505,7 @@ pub fn register_tunnel_openable(
/// base op (the direct op's plan payload is a [`TargetHandle`] exactly
/// like the base op's; plan-flow is unchanged), via
/// `register_openable_with_establisher` (`timeout: None` = the 10s
/// default bound). `dial` is the SAME [`DialFn`] the base op uses —
/// default bound). `dial` is the SAME [`DialFn`] the base op uses;
/// no new substrate code. `identity_witness` is the optional CF-006
/// probe seam (the establisher records the per-call opener identity
/// it saw).
@@ -529,8 +529,8 @@ pub fn register_tunnel_direct_openable(
/// Register the tunnel open op with a LISTEN establisher (shape 2):
/// the same spec + pump handler as [`register_tunnel_openable`]; the
/// establisher pops accepted handles from the injected [`AcceptFn`]
/// instead of dialing. One registration per operation id per registry
/// — a session serves either a dial establisher or a listen
/// instead of dialing. One registration per operation id per registry:
/// a session serves either a dial establisher or a listen
/// establisher for `channels/tunnel/sub`; a producer that does both
/// registers on separate registries (per-connection registries are
/// independent, producer.md §Registration API).
@@ -553,10 +553,10 @@ pub fn register_tunnel_listen_openable(
// --- Bind-first (shape 3, ADR-008) ---------------------------------------
/// The substrate bind function the assembly layer injects (shape 3 —
/// The substrate bind function the assembly layer injects (shape 3,
/// ADR-008 §1): given a substrate + a registry backing (the same
/// backing-string form the dial shape's `DialFn` receives), **bind the
/// listener** and produce it. The protocol never binds (OQ-TN-04) —
/// listener** and produce it. The protocol never binds (OQ-TN-04);
/// this closure is the platform boundary, the same inversion point as
/// `DialFn`. The returned [`BoundListener`] pairs the opaque
/// assembly-typed listener (the accept loop's business) with the
@@ -573,11 +573,11 @@ pub type BindFn = Arc<
/// What the assembly's bind produces (ADR-008 §1): the opaque listener
/// (the accept loop's business) + the OS-chosen bound address (the
/// `bound` reply field's source — the REAL resolved address, port `0`
/// `bound` reply field's source: the REAL resolved address, port `0`
/// resolved; informational surface, never addressing).
pub struct BoundListener {
/// The assembly-typed listener (e.g. the `TcpListener` wrapper the
/// assembly chose). Opaque to the protocol — rides the plan's
/// assembly chose). Opaque to the protocol: rides the plan's
/// `Any`, downcast only inside the assembly's accept closure.
pub listener: Arc<dyn Any + Send + Sync>,
/// The real OS-chosen bound address (the `bound` reply field's
@@ -587,15 +587,15 @@ pub struct BoundListener {
/// The token mint (the discipline pinned on [`BindFirstPlan`]): a
/// non-zero-seeded `AtomicU32` counter held in the establisher's
/// closure state — fresh per open, unique per producer, wraps
/// closure state, fresh per open, unique per producer, wraps
/// acceptable (a wrap collision requires 2^32 concurrent opens on one
/// producer — and that bound is loose: the token is
/// producer, and that bound is loose: the token is
/// per-producer-CONNECTION and each open rides one connection, so the
/// alkcall 256-channels-per-connection cap makes a live-collision
/// window unreachable by a wide margin, review 002 N-5). The counter
/// seeds at 1: `fetch_add` returns the
/// PREVIOUS value, so the first mint is 1 and the minted stream is
/// 1, 2, 3, … — token 0 (an illegal wire value; the forwarded params'
/// 1, 2, 3, …; token 0 (an illegal wire value; the forwarded params'
/// `listen` pin rejects it) is never minted, and a u32 wrap that lands
/// on 0 re-increments (the loop skips it; the 2^32 collision window
/// makes wrap uniqueness a non-issue at the discipline's own terms).
@@ -609,14 +609,14 @@ fn mint_token(counter: &std::sync::atomic::AtomicU32) -> u32 {
}
}
/// The bind-first establisher (shape 3 — producer.md §The Establisher,
/// The bind-first establisher (shape 3, producer.md §The Establisher,
/// ADR-008 §1): the same open op, same params, same typed-error table;
/// the establisher **binds** (via the injected [`BindFn`]) instead of
/// dialing or popping an accept, mints the correlation token
/// (per-open-fresh, per-producer-unique — the discipline pinned on
/// (per-open-fresh, per-producer-unique, the discipline pinned on
/// [`BindFirstPlan`]), contributes
/// the `bound` + `listen` reply fields (the establisher → reply
/// projection, ADR-049 amendment 3 — the wrapper merges them into the
/// projection, ADR-049 amendment 3; the wrapper merges them into the
/// success output after its reserved `channel_id` key), and returns
/// the [`BindFirstPlan`] as the plan payload.
///
@@ -624,7 +624,7 @@ fn mint_token(counter: &std::sync::atomic::AtomicU32) -> u32 {
/// [`listen_establisher`]): the bind-first producer's "backing" is the
/// [`BindFn`], not a dial target; a resource the producer does not
/// produce is still `unknown_resource`. The resolved backing string
/// rides the same signature the dial shape's `DialFn` takes — the bind
/// rides the same signature the dial shape's `DialFn` takes; the bind
/// closure sees the same `(substrate, backing)` form.
///
/// Error mapping (producer.md's shape-3 table): unbindable/held port →
@@ -633,11 +633,11 @@ fn mint_token(counter: &std::sync::atomic::AtomicU32) -> u32 {
/// `unknown_resource`; internal → `handler_error`. The `BindFn`
/// returns the typed errors; the establisher passes them through.
///
/// The token lives in BOTH the reply (`listen` — the opener's
/// The token lives in BOTH the reply (`listen`, the opener's
/// correlation key) and the plan (the accept dispatcher stamps it into
/// every forwarded open). Two opens mint distinct tokens (the mint is
/// per-open); each dispatcher stamps ITS OWN plan's token (the plan
/// flows per-open — R-01).
/// flows per-open, R-01).
pub fn bindfirst_establisher(registry: ResourceRegistry, bind: BindFn) -> OpenEstablisher {
use std::sync::atomic::AtomicU32;
let counter = Arc::new(AtomicU32::new(1));
@@ -673,10 +673,10 @@ pub fn bindfirst_establisher(registry: ResourceRegistry, bind: BindFn) -> OpenEs
}
/// The accepted-connection source the assembly layer injects for the
/// BIND-FIRST producer (shape 3 — ADR-008 §1): pops one accepted
/// BIND-FIRST producer (shape 3, ADR-008 §1): pops one accepted
/// connection from the plan's opaque listener and yields
/// `(SubstrateAddr /* peer */, TargetHandle /* accepted halves */)`.
/// The closure receives the listener as the opaque `Any` — the
/// The closure receives the listener as the opaque `Any`; the
/// assembly downcasts it to ITS concrete listener type inside the
/// closure (the same type-erased injection pattern as
/// `DialFn`/`AcceptFn`; substrate code lives behind the closure,
@@ -685,10 +685,10 @@ pub fn bindfirst_establisher(registry: ResourceRegistry, bind: BindFn) -> OpenEs
///
/// A returned `Err` is a LOOP-CRITICAL condition: the accept loop's
/// substrate source is broken (the listener closed, accept errored
/// unrecoverably) — the dispatcher ends the loop, tears the listener
/// unrecoverably); the dispatcher ends the loop, tears the listener
/// down with the plan, and lets the handler task finish (the channel
/// reaps). Per-ACCEPT failures (forwarded-open failures) do NOT come
/// from this closure — see `make_bindfirst_accept_handler`'s error
/// from this closure; see `make_bindfirst_accept_handler`'s error
/// posture.
pub type BindFirstAcceptLoop = Arc<
dyn Fn(
@@ -706,12 +706,12 @@ pub type BindFirstAcceptLoop = Arc<
/// opener, adopts the opener-allocated ID on the producer-side
/// manager, and returns the adopted channel's halves (the pump side).
///
/// **This closure is the call plane** (the design pin — producer.md
/// **This closure is the call plane** (the design pin: producer.md
/// §The Pump Handler item 5): the per-channel `Connection` the
/// dispatcher receives is stream-less (`open_bi` is `StreamClosed` —
/// dispatcher receives is stream-less (`open_bi` is `StreamClosed`,
/// ADR-035) and carries no call surface, so the registration closes
/// over the serving session's `Arc<CallConnection>` (the assembly has
/// it — it constructed the serving session). The shape is the reverse
/// it; it constructed the serving session). The shape is the reverse
/// POC's `open_reverse_channel`: `call_with_payload` on channel 0 →
/// extract `channel_id` from the reply → `manager.adopt_channel`. The
/// accepted halves pump against the adopted channel's halves.
@@ -731,41 +731,41 @@ pub type ForwardedOpener = Arc<
+ Sync,
>;
/// The bind-first pump handler — the ACCEPT DISPATCHER (ADR-008 §1).
/// The bind-first pump handler: the ACCEPT DISPATCHER (ADR-008 §1).
///
/// Handler contract (producer.md §The Pump Handler item 5 + R-02):
///
/// 1. `conn.accept_bi()` FIRST (the yield-once contract — the
/// 1. `conn.accept_bi()` FIRST (the yield-once contract: the
/// wrapper's accepted-flag telemetry requires it), downcast the
/// plan to [`BindFirstPlan`].
/// 2. The listen channel's own `BiStream` stays idle (no pumps on
/// it); **its EOF cancels the loop and tears the listener down** —
/// a 1-byte `read` on the idle stream observes the opener dropping
/// it); **its EOF cancels the loop and tears the listener down**: a
/// 1-byte `read` on the idle stream observes the opener dropping
/// the listener session (the zero-byte EOF sentinel; ADR-050's
/// teardown contract). `select!` races the EOF watch against the
/// accept loop.
/// 3. Per accept: open the forwarded channel ([`ForwardedOpener`] —
/// 3. Per accept: open the forwarded channel ([`ForwardedOpener`],
/// the call plane, `{listen: token, substrate, peer}` params),
/// then `pump_bidi` the accepted halves against the adopted
/// channel inside a spawned task tracked in the handler's handle
/// set. A forwarded-open failure is per-accept isolated: log and
/// drop that accepted connection — the loop continues (the
/// drop that accepted connection; the loop continues (the
/// listener's overall health is not hostage to one failed accept;
/// v1: no retry/backoff). A loop-critical accept error ([`BindFirstAcceptLoop`]'s
/// contract) ends the loop.
/// 4. **Lifetime (R-02):** the handler's returned `JoinHandle` tracks
/// the whole accept loop + every per-accept pump it spawned — the
/// the whole accept loop + every per-accept pump it spawned; the
/// handler task awaits the pump handles on loop exit (a forwarded
/// pump outliving its listener is a leak). Handler exit → wrapper
/// teardown (the listen channel reaps) → the far listener tears
/// down (the plan's listener drops with the loop).
///
/// The substrate discriminator for the forwarded opens rides the
/// plan ([`BindFirstPlan`].substrate — stamped at establisher time
/// plan ([`BindFirstPlan`].substrate, stamped at establisher time
/// from the one successful `TunnelParams` parse; the listen tunnel's
/// substrate; ADR-008 §3 "the listen tunnel's substrate shapes the
/// forwarded channel's data-plane framing"). The dispatcher never
/// re-parses wire input — no fallback branch exists.
/// re-parses wire input: no fallback branch exists.
pub fn make_bindfirst_accept_handler(
accept_loop: BindFirstAcceptLoop,
open_forwarded: ForwardedOpener,
@@ -784,14 +784,14 @@ pub fn make_bindfirst_accept_handler(
return;
};
// The substrate rides the plan (stamped at establisher
// time — one parse, one source of truth): the
// time, one parse, one source of truth): the
// dispatcher must not re-parse wire input (a parse
// failure here would have nowhere loud to go; review
// 002 M-1 removed the silent `Tcp` fallback instead).
let substrate = plan.substrate;
let (mut idle_read, _idle_write) = tokio::io::split(bidi);
// The per-accept pump task set — tracked by the
// The per-accept pump task set, tracked by the
// handler's JoinHandle (R-02: every pump the handler
// spawned is awaited on loop exit).
let mut pumps: Vec<tokio::task::JoinHandle<(u64, u64)>> = Vec::new();
@@ -799,15 +799,15 @@ pub fn make_bindfirst_accept_handler(
let loop_result = loop {
// The idle-stream EOF watch vs the accept loop
// (select!): the opener dropping the listener
// session EOFs the listen channel's stream — the
// zero-byte sentinel — and cancels the loop.
// session EOFs the listen channel's stream (the
// zero-byte sentinel) and cancels the loop.
let eof_watch = async {
// The 1-byte read on the idle stream: `Ok(0)`
// (or an error) = the opener dropped the
// session; `Ok(1..)` never happens on a well-
// behaved opener (the stream is idle — a peer
// behaved opener (the stream is idle: a peer
// that writes is misbehaving; treat any
// non-empty read as an EOF trigger too — the
// non-empty read as an EOF trigger too; the
// channel is not a data plane).
let mut byte = [0u8; 1];
match idle_read.read(&mut byte).await {
@@ -845,8 +845,8 @@ pub fn make_bindfirst_accept_handler(
// under the serving side's
// establisher variant, ADR-008
// Amendment 2; other establisher
// failures → handler_error-class)
// — drop this accepted connection,
// failures → handler_error-class):
// drop this accepted connection,
// keep serving. v1: no retry/
// backoff (documented posture).
tracing::warn!(
@@ -861,7 +861,7 @@ pub fn make_bindfirst_accept_handler(
Err(e) => {
// Loop-critical (the listener's
// substrate source broke): end the
// loop — the listener tears down with
// loop; the listener tears down with
// the plan.
tracing::warn!(
listen = plan.token,
@@ -878,11 +878,11 @@ pub fn make_bindfirst_accept_handler(
reason = loop_result,
"tunnel bind-first: accept loop exited"
);
// The listener drops with the plan (the last Arc) — the
// The listener drops with the plan (the last Arc); the
// far listener tears down.
drop(plan);
// R-02: the handler's JoinHandle tracks every per-accept
// pump it spawned — await them on loop exit (a forwarded
// pump it spawned; await them on loop exit (a forwarded
// pump outliving its listener is a leak).
for pump in pumps.drain(..) {
let _ = pump.await;
@@ -894,16 +894,16 @@ pub fn make_bindfirst_accept_handler(
/// The bind-first registration's injected seams (the assembly layer
/// owns all four; bundling keeps the registration's signature to the
/// core pair + the bundle — the seams are one coherent injection
/// core pair + the bundle; the seams are one coherent injection
/// point, the same posture as `DialFn` riding the dial registration):
/// - `bind` — the substrate bind function ([`BindFn`]; the protocol
/// - `bind`: the substrate bind function ([`BindFn`]; the protocol
/// never binds, OQ-TN-04).
/// - `accept_loop` — the per-listener accept loop
/// - `accept_loop`: the per-listener accept loop
/// ([`BindFirstAcceptLoop`]; receives the opaque listener, yields
/// `(peer, halves)` per accept).
/// - `open_forwarded` — the forwarded-open caller ([`ForwardedOpener`];
/// - `open_forwarded`: the forwarded-open caller ([`ForwardedOpener`];
/// the call plane: the serving session's `Arc<CallConnection>` +
/// the producer-side `ChannelManager` for the adopt — the assembly
/// the producer-side `ChannelManager` for the adopt; the assembly
/// has both, it constructed the serving session; the reverse POC's
/// `open_reverse_channel` shape).
pub struct BindFirstInjections {
@@ -916,9 +916,9 @@ pub struct BindFirstInjections {
}
/// Register the bind-first openable (shape 3, ADR-008 §1): the SAME
/// spec + scope gate (`tunnel:open`) as the dial/listen variants — the
/// spec + scope gate (`tunnel:open`) as the dial/listen variants; the
/// bind-first shape is a per-registration establisher choice over the
/// same op (ADR-008 §5's per-registration posture) — with
/// same op (ADR-008 §5's per-registration posture), with
/// [`bindfirst_establisher`] as the establisher and
/// [`make_bindfirst_accept_handler`] as the pump. One establisher per
/// op id per registry (same note as
@@ -977,7 +977,7 @@ fn witness_establisher(
/// The correlation material the opener-side handler yields per
/// forwarded open: the parsed params + the channel's `BiStream`. The
/// ASSEMBLY correlates (token → its waiting listener session
/// bookkeeping) and owns the pump — it awaits the returned
/// bookkeeping) and owns the pump; it awaits the returned
/// `JoinHandle` inline (R-02: the forwarded channel's lifetime tracks
/// it; early return = teardown-at-birth reaps the channel), because
/// the local halves are assembly state (the SOCKS5 BIND client's
@@ -990,9 +990,9 @@ pub type ForwardedHandler =
/// The forwarded op's opener-side pump handler (ADR-008 §3, the
/// serving-shape wrapper both registrations share): parse
/// `TunnelForwardedParams`
/// (`deny_unknown_fields` + the positive-`listen` serde pin — a
/// (`deny_unknown_fields` + the positive-`listen` serde pin: a
/// malformed open fails the parse LOUD; the wrapper then returns early
/// and the channel tears down at birth — no phantom channel, the
/// and the channel tears down at birth; no phantom channel, the
/// correct shape for a malformed ingress), accept the channel's
/// `BiStream` (yield-once; accepting BEFORE the assembly handler runs
/// keeps the R-02 accepted-flag telemetry truthful), and hand the
@@ -1001,19 +1001,19 @@ pub type ForwardedHandler =
/// **Parse-before-accept ordering (the R-02 telemetry note):** the
/// params parse happens BEFORE `conn.accept_bi()`, so for a
/// malformed-params open the R-02 accepted-flag telemetry reads
/// `false` — by design. A malformed-params open IS a birth-teardown
/// `false`, by design. A malformed-params open IS a birth-teardown
/// shape (the exact case the telemetry's birth-teardown hint exists
/// for); accepting first would hold a stream open for a params set
/// the producer already got wrong. The hint firing here is correct,
/// not noise — future alkcall reviewers: do not flag the flag
/// not noise; future alkcall reviewers: do not flag the flag
/// (review 002 N-3).
///
/// **Pump responsibility split (the honest-shape decision):** the
/// injected handler receives the channel `BiStream` and owns the pump
/// — `pump_bidi(forwarded_bidi, local_read, local_write)` — because
/// (`pump_bidi(forwarded_bidi, local_read, local_write)`) because
/// the local halves are assembly state. The protocol-side built-in
/// handler only bridges wrapper→assembly; it awaits the injected
/// handler's `JoinHandle` inline (R-02 — the forwarded channel's
/// handler's `JoinHandle` inline (R-02: the forwarded channel's
/// lifetime tracks it, so a correlation miss or a dropped assembly
/// handler tears the channel down).
///
@@ -1022,16 +1022,16 @@ pub type ForwardedHandler =
/// establisher-less), a `listen` token with no matching waiting
/// session succeeds at the call plane and the miss is the ASSEMBLY's
/// post-open rejection (it holds the map): its handler's error path
/// drops the stream and returns — teardown-at-birth reaps the
/// drops the stream and returns; teardown-at-birth reaps the
/// forwarded channel (`correlation-miss`, the assembled decline
/// event at `debug`). Under the ESTABLISHER variant
/// ([`register_tunnel_forwarded_openable_with_establisher`] — the
/// ([`register_tunnel_forwarded_openable_with_establisher`], the
/// recommended shape), a no-route token fails at the open as a typed
/// `channel:open_failed` before the channel is born; THIS handler's
/// post-open miss is then only the residual race (a token vanishing
/// between gate pass and handler run) — teardown-at-birth remains the
/// between gate pass and handler run); teardown-at-birth remains the
/// fallback shape under both. The producer's accept loop drops the
/// accepted connection when that open fails (the shapes mirror —
/// accepted connection when that open fails (the shapes mirror:
/// per-accept isolation, `tunnels/bindfirst-producer`'s posture).
pub fn make_forwarded_op_handler(handler: ForwardedHandler) -> OpenHandler {
Arc::new(
@@ -1059,18 +1059,18 @@ pub fn make_forwarded_op_handler(handler: ForwardedHandler) -> OpenHandler {
)
}
/// Register the forwarded op's opener-side serving openable — the
/// BASE shape (ADR-008 §3 as amended): `ChannelCore::register_openable`
/// — **no establisher** (the forwarded channel needs no establishment;
/// its data plane starts immediately) — scope-gated `tunnel:forwarded`
/// Register the forwarded op's opener-side serving openable, the
/// BASE shape (ADR-008 §3 as amended): `ChannelCore::register_openable`;
/// **no establisher** (the forwarded channel needs no establishment;
/// its data plane starts immediately), scope-gated `tunnel:forwarded`
/// (ingress, a separate grant from `tunnel:open`/`tunnel:direct`,
/// never implied by either — ADR-006 Amendment 1), via
/// never implied by either, ADR-006 Amendment 1), via
/// [`crate::params::tunnel_forwarded_spec`]. The wrapper's
/// `channel_id` allocation IS the reply (the producer adopts it).
///
/// **Two-shape posture (ADR-008 Amendment 2):** this base shape stays
/// byte-for-byte compatible for assemblies that prefer the post-open
/// decline (and as the residual-race fallback posture — see
/// decline (and as the residual-race fallback posture; see
/// [`make_forwarded_op_handler`]'s correlation-miss block). The
/// ESTABLISHER variant
/// ([`register_tunnel_forwarded_openable_with_establisher`]) is the
@@ -1082,16 +1082,16 @@ pub fn make_forwarded_op_handler(handler: ForwardedHandler) -> OpenHandler {
/// Identity (CF-005 precedence, upstream machinery): the opener-side
/// gate sees the PRODUCER's per-call identity (the producer may carry
/// an `auth_token` when it is itself forwarding); the `peer` param is
/// producer-asserted metadata, not transport truth — the ACL trusts
/// producer-asserted metadata, not transport truth: the ACL trusts
/// the grant, the application treats `peer` as metadata. Nothing to
/// implement here (upstream's per-call overlay handles it); the test
/// suite pins it.
///
/// Post-hoc registration is supported (W2) — the dispatcher reads
/// Post-hoc registration is supported (W2): the dispatcher reads
/// through the shared `Arc<OperationRegistry>` per dispatch, the same
/// posture as [`register_tunnel_openable`]. One registration per op id
/// per registry; the op id is `channels/tunnel/forwarded` (distinct
/// from the base op — a session may serve both).
/// from the base op; a session may serve both).
pub fn register_tunnel_forwarded_openable(
core: &ChannelCore,
on_registry: &Arc<OperationRegistry>,
@@ -1107,7 +1107,7 @@ pub fn register_tunnel_forwarded_openable(
}
/// The forwarded op's establisher plan payload (ADR-008 Amendment 2):
/// an empty marker — a gate pass establishes nothing (the correlation
/// an empty marker: a gate pass establishes nothing (the correlation
/// map is assembly state and the handler needs nothing it does not
/// re-derive), the same typed-opaque [`ChannelPlan`] posture as ADR-049
/// amendment 2 with the payload choice pinned at "marker". Its `Some`
@@ -1118,26 +1118,26 @@ pub struct ForwardedPlan;
/// The injected correlation gate (ADR-008 Amendment 2, plan 003 W1):
/// the assembly's one-shot correlation check for the forwarded op's
/// serving half. `Err` at the gate → a typed `channel:open_failed`
/// BEFORE the channel is born (no phantom channel — ADR-049's posture
/// BEFORE the channel is born (no phantom channel; ADR-049's posture
/// extended to the forwarded op's serving side).
///
/// The gate receives the PARSED [`TunnelForwardedParams`] and runs
/// synchronously — it must be an in-memory map lookup (an await there
/// synchronously; it must be an in-memory map lookup (an await there
/// would extend the open's call-plane latency; the lookup shape is
/// what every assembly's correlation registry provides).
///
/// **Gate error convention (the pin):** a stale/missing `listen`
/// token SHOULD return [`TunnelEstablishError::UnknownResource`] —
/// token SHOULD return [`TunnelEstablishError::UnknownResource`]:
/// the waiting session IS a resource that is gone, and that maps to
/// `details.reason == "unknown_resource"` on the wire (the canonical
/// reason ADR-008 Amendment 2 pins; the wire vocabulary is NOT
/// extended). The crate passes through whatever the gate returns —
/// extended). The crate passes through whatever the gate returns;
/// the gate owns its rejection vocabulary, within the existing
/// [`TunnelEstablishError`] table.
pub type ForwardedGate =
Arc<dyn Fn(&TunnelForwardedParams) -> Result<(), TunnelEstablishError> + Send + Sync>;
/// The gate establisher (ADR-008 Amendment 2 — the correlation gate
/// The gate establisher (ADR-008 Amendment 2: the correlation gate
/// wrapped into the alkcall [`OpenEstablisher`] dance; assemblies do
/// not hand-roll the plumbing, the crate-owned registration surface
/// below is the only supported path).
@@ -1146,7 +1146,7 @@ pub type ForwardedGate =
/// runs the `tunnel:forwarded` ACL BEFORE the establishment phase
/// (alkcall's open-op wrapper order: schema → ACL → allocate →
/// establisher → handler), so the gate never sees an unauthorized
/// open — a scope-less producer is `FORBIDDEN` before any gate
/// open; a scope-less producer is `FORBIDDEN` before any gate
/// lookup. The establisher then parses the input and runs the gate;
/// the base shape's handler-side parse stays (see below), so
/// `deny_unknown_fields` + the positive-`listen` serde pin gate the
@@ -1159,7 +1159,7 @@ pub type ForwardedGate =
/// `channel:open_failed` with `details.reason == "handler_error"`.
/// This is deliberate: a malformed ingress becomes a typed
/// call-plane failure instead of the base shape's silent
/// birth-teardown (an improvement ADR-008 Amendment 2 pins — do
/// birth-teardown (an improvement ADR-008 Amendment 2 pins; do
/// not drift it).
/// - Gate `Err(TunnelEstablishError)` → the same existing table, 1:1
/// (a stale token: `UnknownResource` → `details.reason ==
@@ -1170,7 +1170,7 @@ pub type ForwardedGate =
/// `channel_id`).
///
/// `timeout: None` (the alkcall 10s default, as the other
/// registrations) — the gate is an in-memory lookup; the bound is
/// registrations); the gate is an in-memory lookup; the bound is
/// belt-and-braces, not workload.
fn make_forwarded_gate_establisher(gate: ForwardedGate) -> OpenEstablisher {
Arc::new(move |input: Value, _auth: AuthContext| {
@@ -1185,24 +1185,24 @@ fn make_forwarded_gate_establisher(gate: ForwardedGate) -> OpenEstablisher {
}
/// Register the forwarded op's opener-side serving openable WITH the
/// correlation-gate establisher (ADR-008 Amendment 2 — the
/// correlation-gate establisher (ADR-008 Amendment 2, the
/// recommended shape): `ChannelCore::register_openable_with_establisher`
/// with the crate-owned gate establisher (private
/// `make_forwarded_gate_establisher`) wrapping the injected
/// [`ForwardedGate`]. A stale/missing `listen` token fails the open
/// as a typed `channel:open_failed` BEFORE the channel's data plane
/// is served — the phantom-channel shape (succeed at the call plane,
/// is served; the phantom-channel shape (succeed at the call plane,
/// decline post-open, pump EOFs silently) is closed at the open.
///
/// The same op, params, scope (`tunnel:forwarded`, run before the
/// establisher — see the
/// establisher; see the
/// `make_forwarded_gate_establisher` doc) and reply
/// surface as [`register_tunnel_forwarded_openable`] ride; the wire
/// delta of Amendment 2 is empty (additive registration only).
///
/// The base shape (establisher-less) stays for compatibility + as the
/// residual-race fallback posture: a token that vanishes between
/// gate pass and handler run is post-open — the handler's
/// gate pass and handler run is post-open; the handler's
/// teardown-at-birth semantics (see [`make_forwarded_op_handler`]'s
/// correlation-miss block) are the only thing that can reap it, the
/// same under both shapes. New assemblies take this shape by default
@@ -1230,7 +1230,7 @@ mod tests {
/// The plan-payload contract: `BindFirstPlan` rides the
/// typed-opaque `ChannelPlan` (`Arc<dyn Any + Send + Sync>`) and
/// the accept dispatcher recovers it by downcast — the same
/// the accept dispatcher recovers it by downcast: the same
/// establisher/handler agreement `TargetHandle` pins. The opaque
/// listener half downcasts to the assembly's concrete type.
#[test]
@@ -1268,7 +1268,7 @@ mod tests {
/// The mint discipline: the counter seeds at 1, the first mint is
/// 1, the stream is per-open fresh + per-producer unique, and
/// token 0 is never minted (the loop skips it — a wrap that lands
/// token 0 is never minted (the loop skips it: a wrap that lands
/// on 0 re-increments).
#[test]
fn mint_token_seeds_nonzero_and_increments() {
@@ -1285,7 +1285,7 @@ mod tests {
/// re-increments (the loop skips the illegal wire value).
#[test]
fn mint_token_never_yields_zero() {
// Force the wrap shape: seed the counter at u32::MAX — the
// Force the wrap shape: seed the counter at u32::MAX; the
// next fetch_add returns u32::MAX (legal), then wraps to 0,
// which the loop must skip (the mint after the wrap is 1).
let counter = std::sync::atomic::AtomicU32::new(u32::MAX);
@@ -1393,7 +1393,7 @@ mod tests {
assert_ne!(t1, t2, "two opens mint distinct tokens");
}
/// The registry miss → `unknown_resource` (the namespace gate —
/// The registry miss → `unknown_resource` (the namespace gate,
/// the same posture as the listen establisher).
#[tokio::test]
async fn bindfirst_establisher_registry_miss_is_unknown_resource() {
src/wire.rs
+13 -13
View File
@@ -4,14 +4,14 @@
//! carries its BAST contract at `docs/architecture/bast.md`.
//!
//! The channels layer already length-prefixes every chunk (8-byte
//! header), so stream substrates need zero tunnel-level framing — the
//! header), so stream substrates need zero tunnel-level framing: the
//! halves are the tunnel, and a zero-length read is genuinely EOF
//! (stream semantics). UDP needs boundary re-framing only: datagram
//! boundaries do not survive the chunk stream, and `u16` suffices
//! (UDP's max payload 65507 < 65535).
//!
//! Sentinel layering (the F-2 invariant, load-bearing): `len == 0` is
//! a legal *empty datagram* — a real datagram, never an EOF signal.
//! a legal *empty datagram* (a real datagram, never an EOF signal).
//! EOF is the channels-level sentinel on the `BiStream` and lives at
//! a different layer; the codec never emits zero-length reads, so the
//! layers never collide. This is why raw pass-through is structurally
@@ -20,31 +20,31 @@
//!
//! No chunk-type byte, no 5-byte header: a tunnel has exactly one data
//! stream per direction (the channel's read/write halves), so there is
//! no sub-demux key — the structural difference from alktty's 5-byte
//! no sub-demux key: the structural difference from alktty's 5-byte
//! `[stream_type][len]` header.
use bytes::{Buf, BufMut, Bytes, BytesMut};
use thiserror::Error;
/// The u16 length-field bound: datagrams larger than this are
/// rejected at frame time (ADR-003 — never a wire overflow).
/// rejected at frame time (ADR-003; never a wire overflow).
pub const MAX_DATAGRAM_LEN: usize = u16::MAX as usize;
const DATAGRAM_LEN_FIELD: usize = 2;
/// Codec errors: oversize at frame time. The truncated-stream case is
/// not a codec error — the decoder buffers partial frames indefinitely
/// not a codec error: the decoder buffers partial frames indefinitely
/// (`is_mid_datagram` is the diagnostic) and the SESSION surfaces a
/// stream end mid-datagram as `TunnelIoError::TruncatedDatagram` (the
/// fail-loud posture, OQ-TN-13 — never a silent partial datagram).
/// fail-loud posture, OQ-TN-13: never a silent partial datagram).
///
/// The decode direction (`DatagramReader::feed`) is infallible: with
/// the oversize bound enforced at frame time, a declared length can
/// never exceed what the frame itself carries, so the incremental
/// decode cannot fail — only buffer and complete.
/// decode cannot fail, only buffer and complete.
#[derive(Debug, Error)]
pub enum DatagramCodecError {
/// A datagram larger than [`MAX_DATAGRAM_LEN`] was framed — the
/// A datagram larger than [`MAX_DATAGRAM_LEN`] was framed; the
/// u16 length field would wrap. Rejected at frame time.
#[error("datagram too large for u16 length prefix: {0} bytes (max {1})")]
Oversize(usize, usize),
@@ -68,7 +68,7 @@ pub fn frame_datagram(payload: &[u8]) -> Result<Bytes, DatagramCodecError> {
/// Incremental datagram decoder over the channel's read half. Feed it
/// whatever `read()` yields; it reassembles `[len: u16 BE][payload]`
/// frames across chunk boundaries — a datagram may be split across
/// frames across chunk boundaries: a datagram may be split across
/// two chunks, two datagrams may share one chunk, and partial headers
/// wait for more bytes. This is the only consumer-visible decode path
/// (ADR-003; chunk splitting/batching is transparent).
@@ -93,10 +93,10 @@ impl DatagramReader {
}
/// Feed raw bytes from the read half. Returns every datagram that
/// completed with this feed, in order — zero or more per chunk.
/// completed with this feed, in order (zero or more per chunk).
/// Buffers partial frames across chunk boundaries. Infallible: the
/// decode direction cannot fail (see the [`DatagramCodecError`]
/// docs — the oversize bound is enforced at frame time).
/// docs: the oversize bound is enforced at frame time).
pub fn feed(&mut self, chunk: &[u8]) -> Vec<Bytes> {
self.buf.extend_from_slice(chunk);
let mut out = Vec::new();
@@ -134,7 +134,7 @@ impl DatagramReader {
}
/// `true` while a partial datagram is in flight (for teardown
/// diagnostics — a clean peer never ends the stream mid-datagram;
/// diagnostics: a clean peer never ends the stream mid-datagram;
/// an end while this is set is the session's
/// `TruncatedDatagram` condition).
pub fn is_mid_datagram(&self) -> bool {
@@ -144,7 +144,7 @@ impl DatagramReader {
/// Bytes currently retained by the decoder state: the leftover
/// buffer plus the in-flight partial datagram's accumulated
/// payload. Post-feed this is bounded by one frame
/// (≤ [`MAX_DATAGRAM_LEN`] + 2) — the decoder never grows its
/// (≤ [`MAX_DATAGRAM_LEN`] + 2); the decoder never grows its
/// state with fed bytes beyond the frame in progress (pinned by
/// the fuzz harness, `docs/research/fuzzing.md` §3.1).
pub fn retained_bytes(&self) -> usize {