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