refactor(core): implement BidiStreamSource trait + AuthContext::anonymous (ADR-070, REQ-CORE-01/02/03)

Extract stream-yield ops from Connection into a BidiStreamSource trait;
Connection now holds Box<dyn BidiStreamSource> instead of the closed
ConnectionKind enum. Three crate-private impls wrap the existing
constructors: QuinnBidiStreamSource (feature quinn), IrohBidiStreamSource
(feature iroh), StreamBidiStreamSource (no gate, yield-once per ADR-065).
Public Connection API is preserved verbatim — handlers dispatch through
the trait object transparently.

The StreamBidiStreamSource::close impl prefixes code/reason with _ and
documents why they're ignored (the drop is the close — ADR-065). This
resolves the ADR-065 leftover clippy warning under --no-default-features
(REQ-CORE-02).

Add AuthContext::anonymous(alpn) convenience constructor (REQ-CORE-03):
sets identity/fingerprint/remote_addr to None, only alpn is populated.
Removes the four-None-field literal that recurred in handler POCs/tests.

Verified: cargo test + clippy clean across default, --no-default-features,
--features iroh, and --all-features; cargo check -p alknet-call succeeds
unchanged; cargo fmt --check passes.
This commit is contained in:
glm-5.2 committed 2026-07-12 09:52:08 +00:00
1 parent f8f5f27ce0
commit 60cce228d2
3 files changed
+178 -66

No files matched your search

+26
View File
@@ -80,6 +80,23 @@ pub struct AuthContext {
pub tls_client_fingerprint: Option<String>, pub tls_client_fingerprint: Option<String>,
} }
impl AuthContext {
/// Construct an `AuthContext` with no identity, no fingerprint, and no
/// remote address — only the ALPN is set. For POCs, tests, and handlers
/// that don't require auth. Not gated behind a `test-utils` feature:
/// it's a plain `pub fn` useful for any caller that constructs an
/// `AuthContext` outside the endpoint's resolution path. The name is
/// honest about the semantics: no identity, no fingerprint.
pub fn anonymous(alpn: impl Into<Vec<u8>>) -> Self {
Self {
identity: None,
alpn: alpn.into(),
remote_addr: None,
tls_client_fingerprint: None,
}
}
}
pub trait IdentityProvider: Send + Sync + 'static { pub trait IdentityProvider: Send + Sync + 'static {
fn resolve_from_fingerprint(&self, fingerprint: &str) -> Option<Identity>; fn resolve_from_fingerprint(&self, fingerprint: &str) -> Option<Identity>;
fn resolve_from_token(&self, token: &AuthToken) -> Option<Identity>; fn resolve_from_token(&self, token: &AuthToken) -> Option<Identity>;
@@ -220,6 +237,15 @@ mod tests {
assert!(cloned.identity.is_none()); assert!(cloned.identity.is_none());
} }
#[test]
fn auth_context_anonymous_sets_alpn_only() {
let ctx = AuthContext::anonymous(b"alknet/test");
assert_eq!(ctx.alpn, b"alknet/test");
assert!(ctx.identity.is_none());
assert!(ctx.remote_addr.is_none());
assert!(ctx.tls_client_fingerprint.is_none());
}
#[test] #[test]
fn fingerprint_resolution_known_returns_some() { fn fingerprint_resolution_known_returns_some() {
let (provider, _) = make_provider(config_with_peer_entry("worker-a", "SHA256:abc123")); let (provider, _) = make_provider(config_with_peer_entry("worker-a", "SHA256:abc123"));
+151 -65
View File
@@ -355,21 +355,157 @@ impl AsyncRead for RecvStream {
} }
} }
enum ConnectionKind { /// Yield bidirectional streams to a `Connection`. Downstream crates implement
#[cfg(feature = "quinn")] /// this trait to add connection shapes (channels, a future transport, a test
Quinn(quinn::Connection), /// double beyond the `from_stream` case) without editing `alknet-core`. See
#[cfg(feature = "iroh")] /// ADR-070 for the full rationale and ADR-065 for the yield-once contract the
Iroh(iroh::endpoint::Connection), /// `StreamBidiStreamSource` impl preserves.
Stream(StreamConn), #[async_trait]
pub trait BidiStreamSource: Send + Sync + 'static {
/// Yield the next bidirectional stream this connection provides.
///
/// Transport semantics (carried from ADR-065):
/// - QUIC (quinn/iroh): returns a new bidi stream on each call,
/// `ConnectionClosed` when the underlying connection closes.
/// - Single-stream (TCP+TLS, SSH channel, WebTransport stream, wasm):
/// yields the underlying stream on the first call, then
/// `ConnectionClosed` on all subsequent calls.
/// - Channels: yields one bidi stream per channel, `ConnectionClosed`
/// when the channels connection closes.
async fn accept_bi(&self) -> Result<(SendStream, RecvStream), StreamError>;
/// Open a bidirectional stream to the peer.
///
/// Single-stream sources return `StreamClosed` (a single stream cannot
/// open new application streams — ADR-065). QUIC and channels sources
/// open new streams.
async fn open_bi(&self) -> Result<(SendStream, RecvStream), StreamError>;
/// The peer's address, if available. Informational (NAT/proxy).
fn remote_addr(&self) -> Option<SocketAddr>;
/// Close the connection. The `code`/`reason` args are QUIC application-
/// level close codes; non-QUIC sources ignore them (the drop is the
/// close — ADR-065 §"Negative"). See ADR-070 §"REQ-CORE-02" for the
/// rationale for keeping the QUIC-shaped signature on the trait.
fn close(&self, code: u32, reason: &str);
} }
struct StreamConn { /// QUIC-backed `BidiStreamSource` (quinn). Crate-private; constructed via
/// `Connection::from_quinn` / `from_quinn_with_alpn` (feature `quinn`).
#[cfg(feature = "quinn")]
struct QuinnBidiStreamSource {
conn: quinn::Connection,
}
#[cfg(feature = "quinn")]
#[async_trait]
impl BidiStreamSource for QuinnBidiStreamSource {
async fn accept_bi(&self) -> Result<(SendStream, RecvStream), StreamError> {
let (send, recv) = self
.conn
.accept_bi()
.await
.map_err(map_quinn_connection_error)?;
Ok((SendStream::from_quinn(send), RecvStream::from_quinn(recv)))
}
async fn open_bi(&self) -> Result<(SendStream, RecvStream), StreamError> {
let (send, recv) = self
.conn
.open_bi()
.await
.map_err(map_quinn_connection_error)?;
Ok((SendStream::from_quinn(send), RecvStream::from_quinn(recv)))
}
fn remote_addr(&self) -> Option<SocketAddr> {
Some(self.conn.remote_address())
}
fn close(&self, code: u32, reason: &str) {
let code = quinn::VarInt::from(code);
self.conn.close(code, reason.as_bytes());
}
}
/// QUIC-backed `BidiStreamSource` (iroh). Crate-private; constructed via
/// `Connection::from_iroh` (feature `iroh`).
#[cfg(feature = "iroh")]
struct IrohBidiStreamSource {
conn: iroh::endpoint::Connection,
}
#[cfg(feature = "iroh")]
#[async_trait]
impl BidiStreamSource for IrohBidiStreamSource {
async fn accept_bi(&self) -> Result<(SendStream, RecvStream), StreamError> {
let (send, recv) = self
.conn
.accept_bi()
.await
.map_err(map_iroh_connection_error)?;
Ok((SendStream::from_iroh(send), RecvStream::from_iroh(recv)))
}
async fn open_bi(&self) -> Result<(SendStream, RecvStream), StreamError> {
let (send, recv) = self
.conn
.open_bi()
.await
.map_err(map_iroh_connection_error)?;
Ok((SendStream::from_iroh(send), RecvStream::from_iroh(recv)))
}
fn remote_addr(&self) -> Option<SocketAddr> {
None
}
fn close(&self, code: u32, reason: &str) {
let code = iroh::endpoint::VarInt::from(code);
self.conn.close(code, reason.as_bytes());
}
}
/// Single-stream `BidiStreamSource` (TCP+TLS, SSH channel, WebTransport
/// stream, wasm stream — ADR-065). Crate-private; constructed via
/// `Connection::from_stream` / `from_bidi` (no feature gate). `accept_bi`
/// yields the underlying stream once, then `ConnectionClosed`; `open_bi`
/// returns `StreamClosed`.
struct StreamBidiStreamSource {
stream: Mutex<Option<(SendStream, RecvStream)>>, stream: Mutex<Option<(SendStream, RecvStream)>>,
remote_addr: Option<SocketAddr>, remote_addr: Option<SocketAddr>,
} }
#[async_trait]
impl BidiStreamSource for StreamBidiStreamSource {
async fn accept_bi(&self) -> Result<(SendStream, RecvStream), StreamError> {
let mut guard = self.stream.lock().expect("stream mutex poisoned");
match guard.take() {
Some(pair) => Ok(pair),
None => Err(StreamError::ConnectionClosed),
}
}
async fn open_bi(&self) -> Result<(SendStream, RecvStream), StreamError> {
Err(StreamError::StreamClosed)
}
fn remote_addr(&self) -> Option<SocketAddr> {
self.remote_addr
}
/// `code`/`reason` are ignored: a single stream has no QUIC-shaped
/// application-level close codes. The drop is the close (ADR-065
/// §"Negative"). The `_` prefix is intentional — the signature matches
/// the public `Connection::close` API (ADR-070 §"REQ-CORE-02").
fn close(&self, _code: u32, _reason: &str) {
let _ = self.stream.lock().expect("stream mutex poisoned").take();
}
}
pub struct Connection { pub struct Connection {
kind: ConnectionKind, source: Box<dyn BidiStreamSource>,
alpn: Vec<u8>, alpn: Vec<u8>,
identity: OnceLock<Identity>, identity: OnceLock<Identity>,
} }
@@ -383,7 +519,7 @@ impl Connection {
#[cfg(feature = "quinn")] #[cfg(feature = "quinn")]
pub fn from_quinn_with_alpn(conn: quinn::Connection, alpn: Vec<u8>) -> Self { pub fn from_quinn_with_alpn(conn: quinn::Connection, alpn: Vec<u8>) -> Self {
Self { Self {
kind: ConnectionKind::Quinn(conn), source: Box::new(QuinnBidiStreamSource { conn }),
alpn, alpn,
identity: OnceLock::new(), identity: OnceLock::new(),
} }
@@ -393,7 +529,7 @@ impl Connection {
pub fn from_iroh(conn: iroh::endpoint::Connection) -> Self { pub fn from_iroh(conn: iroh::endpoint::Connection) -> Self {
let alpn = conn.alpn().to_vec(); let alpn = conn.alpn().to_vec();
Self { Self {
kind: ConnectionKind::Iroh(conn), source: Box::new(IrohBidiStreamSource { conn }),
alpn, alpn,
identity: OnceLock::new(), identity: OnceLock::new(),
} }
@@ -409,7 +545,7 @@ impl Connection {
remote_addr: Option<SocketAddr>, remote_addr: Option<SocketAddr>,
) -> Self { ) -> Self {
Self { Self {
kind: ConnectionKind::Stream(StreamConn { source: Box::new(StreamBidiStreamSource {
stream: Mutex::new(Some(( stream: Mutex::new(Some((
SendStream::from_stream(send), SendStream::from_stream(send),
RecvStream::from_stream(recv), RecvStream::from_stream(recv),
@@ -446,41 +582,11 @@ impl Connection {
/// per single-stream connection; handlers that call once (e.g. /// per single-stream connection; handlers that call once (e.g.
/// `HttpAdapter`) get the stream directly. Both are correct. /// `HttpAdapter`) get the stream directly. Both are correct.
pub async fn accept_bi(&self) -> Result<(SendStream, RecvStream), StreamError> { pub async fn accept_bi(&self) -> Result<(SendStream, RecvStream), StreamError> {
match &self.kind { self.source.accept_bi().await
#[cfg(feature = "quinn")]
ConnectionKind::Quinn(c) => {
let (send, recv) = c.accept_bi().await.map_err(map_quinn_connection_error)?;
Ok((SendStream::from_quinn(send), RecvStream::from_quinn(recv)))
}
#[cfg(feature = "iroh")]
ConnectionKind::Iroh(c) => {
let (send, recv) = c.accept_bi().await.map_err(map_iroh_connection_error)?;
Ok((SendStream::from_iroh(send), RecvStream::from_iroh(recv)))
}
ConnectionKind::Stream(sc) => {
let mut guard = sc.stream.lock().expect("stream mutex poisoned");
match guard.take() {
Some(pair) => Ok(pair),
None => Err(StreamError::ConnectionClosed),
}
}
}
} }
pub async fn open_bi(&self) -> Result<(SendStream, RecvStream), StreamError> { pub async fn open_bi(&self) -> Result<(SendStream, RecvStream), StreamError> {
match &self.kind { self.source.open_bi().await
#[cfg(feature = "quinn")]
ConnectionKind::Quinn(c) => {
let (send, recv) = c.open_bi().await.map_err(map_quinn_connection_error)?;
Ok((SendStream::from_quinn(send), RecvStream::from_quinn(recv)))
}
#[cfg(feature = "iroh")]
ConnectionKind::Iroh(c) => {
let (send, recv) = c.open_bi().await.map_err(map_iroh_connection_error)?;
Ok((SendStream::from_iroh(send), RecvStream::from_iroh(recv)))
}
ConnectionKind::Stream(_) => Err(StreamError::StreamClosed),
}
} }
pub fn remote_alpn(&self) -> &[u8] { pub fn remote_alpn(&self) -> &[u8] {
@@ -488,31 +594,11 @@ impl Connection {
} }
pub fn remote_addr(&self) -> Option<SocketAddr> { pub fn remote_addr(&self) -> Option<SocketAddr> {
match &self.kind { self.source.remote_addr()
#[cfg(feature = "quinn")]
ConnectionKind::Quinn(c) => Some(c.remote_address()),
#[cfg(feature = "iroh")]
ConnectionKind::Iroh(_) => None,
ConnectionKind::Stream(sc) => sc.remote_addr,
}
} }
pub fn close(&self, code: u32, reason: &str) { pub fn close(&self, code: u32, reason: &str) {
match &self.kind { self.source.close(code, reason)
#[cfg(feature = "quinn")]
ConnectionKind::Quinn(c) => {
let code = quinn::VarInt::from(code);
c.close(code, reason.as_bytes());
}
#[cfg(feature = "iroh")]
ConnectionKind::Iroh(c) => {
let code = iroh::endpoint::VarInt::from(code);
c.close(code, reason.as_bytes());
}
ConnectionKind::Stream(sc) => {
let _ = sc.stream.lock().expect("stream mutex poisoned").take();
}
}
} }
pub fn set_identity(&self, identity: Identity) -> Result<(), IdentityAlreadySet> { pub fn set_identity(&self, identity: Identity) -> Result<(), IdentityAlreadySet> {
+1 -1
View File
@@ -1,7 +1,7 @@
--- ---
id: core/bidistreamsource-trait id: core/bidistreamsource-trait
name: "Implement BidiStreamSource trait + AuthContext::anonymous (ADR-070, REQ-CORE-01/02/03)" name: "Implement BidiStreamSource trait + AuthContext::anonymous (ADR-070, REQ-CORE-01/02/03)"
status: pending status: completed
depends_on: [] depends_on: []
scope: broad scope: broad
risk: medium risk: medium