seams: the three seam types — dial / egress / accept (ADR-004)
- egress.rs: the I/O-plane inversion point, function-not-trait at the three factories; Egress is the one trait (the per-association returned object) - dial: TargetHandle (boxed halves, +Sync per the ChannelPlan bound), DialFn over the typed TargetAddr (the POCs' JSON stand-in retired), SocksDialError with the five-variant RFC-faithful mapping - egress: Egress (send payload-only; next_reply returns inner bytes; the readiness contract — Pending when nothing is ready — documented on the trait), EgressFactory, EgressError, the egress_factory sync-lift constructor - accept: BindAccept (the accept future owns its resources end-to-end; the quiet wait's drop IS the client-gone teardown), BindTargetHandle (ownership structural; the composed keep-alive owner rides inside the halves provider-side), BindAcceptFactory, the two-stage error pair (factory → reply#1; accept-stage → reply#2) - no substrate type named anywhere in the module; ownership rides the handle structurally (no _ bindings) - 9 new tests: the four error→reply-code tables (all variants, byte-compared), the sync-lift Ok/Err paths, the closure compile shapes, the Send/Sync/Unpin assertions; lib.rs re-exports the surface Verified: cargo test (113 green), --all-features green, clippy -D warnings clean (native + wasm32), fmt clean, cargo doc zero warnings, wasm32 check clean.
This commit is contained in:
1 parent
1ccd5bbe0c
commit
ac361f7610
3 files changed
+684
-14
No files matched your search
+593
-5
@@ -1,6 +1,594 @@
|
||||
//! The egress/accept seam types: the `DialFn` / `EgressFactory` /
|
||||
//! `BindAcceptFactory` contracts (function-not-trait, ownership rides
|
||||
//! the handle — ADR-004); the inversion point where no substrate type
|
||||
//! is ever named.
|
||||
//! The three I/O-plane seam types (ADR-004) — the dial seam (CONNECT),
|
||||
//! the egress seam (UDP ASSOCIATE), the accept seam (TCP BIND) — as
|
||||
//! function-shaped callbacks with their error vocabularies. These types
|
||||
//! are the inversion point (AGENTS.md convention 7): the protocol layer
|
||||
//! never names a substrate type; the feature-gated providers
|
||||
//! (`local`/`composed`, batch 2) and the test harness construct them.
|
||||
//!
|
||||
//! Stub: filled by the `implementation/core-seam-types` task.
|
||||
//! **Function, not trait** (ADR-004): [`DialFn`], [`EgressFactory`],
|
||||
//! and [`BindAcceptFactory`] are function-shaped callbacks — "produce
|
||||
//! boxed halves for a resource" is a function, not a trait (the
|
||||
//! alktunnels ADR-004 finding, extended to all three SOCKS5 seams by
|
||||
//! POC #1/#2/#5). [`Egress`] is the one trait: the per-association
|
||||
//! egress is a returned object, not a service. The re-evaluation
|
||||
//! threshold is three real implementations as trait impls in practice
|
||||
//! (the alktunnels ADR-004 posture).
|
||||
//!
|
||||
//! Contracts (POC-pinned, load-bearing API surface):
|
||||
//!
|
||||
//! - **Ownership rides the handle structurally** ([`TargetHandle`],
|
||||
//! [`BindTargetHandle`]): the payloads own their halves; any
|
||||
//! keep-alive state a substrate needs (e.g. a composed variant's
|
||||
//! channel session) rides inside the halves — a provider-side owner
|
||||
//! wrapper — so dropping the handle drops everything it dialed or
|
||||
//! accepted. No `_` bindings of owning values: the
|
||||
//! teardown-at-birth discipline, compiler-enforced by the shape.
|
||||
//! - **The egress speaks inner bytes**: [`Egress::send`] takes the
|
||||
//! payload only; [`Egress::next_reply`] returns `addr + data` — the
|
||||
//! frame codec lives on the wire side, applied exactly once
|
||||
//! ([`crate::wire`]).
|
||||
//! - **The readiness contract** ([`Egress::next_reply`]): `Pending`
|
||||
//! when no reply is ready — an instantly-ready `Ok(None)` starves a
|
||||
//! biased `select!` into livelock (POC #2's structural finding).
|
||||
//! - **The accept future owns its resources end-to-end**
|
||||
//! ([`BindAccept`]): the quiet wait's `select!` drop IS the
|
||||
//! client-gone teardown (ADR-004/ADR-007).
|
||||
//! - **Refusal is RFC-faithful**: every error type maps to RFC 1928
|
||||
//! reply codes via `to_reply_error()`; a dial/egress/accept failure
|
||||
//! is reply-code-shaped (never coerced to a channels-level error —
|
||||
//! session-level refusal stays channels-level, target-level refusal
|
||||
//! is in-stream).
|
||||
//!
|
||||
//! The readiness-contract behavioral test (an instant-`Ok(None)`
|
||||
//! egress must not starve the datagram stage) lands with the
|
||||
//! producer-associate task — it needs the stage. This module carries
|
||||
//! the error→reply-code mapping tables, the sync-lift factory
|
||||
//! constructor, and the compile-shape assertions.
|
||||
|
||||
use std::future::Future;
|
||||
use std::net::SocketAddr;
|
||||
use std::sync::Arc;
|
||||
|
||||
use bytes::Bytes;
|
||||
use futures::future::BoxFuture;
|
||||
use tokio::io::{AsyncRead, AsyncWrite};
|
||||
|
||||
pub use crate::rfc1928::util::target_addr::TargetAddr;
|
||||
use crate::rfc1928::ReplyError;
|
||||
|
||||
/// The dialed CONNECT target — boxed halves, `+ Sync` per the
|
||||
/// `ChannelPlan` payload bound (ADR-004). Ownership rides the handle
|
||||
/// structurally: the payloads own their connection halves; keep-alive
|
||||
/// state a substrate needs must ride inside the halves (a provider-side
|
||||
/// owner wrapper), never in a separate binding the compiler could see
|
||||
/// dropped early.
|
||||
pub struct TargetHandle {
|
||||
pub read: Box<dyn AsyncRead + Send + Sync + Unpin>,
|
||||
pub write: Box<dyn AsyncWrite + Send + Sync + Unpin>,
|
||||
}
|
||||
|
||||
impl std::fmt::Debug for TargetHandle {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
f.debug_struct("TargetHandle").finish_non_exhaustive()
|
||||
}
|
||||
}
|
||||
|
||||
/// The dial seam (ADR-004): the substrate dial function, injected at
|
||||
/// registration/assembly. Called once per CONNECT at the command-read
|
||||
/// interception point with the RFC 1928 target the client requested
|
||||
/// (domain form passes through unresolved — ADR-012; resolution is the
|
||||
/// provider's job). Per-target refusal is the returned
|
||||
/// [`SocksDialError`] (the producer replies with the mapped RFC code —
|
||||
/// no phantom data plane); session-level refusal stays channels-level.
|
||||
pub type DialFn = Arc<
|
||||
dyn Fn(TargetAddr) -> BoxFuture<'static, Result<TargetHandle, SocksDialError>> + Send + Sync,
|
||||
>;
|
||||
|
||||
/// A dial failure, mapped to RFC 1928 reply codes by
|
||||
/// [`SocksDialError::to_reply_error`] (never a generic error —
|
||||
/// AGENTS.md convention 2's faithful `ReplyError` mapping).
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum SocksDialError {
|
||||
#[error("connection refused: {0}")]
|
||||
Refused(String),
|
||||
#[error("network unreachable: {0}")]
|
||||
NetworkUnreachable(String),
|
||||
#[error("host unreachable: {0}")]
|
||||
HostUnreachable(String),
|
||||
#[error("connection not allowed by ruleset: {0}")]
|
||||
NotAllowed(String),
|
||||
#[error("general failure: {0}")]
|
||||
General(String),
|
||||
}
|
||||
|
||||
impl SocksDialError {
|
||||
/// The RFC 1928 reply code for the reply the producer sends on this
|
||||
/// refusal. `Refused` → `ConnectionRefused` (0x05),
|
||||
/// `NetworkUnreachable` → `NetworkUnreachable` (0x03),
|
||||
/// `HostUnreachable` → `HostUnreachable` (0x04), `NotAllowed` →
|
||||
/// `ConnectionNotAllowed` (0x02), `General` → `GeneralFailure`
|
||||
/// (0x01) — the POC #1 mapping, table-tested in this module.
|
||||
pub fn to_reply_error(&self) -> ReplyError {
|
||||
match self {
|
||||
Self::Refused(_) => ReplyError::ConnectionRefused,
|
||||
Self::NetworkUnreachable(_) => ReplyError::NetworkUnreachable,
|
||||
Self::HostUnreachable(_) => ReplyError::HostUnreachable,
|
||||
Self::NotAllowed(_) => ReplyError::ConnectionNotAllowed,
|
||||
Self::General(_) => ReplyError::GeneralFailure,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// One association's egress failure. Mapped to RFC reply codes only at
|
||||
/// association setup ([`EgressError::to_reply_error`] — an egress that
|
||||
/// cannot start refuses the ASSOCIATE pre-reply); per-datagram failures
|
||||
/// are not per-datagram refusals — the producer stage ends the
|
||||
/// association on an `Err` from the seam (the lossy-datagram posture).
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum EgressError {
|
||||
#[error("egress refused: {0}")]
|
||||
Refused(String),
|
||||
#[error("egress unreachable: {0}")]
|
||||
Unreachable(String),
|
||||
#[error("egress general failure: {0}")]
|
||||
General(String),
|
||||
}
|
||||
|
||||
impl EgressError {
|
||||
/// The RFC 1928 reply code for the ASSOCIATE refusal: `Refused` →
|
||||
/// `ConnectionRefused` (0x05), `Unreachable` →
|
||||
/// `NetworkUnreachable` (0x03), `General` → `GeneralFailure`
|
||||
/// (0x01) — the POC #2 mapping, table-tested in this module.
|
||||
pub fn to_reply_error(&self) -> ReplyError {
|
||||
match self {
|
||||
Self::Refused(_) => ReplyError::ConnectionRefused,
|
||||
Self::Unreachable(_) => ReplyError::NetworkUnreachable,
|
||||
Self::General(_) => ReplyError::GeneralFailure,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The per-association datagram egress — the ASSOCIATE seam's returned
|
||||
/// object and the module's one trait (ADR-004: the factory is
|
||||
/// function-shaped; the instance it returns is a trait object,
|
||||
/// `Box<dyn Egress + Send>`, owned by the producer's datagram stage —
|
||||
/// stage exit drops it, so egress state never outlives the channel).
|
||||
///
|
||||
/// The stage never sees which substrate backs the association
|
||||
/// (convention 7 at the egress layer).
|
||||
pub trait Egress {
|
||||
/// Egress one datagram to `addr`. Payload-only: the producer stage
|
||||
/// decoded the wire frame and hands the bare datagram here — the
|
||||
/// egress never sees a length prefix (framing is applied exactly
|
||||
/// once, by the wire layer). A zero-length payload is a legal
|
||||
/// empty datagram, not a signal of any kind (the F-2 invariant).
|
||||
fn send(&mut self, addr: TargetAddr, payload: Bytes) -> BoxFuture<'_, Result<(), EgressError>>;
|
||||
|
||||
/// Yield the next inbound datagram's *inner* bytes (`addr + data`
|
||||
/// — the RFC address encoding followed by the payload, exactly
|
||||
/// [`TargetAddr::to_be_bytes()`] + data). The length prefix is the
|
||||
/// wire layer's, added by the producer's frame writer — an egress
|
||||
/// that prefixes again is the double-framing bug class POC #2
|
||||
/// caught.
|
||||
///
|
||||
/// # The readiness contract (load-bearing API surface)
|
||||
///
|
||||
/// Must return `Pending` when no reply is ready. An
|
||||
/// instantly-ready `Ok(None)` starves a biased `select!` into
|
||||
/// livelock: the stage parks on `next_reply` ∥ the channel read,
|
||||
/// and a hot `Ok(None)` never yields to the channel branch (nor to
|
||||
/// the runtime's other tasks on a current-thread scheduler).
|
||||
/// `Ok(None)` is tolerated as a spurious wakeup (the stage
|
||||
/// re-parks), not as a readiness signal — implement the wait as a
|
||||
/// real async park. An `Err` ends the association (the stage
|
||||
/// returns; the egress drops with it).
|
||||
fn next_reply(&mut self) -> BoxFuture<'_, Result<Option<Bytes>, EgressError>>;
|
||||
}
|
||||
|
||||
/// The egress seam (ADR-004): the per-association egress factory — the
|
||||
/// [`DialFn`] analogue for the datagram stage, injected at
|
||||
/// registration/assembly. Called once per UDP ASSOCIATE; the stage owns
|
||||
/// the returned egress; stage exit drops it.
|
||||
pub type EgressFactory =
|
||||
Arc<dyn Fn() -> BoxFuture<'static, Result<Box<dyn Egress + Send>, EgressError>> + Send + Sync>;
|
||||
|
||||
/// A convenience constructor: lift a sync factory into the async shape
|
||||
/// the producer consumes. The sync result is captured and delivered
|
||||
/// through the boxed future verbatim (an `Err` still refuses the
|
||||
/// ASSOCIATE with the mapped code).
|
||||
pub fn egress_factory<F>(f: F) -> EgressFactory
|
||||
where
|
||||
F: Fn() -> Result<Box<dyn Egress + Send>, EgressError> + Send + Sync + 'static,
|
||||
{
|
||||
Arc::new(move || {
|
||||
let result = f();
|
||||
Box::pin(async move { result })
|
||||
})
|
||||
}
|
||||
|
||||
/// The one accepted inbound connection's handle — reply#2's BND.ADDR
|
||||
/// carrier plus the data-plane halves. Ownership rides the handle
|
||||
/// structurally: the payloads own their accepted halves; keep-alive
|
||||
/// state a substrate needs (e.g. a composed variant's channel session)
|
||||
/// rides inside the halves (a provider-side owner wrapper), so the
|
||||
/// compiler enforces the teardown contract — no `_` bindings of owning
|
||||
/// values (the third teardown-at-birth witness, ADR-004).
|
||||
pub struct BindTargetHandle {
|
||||
/// The accepted connection's peer address (reply#2's BND.ADDR —
|
||||
/// the RFC's "client IP and port of the incoming connection").
|
||||
pub peer: TargetAddr,
|
||||
pub read: Box<dyn AsyncRead + Send + Sync + Unpin>,
|
||||
pub write: Box<dyn AsyncWrite + Send + Sync + Unpin>,
|
||||
}
|
||||
|
||||
impl std::fmt::Debug for BindTargetHandle {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
f.debug_struct("BindTargetHandle")
|
||||
.field("peer", &self.peer)
|
||||
.finish_non_exhaustive()
|
||||
}
|
||||
}
|
||||
|
||||
/// One BIND session's accept source. `bind_addr` is the address
|
||||
/// reply#1 advertises (BND.ADDR — the real bound listener on the local
|
||||
/// variant, the advertised address on the composed one); the accept
|
||||
/// future is awaited exactly once, in the producer's quiet-wait phase.
|
||||
///
|
||||
/// **The accept future owns its resources end-to-end** (ADR-004/
|
||||
/// ADR-007): whatever the provider opened (the listener, the tunnel)
|
||||
/// lives inside the future, so the quiet wait's `select!` drop of that
|
||||
/// future IS the client-gone teardown — the listener closes when the
|
||||
/// client vanishes, with no extra teardown path.
|
||||
pub struct BindAccept {
|
||||
bind_addr: SocketAddr,
|
||||
accept: BoxFuture<'static, Result<BindTargetHandle, BindAcceptError>>,
|
||||
}
|
||||
|
||||
impl BindAccept {
|
||||
/// Build the session's accept source: `bind_addr` for reply#1, the
|
||||
/// one-inbound-accept future (owned end-to-end — see the type
|
||||
/// docs).
|
||||
pub fn new(
|
||||
bind_addr: SocketAddr,
|
||||
accept: impl Future<Output = Result<BindTargetHandle, BindAcceptError>> + Send + 'static,
|
||||
) -> Self {
|
||||
Self {
|
||||
bind_addr,
|
||||
accept: Box::pin(accept),
|
||||
}
|
||||
}
|
||||
|
||||
/// The address reply#1 advertises (BND.ADDR).
|
||||
pub fn bind_addr(&self) -> SocketAddr {
|
||||
self.bind_addr
|
||||
}
|
||||
|
||||
/// Await the one inbound accept (the quiet wait's accept branch —
|
||||
/// exactly once; the future is consumed).
|
||||
pub async fn accept(self) -> Result<BindTargetHandle, BindAcceptError> {
|
||||
self.accept.await
|
||||
}
|
||||
}
|
||||
|
||||
impl std::fmt::Debug for BindAccept {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
f.debug_struct("BindAccept")
|
||||
.field("bind_addr", &self.bind_addr)
|
||||
.finish_non_exhaustive()
|
||||
}
|
||||
}
|
||||
|
||||
/// Accept-stage failures (post-reply#1): the producer maps them to RFC
|
||||
/// reply codes for reply#2 — no data plane on refusal.
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum BindAcceptError {
|
||||
#[error("accept refused: {0}")]
|
||||
Refused(String),
|
||||
#[error("accept not allowed: {0}")]
|
||||
NotAllowed(String),
|
||||
#[error("accept general failure: {0}")]
|
||||
General(String),
|
||||
}
|
||||
|
||||
impl BindAcceptError {
|
||||
/// The RFC 1928 reply code for reply#2 on this refusal: `Refused` →
|
||||
/// `ConnectionRefused` (0x05), `NotAllowed` →
|
||||
/// `ConnectionNotAllowed` (0x02), `General` → `GeneralFailure`
|
||||
/// (0x01) — the POC #5 mapping, table-tested in this module.
|
||||
pub fn to_reply_error(&self) -> ReplyError {
|
||||
match self {
|
||||
Self::Refused(_) => ReplyError::ConnectionRefused,
|
||||
Self::NotAllowed(_) => ReplyError::ConnectionNotAllowed,
|
||||
Self::General(_) => ReplyError::GeneralFailure,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Factory failures (pre-reply#1): the producer maps them to RFC reply
|
||||
/// codes for reply#1 — no phantom listener.
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum BindAcceptFactoryError {
|
||||
#[error("bind refused: {0}")]
|
||||
Refused(String),
|
||||
#[error("bind not allowed: {0}")]
|
||||
NotAllowed(String),
|
||||
#[error("bind resource shortage: {0}")]
|
||||
Shortage(String),
|
||||
#[error("bind general failure: {0}")]
|
||||
General(String),
|
||||
}
|
||||
|
||||
impl BindAcceptFactoryError {
|
||||
/// The RFC 1928 reply code for reply#1 on this refusal: `Refused` →
|
||||
/// `ConnectionRefused` (0x05), `NotAllowed` →
|
||||
/// `ConnectionNotAllowed` (0x02), `Shortage`/`General` →
|
||||
/// `GeneralFailure` (0x01 — the RFC's catch-all; a resource
|
||||
/// shortage has no dedicated code) — the POC #5 mapping,
|
||||
/// table-tested in this module.
|
||||
pub fn to_reply_error(&self) -> ReplyError {
|
||||
match self {
|
||||
Self::Refused(_) => ReplyError::ConnectionRefused,
|
||||
Self::NotAllowed(_) => ReplyError::ConnectionNotAllowed,
|
||||
Self::Shortage(_) => ReplyError::GeneralFailure,
|
||||
Self::General(_) => ReplyError::GeneralFailure,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The accept seam (ADR-004): the per-BIND accept factory — the
|
||||
/// [`DialFn`]'s mirror, injected at registration/assembly. Called once
|
||||
/// per TCP BIND with the requested target. The requested port is the
|
||||
/// only field a real bind can honor (domains are meaningless as bind
|
||||
/// addresses); the address policy is the factory's (ADR-006/ADR-012:
|
||||
/// the callbacks refuse — RFC reply codes).
|
||||
///
|
||||
/// Failure mapping is two-stage (ADR-007): factory failure
|
||||
/// ([`BindAcceptFactoryError`]) → reply#1 with the mapped code;
|
||||
/// accept-stage failure ([`BindAcceptError`]) → reply#2 with the
|
||||
/// mapped code.
|
||||
pub type BindAcceptFactory = Arc<
|
||||
dyn Fn(TargetAddr) -> BoxFuture<'static, Result<BindAccept, BindAcceptFactoryError>>
|
||||
+ Send
|
||||
+ Sync,
|
||||
>;
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::collections::VecDeque;
|
||||
|
||||
fn domain(port: u16) -> TargetAddr {
|
||||
TargetAddr::Domain("example.com".to_owned(), port)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn dial_error_maps_all_variants_to_rfc_reply_codes() {
|
||||
for (err, code) in [
|
||||
(
|
||||
SocksDialError::Refused("x".into()),
|
||||
ReplyError::ConnectionRefused,
|
||||
),
|
||||
(
|
||||
SocksDialError::NetworkUnreachable("x".into()),
|
||||
ReplyError::NetworkUnreachable,
|
||||
),
|
||||
(
|
||||
SocksDialError::HostUnreachable("x".into()),
|
||||
ReplyError::HostUnreachable,
|
||||
),
|
||||
(
|
||||
SocksDialError::NotAllowed("x".into()),
|
||||
ReplyError::ConnectionNotAllowed,
|
||||
),
|
||||
(
|
||||
SocksDialError::General("x".into()),
|
||||
ReplyError::GeneralFailure,
|
||||
),
|
||||
] {
|
||||
assert_eq!(
|
||||
err.to_reply_error().as_u8(),
|
||||
code.as_u8(),
|
||||
"{err} must map to {code:?}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn egress_error_maps_all_variants_to_rfc_reply_codes() {
|
||||
for (err, code) in [
|
||||
(
|
||||
EgressError::Refused("x".into()),
|
||||
ReplyError::ConnectionRefused,
|
||||
),
|
||||
(
|
||||
EgressError::Unreachable("x".into()),
|
||||
ReplyError::NetworkUnreachable,
|
||||
),
|
||||
(EgressError::General("x".into()), ReplyError::GeneralFailure),
|
||||
] {
|
||||
assert_eq!(
|
||||
err.to_reply_error().as_u8(),
|
||||
code.as_u8(),
|
||||
"{err} must map to {code:?}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn bind_accept_error_maps_all_variants_to_rfc_reply_codes() {
|
||||
for (err, code) in [
|
||||
(
|
||||
BindAcceptError::Refused("x".into()),
|
||||
ReplyError::ConnectionRefused,
|
||||
),
|
||||
(
|
||||
BindAcceptError::NotAllowed("x".into()),
|
||||
ReplyError::ConnectionNotAllowed,
|
||||
),
|
||||
(
|
||||
BindAcceptError::General("x".into()),
|
||||
ReplyError::GeneralFailure,
|
||||
),
|
||||
] {
|
||||
assert_eq!(
|
||||
err.to_reply_error().as_u8(),
|
||||
code.as_u8(),
|
||||
"{err} must map to {code:?}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn bind_accept_factory_error_maps_all_variants_to_rfc_reply_codes() {
|
||||
for (err, code) in [
|
||||
(
|
||||
BindAcceptFactoryError::Refused("x".into()),
|
||||
ReplyError::ConnectionRefused,
|
||||
),
|
||||
(
|
||||
BindAcceptFactoryError::NotAllowed("x".into()),
|
||||
ReplyError::ConnectionNotAllowed,
|
||||
),
|
||||
(
|
||||
BindAcceptFactoryError::Shortage("x".into()),
|
||||
ReplyError::GeneralFailure,
|
||||
),
|
||||
(
|
||||
BindAcceptFactoryError::General("x".into()),
|
||||
ReplyError::GeneralFailure,
|
||||
),
|
||||
] {
|
||||
assert_eq!(
|
||||
err.to_reply_error().as_u8(),
|
||||
code.as_u8(),
|
||||
"{err} must map to {code:?}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
struct QueueEgress {
|
||||
replies: VecDeque<Bytes>,
|
||||
}
|
||||
|
||||
impl Egress for QueueEgress {
|
||||
fn send(
|
||||
&mut self,
|
||||
_addr: TargetAddr,
|
||||
payload: Bytes,
|
||||
) -> BoxFuture<'_, Result<(), EgressError>> {
|
||||
Box::pin(async move {
|
||||
self.replies.push_back(payload);
|
||||
Ok(())
|
||||
})
|
||||
}
|
||||
|
||||
fn next_reply(&mut self) -> BoxFuture<'_, Result<Option<Bytes>, EgressError>> {
|
||||
Box::pin(async move { Ok(self.replies.pop_front()) })
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn egress_factory_lifts_a_sync_factory_into_the_async_shape() {
|
||||
let factory: EgressFactory = egress_factory(|| {
|
||||
Ok(Box::new(QueueEgress {
|
||||
replies: VecDeque::new(),
|
||||
}) as Box<dyn Egress + Send>)
|
||||
});
|
||||
let mut egress = factory().await.expect("Ok lifts verbatim");
|
||||
egress
|
||||
.send(domain(53), Bytes::from_static(b"query"))
|
||||
.await
|
||||
.expect("send succeeds");
|
||||
let reply = egress.next_reply().await.expect("next_reply succeeds");
|
||||
assert_eq!(
|
||||
reply.expect("the queued reply is ready"),
|
||||
Bytes::from_static(b"query")
|
||||
);
|
||||
assert!(egress.next_reply().await.expect("drained").is_none());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn egress_factory_delivers_a_sync_err_verbatim() {
|
||||
let factory: EgressFactory =
|
||||
egress_factory(|| Err(EgressError::Unreachable("no egress".into())));
|
||||
let err = match factory().await {
|
||||
Err(err) => err,
|
||||
Ok(_) => panic!("expected the sync Err to lift verbatim"),
|
||||
};
|
||||
assert_eq!(
|
||||
err.to_reply_error().as_u8(),
|
||||
ReplyError::NetworkUnreachable.as_u8()
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn dial_fn_shape_accepts_a_closure_and_yields_the_handle() {
|
||||
let dial: DialFn = Arc::new(|target: TargetAddr| {
|
||||
Box::pin(async move {
|
||||
assert_eq!(target, domain(443));
|
||||
let (client_read, target_read) = tokio::io::duplex(64);
|
||||
let (target_write, client_write) = tokio::io::duplex(64);
|
||||
std::mem::forget((client_read, client_write));
|
||||
Ok(TargetHandle {
|
||||
read: Box::new(target_read) as Box<dyn AsyncRead + Send + Sync + Unpin>,
|
||||
write: Box::new(target_write) as Box<dyn AsyncWrite + Send + Sync + Unpin>,
|
||||
})
|
||||
})
|
||||
});
|
||||
let handle = dial(domain(443)).await.expect("the dial resolves");
|
||||
let _ = (&handle.read, &handle.write);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn bind_accept_factory_shape_yields_bind_addr_then_the_handle() {
|
||||
let factory: BindAcceptFactory = Arc::new(|_target: TargetAddr| {
|
||||
Box::pin(async move {
|
||||
Ok(BindAccept::new(
|
||||
SocketAddr::from(([127, 0, 0, 1], 2121)),
|
||||
async {
|
||||
let (client_read, target_read) = tokio::io::duplex(64);
|
||||
let (target_write, client_write) = tokio::io::duplex(64);
|
||||
std::mem::forget((client_read, client_write));
|
||||
Ok(BindTargetHandle {
|
||||
peer: TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 7], 44444))),
|
||||
read: Box::new(target_read) as Box<dyn AsyncRead + Send + Sync + Unpin>,
|
||||
write: Box::new(target_write)
|
||||
as Box<dyn AsyncWrite + Send + Sync + Unpin>,
|
||||
})
|
||||
},
|
||||
))
|
||||
})
|
||||
});
|
||||
let accept = factory(domain(2121)).await.expect("the factory resolves");
|
||||
assert_eq!(accept.bind_addr(), SocketAddr::from(([127, 0, 0, 1], 2121)));
|
||||
let handle = accept.accept().await.expect("the accept resolves");
|
||||
assert_eq!(
|
||||
handle.peer,
|
||||
TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 7], 44444)))
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn seam_types_have_the_poc_compile_shapes() {
|
||||
fn assert_send<T: Send>() {}
|
||||
fn assert_send_sync<T: Send + Sync>() {}
|
||||
fn assert_unpin<T: Unpin>() {}
|
||||
|
||||
assert_send_sync::<TargetHandle>();
|
||||
assert_send_sync::<BindTargetHandle>();
|
||||
assert_unpin::<TargetHandle>();
|
||||
assert_unpin::<BindTargetHandle>();
|
||||
|
||||
assert_send_sync::<DialFn>();
|
||||
assert_send_sync::<EgressFactory>();
|
||||
assert_send_sync::<BindAcceptFactory>();
|
||||
|
||||
assert_send::<BindAccept>();
|
||||
assert_unpin::<BindAccept>();
|
||||
|
||||
assert_send_sync::<SocksDialError>();
|
||||
assert_send_sync::<EgressError>();
|
||||
assert_send_sync::<BindAcceptError>();
|
||||
assert_send_sync::<BindAcceptFactoryError>();
|
||||
}
|
||||
}
|
||||
@@ -29,6 +29,10 @@ pub mod params;
|
||||
pub mod producer;
|
||||
pub mod wire;
|
||||
|
||||
pub use egress::{
|
||||
egress_factory, BindAccept, BindAcceptError, BindAcceptFactory, BindAcceptFactoryError,
|
||||
BindTargetHandle, DialFn, Egress, EgressError, EgressFactory, SocksDialError, TargetHandle,
|
||||
};
|
||||
pub use error::{
|
||||
ReplyError, SocksError, SocksEstablishmentReason, SocksOpenError, SocksServerError,
|
||||
SocksSessionError,
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
---
|
||||
id: implementation/core-seam-types
|
||||
name: Core — the three seam types (DialFn / Egress / BindAccept families)
|
||||
status: pending
|
||||
status: completed
|
||||
depends_on: [implementation/core-crate-deps]
|
||||
scope: moderate
|
||||
risk: low
|
||||
@@ -76,14 +76,14 @@ with the producer-associate task (it needs the stage).
|
||||
|
||||
## Acceptance Criteria
|
||||
|
||||
- [ ] All three seam families implemented with the POC shapes
|
||||
- [ ] Every error type's `to_reply_error()` mapping table-tested
|
||||
- [x] All three seam families implemented with the POC shapes
|
||||
- [x] Every error type's `to_reply_error()` mapping table-tested
|
||||
(all variants, RFC-faithful codes)
|
||||
- [ ] Readiness contract documented on `Egress::next_reply`
|
||||
- [ ] Ownership-rides-the-handle shapes (no `_` bindings; handles
|
||||
- [x] Readiness contract documented on `Egress::next_reply`
|
||||
- [x] Ownership-rides-the-handle shapes (no `_` bindings; handles
|
||||
own their payloads structurally)
|
||||
- [ ] No substrate types named anywhere in the module
|
||||
- [ ] `cargo test` green; clippy/fmt clean; both wasm checks clean
|
||||
- [x] No substrate types named anywhere in the module
|
||||
- [x] `cargo test` green; clippy/fmt clean; both wasm checks clean
|
||||
|
||||
## References
|
||||
|
||||
@@ -99,8 +99,86 @@ with the producer-associate task (it needs the stage).
|
||||
|
||||
## Notes
|
||||
|
||||
> To be filled by implementation agent
|
||||
> Filled by the implementation agent (2026-09-23).
|
||||
|
||||
- `src/egress.rs`: all three seam families per the POC shapes.
|
||||
- **Dial (CONNECT)**: `TargetHandle {read, write}` (boxed halves
|
||||
`Send + Sync + Unpin` — the `+ Sync` per the `ChannelPlan` payload
|
||||
bound), `DialFn` (`Arc<dyn Fn(TargetAddr) -> BoxFuture<'static,
|
||||
Result<TargetHandle, SocksDialError>> + Send + Sync>`), and
|
||||
`SocksDialError` with the POC #1 five-variant mapping
|
||||
(`Refused`→`ConnectionRefused`, `NetworkUnreachable`→
|
||||
`NetworkUnreachable`, `HostUnreachable`→`HostUnreachable`,
|
||||
`NotAllowed`→`ConnectionNotAllowed`, `General`→`GeneralFailure`).
|
||||
Deliberate change from the POC shape: the seam takes the typed
|
||||
`TargetAddr`, not the POC's `{host, port}` JSON stand-in (the JSON
|
||||
shim existed only because the POCs predated the forked
|
||||
vocabulary).
|
||||
- **Egress (ASSOCIATE)**: the one trait, `Egress` (`send(addr,
|
||||
payload)` payload-only + `next_reply()` returning inner bytes,
|
||||
`BoxFuture` methods → object-safe `Box<dyn Egress + Send>`),
|
||||
`EgressFactory`, `EgressError` (three variants), and the
|
||||
`egress_factory` sync-lift convenience constructor. The readiness
|
||||
contract is documented on `Egress::next_reply` in rustdoc
|
||||
(`Pending` when nothing is ready; the instant-`Ok(None)` livelock
|
||||
hazard and the spurious-wakeup tolerance both stated); the
|
||||
behavioral test lands with the producer-associate task as
|
||||
planned.
|
||||
- **Accept (BIND)**: `BindAccept {bind_addr, accept}` (the
|
||||
`accept()` consumer awaits the boxed future exactly once; the
|
||||
future owns its resources end-to-end — the quiet wait's drop IS
|
||||
the client-gone teardown), `BindTargetHandle {peer, read, write}`
|
||||
(ownership structural; no `_` owner field — keep-alive state
|
||||
rides inside the halves as a provider-side owner wrapper, which
|
||||
is the substrate-agnostic way to carry the POC's `_session`
|
||||
without naming a substrate type in the protocol layer),
|
||||
`BindAcceptFactory`, and the two-stage error pair
|
||||
(`BindAcceptFactoryError` pre-reply#1 → reply#1;
|
||||
`BindAcceptError` post-reply#1 → reply#2; `Shortage`→
|
||||
`GeneralFailure` — no dedicated RFC code for shortage).
|
||||
- No trait for the seams themselves (ADR-004); the re-evaluation
|
||||
threshold (three real trait impls) is stated in the module docs.
|
||||
- No substrate type named anywhere in the module (grep-verified:
|
||||
no `TcpStream`/`TcpListener`/`UdpSocket`/`BiStream`/
|
||||
`ChannelClient`/`TunnelSession`/`tokio::net`).
|
||||
- lib.rs re-exports the full public surface (convention 15), plus
|
||||
`TargetAddr` re-exported through the module (the seam signatures
|
||||
need it in scope).
|
||||
- Deviations from the POC code, all deliberate: `BindTargetHandle`
|
||||
drops the POC's `_session: Option<TunnelSession>` field (substrate
|
||||
types cannot appear here — ownership rides the halves; see the
|
||||
dial bullet), `BindAccept` exposes `bind_addr()` as a getter
|
||||
instead of a public field (the accept future is already private —
|
||||
the struct now encapsulates both), and the test halves use
|
||||
`tokio::io::duplex` (test-only; the POCs' `std::io` empty/sink do
|
||||
not implement tokio's traits).
|
||||
|
||||
## Summary
|
||||
|
||||
> To be filled on completion
|
||||
> Filled on completion (2026-09-23).
|
||||
|
||||
Implemented `src/egress.rs` — the three I/O-plane seam types (ADR-004),
|
||||
the protocol layer's inversion point: the dial seam (`TargetHandle`
|
||||
boxed halves, `DialFn`, `SocksDialError` with the five-variant
|
||||
RFC-faithful mapping), the egress seam (the one trait, `Egress` —
|
||||
`send` payload-only, `next_reply` returning inner bytes with the
|
||||
load-bearing readiness contract documented in rustdoc; `EgressFactory`
|
||||
+ the `egress_factory` sync-lift constructor; `EgressError`), and the
|
||||
accept seam (`BindAccept` with the end-to-end-owned accept future,
|
||||
`BindTargetHandle`, `BindAcceptFactory`, and the two-stage error pair
|
||||
`BindAcceptFactoryError`/`BindAcceptError`). Function-not-trait at the
|
||||
three factories (the re-evaluation threshold noted in the module
|
||||
docs); ownership rides the handle structurally (no `_` bindings; the
|
||||
composed keep-alive owner rides inside the halves provider-side). 9
|
||||
new unit tests: the four error→reply-code mapping tables (all
|
||||
variants, byte-compared against `ReplyError::as_u8`), the sync-lift
|
||||
factory (Ok + Err paths, exercised through the trait object), the
|
||||
closure-shaped `DialFn`/`BindAcceptFactory` compile shapes, and the
|
||||
Send/Sync/Unpin compile-shape assertions. lib.rs re-exports the
|
||||
surface.
|
||||
|
||||
Verification: `cargo test` (113 green: 76 lib + 37 rfc1928), `cargo
|
||||
test --all-features` green, `cargo clippy --all-targets -- -D
|
||||
warnings` clean, `cargo fmt --check` clean, `cargo doc --no-deps` zero
|
||||
warnings, both wasm checks (`cargo check` + `cargo clippy` on
|
||||
`wasm32-unknown-unknown`) clean.
|
||||
Reference in new issue
Block a user