From 8f122b0a38b94474f10028988830d61be02cae41 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Mon, 7 Sep 2026 08:45:09 +0000 Subject: [PATCH] =?UTF-8?q?feat(review=20007=20Unit=202):=20birth-teardown?= =?UTF-8?q?=20telemetry=20=E2=80=94=20unaccepted-stream=20debug=20hint=20(?= =?UTF-8?q?R-02)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The optional hardening half of R-02 (the doc notes landed with Unit 1, dc4ad2b). The wrapper's teardown task now observes whether the pump handler ever accepted the channel's stream: - ChannelBidiStreamSource gains a shared acceptance flag (with_accepted_flag / accepted()); channel_source_with_accepted_flag threads it from run_open_wrapper. - On handler exit without accept, the teardown task logs a debug! naming the contract: "the returned JoinHandle must track the data-plane lifetime ... await pumps inline" — the telemetry hint for the teardown-at-birth shape the alktunnels POC hit empirically. - Telemetry only: no behavior change; the benign no-data close still tears down identically. Verification: cargo test (623 passed, +2: accepted-flag flip on the yield-once accept; accepting vs non-accepting handler teardown behavior), clippy -D warnings, fmt. --- src/channels/operations.rs | 104 ++++++++++++++++++++++++++++++++++++- src/channels/source.rs | 62 +++++++++++++++++++++- 2 files changed, 164 insertions(+), 2 deletions(-) diff --git a/src/channels/operations.rs b/src/channels/operations.rs index 04a042b..0c3837b 100644 --- a/src/channels/operations.rs +++ b/src/channels/operations.rs @@ -805,7 +805,17 @@ async fn run_open_wrapper( }; let remote_addr = manager.remote_addr(); - let source = super::source::channel_source(recv, send, remote_addr); + // The acceptance flag (R-02 telemetry): shared between the + // yield-once source (flips on the handler's accept) and + // the teardown task (logs the birth-teardown hint when the + // handler exits without accepting). + let accepted = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let source = super::source::channel_source_with_accepted_flag( + recv, + send, + remote_addr, + Arc::clone(&accepted), + ); let channel_conn = Connection::from_source(source, alpn.as_bytes().to_vec()); let raw_task = open_handler(input, plan, channel_conn, auth.clone()); @@ -813,6 +823,23 @@ async fn run_open_wrapper( let teardown_policy = Arc::clone(policy); let task = tokio::spawn(async move { let _ = raw_task.await; + if !accepted.load(std::sync::atomic::Ordering::SeqCst) { + // The handler returned without ever accepting the + // channel's stream — almost always the + // "spawned-and-forgotten / returned early" shape + // (R-02): teardown-at-birth, both pumps see + // instant EOF. The benign case (peer never sent, + // handler closed cleanly) is rare by comparison; + // the debug line is the hint to check the handler's + // return-value semantics. + tracing::debug!( + channel_id = id, + "open wrapper: handler exited without accepting the channel's \ + stream — if the handler returns before its pumps finish, the \ + channel tears down at birth; the returned JoinHandle must \ + track the data-plane lifetime (await pumps inline)" + ); + } // Benign-race pin (review 006 E-03): `channel/close`'s // awaited teardown and this handler-exit teardown both // gate on the atomic `opener_ledger().take(id)`, so the @@ -1834,6 +1861,81 @@ mod tests { assert!(spawned.load(Ordering::SeqCst)); } + #[tokio::test] + async fn run_open_wrapper_birth_teardown_hint_fires_only_when_stream_unaccepted() { + // R-02 telemetry: the acceptance flag flips when the handler + // accepts the channel's stream, and does not when it exits + // without accepting. Two handlers over two opens: one accepts + // (the healthy shape), one returns immediately (the + // birth-teardown shape the hint warns about). + let manager = make_manager().await; + let policy = super::super::policy::default_policy(); + let auth = AuthContext::anonymous(b"alk/call"); + let accepting: OpenHandler = Arc::new(|_input, _plan, conn, _auth| { + tokio::spawn(async move { + let _bidi = conn.accept_bi().await.expect("accept"); + }) + }); + let env = run_open_wrapper( + &manager, + &policy, + None, + &accepting, + &auth, + "alk/tty", + json!({}), + "alice".to_string(), + identity("alice"), + "req-accepted".to_string(), + None, + ) + .await; + let id = env.result.as_ref().expect("open ok")["channel_id"] + .as_u64() + .expect("id") as u32; + let accepted_task = manager + .teardown_channel(id) + .expect("teardown") + .expect("task"); + let _ = accepted_task.await; + // (The flag itself is consumed inside the teardown task; the + // source-level test proves the flip. Here we prove the + // accepting handler's teardown task completes cleanly — no + // panic, channel gone.) + + // The non-accepting handler: exits immediately — the + // birth-teardown shape R-02's hint targets. The open still + // succeeds; the teardown task observes the un-accepted stream. + let non_accepting: OpenHandler = + Arc::new(|_input, _plan, _conn, _auth| tokio::spawn(async {})); + let env = run_open_wrapper( + &manager, + &policy, + None, + &non_accepting, + &auth, + "alk/tty", + json!({}), + "alice".to_string(), + identity("alice"), + "req-not-accepted".to_string(), + None, + ) + .await; + let id = env.result.as_ref().expect("open ok")["channel_id"] + .as_u64() + .expect("id") as u32; + let task = manager + .teardown_channel(id) + .expect("teardown") + .expect("task"); + let _ = task.await; + assert!( + !manager.has_channel(id), + "birth-teardown shape: the channel is gone when the handler exits" + ); + } + #[test] fn establishment_error_reason_mapping_covers_the_vocabulary() { let cases: Vec<(EstablishmentError, &str)> = vec![ diff --git a/src/channels/source.rs b/src/channels/source.rs index 66d6eee..c853760 100644 --- a/src/channels/source.rs +++ b/src/channels/source.rs @@ -10,6 +10,8 @@ //! to the mux write half via `BiStream::from_joined`. use std::net::SocketAddr; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; use async_trait::async_trait; use parking_lot::Mutex; @@ -28,9 +30,16 @@ use crate::core::types::{BiStream, BidiStreamSource, StreamError}; /// yield-once contract (ADR-007) and the "split never crosses a crate /// boundary as part of a constructor" rule (ADR-009) — the join happens /// here, in the `BidiStreamSource` impl, not per-handler. +/// +/// The `accepted` flag (R-02 telemetry) records whether `accept_bi` +/// ever yielded: the open wrapper's teardown task checks it when the +/// pump handler exits and logs a `debug!` hint when the channel's +/// stream was never accepted — the "teardown at birth" shape a +/// fire-and-forget handler produces (review 007 R-02). pub struct ChannelBidiStreamSource { stream: Mutex>, remote_addr: Option, + accepted: Arc, } impl ChannelBidiStreamSource { @@ -38,11 +47,29 @@ impl ChannelBidiStreamSource { /// half joined to the mux write half). The handler will call /// `accept_bi()` once and receive this stream. pub fn new(stream: BiStream, remote_addr: Option) -> Self { + Self::with_accepted_flag(stream, remote_addr, Arc::new(AtomicBool::new(false))) + } + + /// Construct with a shared acceptance flag — the open wrapper + /// passes its own so the teardown task can observe whether the + /// handler ever accepted the channel's stream (R-02 telemetry). + pub fn with_accepted_flag( + stream: BiStream, + remote_addr: Option, + accepted: Arc, + ) -> Self { Self { stream: Mutex::new(Some(stream)), remote_addr, + accepted, } } + + /// Whether `accept_bi` ever yielded the stream. `false` after + /// construction; flips once on the first successful accept. + pub fn accepted(&self) -> bool { + self.accepted.load(Ordering::SeqCst) + } } #[async_trait] @@ -50,7 +77,10 @@ impl BidiStreamSource for ChannelBidiStreamSource { async fn accept_bi(&self) -> Result { let mut guard = self.stream.lock(); match guard.take() { - Some(stream) => Ok(stream), + Some(stream) => { + self.accepted.store(true, Ordering::SeqCst); + Ok(stream) + } None => Err(StreamError::ConnectionClosed), } } @@ -89,11 +119,26 @@ pub fn channel_source( ChannelBidiStreamSource::new(stream, remote_addr) } +/// [`channel_source`] with a shared acceptance flag (R-02 telemetry): +/// the open wrapper passes the flag, hands the source to the +/// handler's `Connection`, and the teardown task reads it to log the +/// "handler exited without accepting the channel's stream" hint. +pub fn channel_source_with_accepted_flag( + recv: super::reassembly::MpscRecvStream, + send: super::reassembly::MpscSendStream, + remote_addr: Option, + accepted: Arc, +) -> ChannelBidiStreamSource { + let stream = BiStream::from_joined(recv, send); + ChannelBidiStreamSource::with_accepted_flag(stream, remote_addr, accepted) +} + #[cfg(test)] mod tests { use super::*; use crate::channels::reassembly::{MpscRecvStream, MpscSendStream}; use bytes::Bytes; + use std::sync::Arc; use tokio::io::{AsyncReadExt, AsyncWriteExt}; fn make_pair() -> ( @@ -166,4 +211,19 @@ mod tests { bidi.read_exact(&mut buf).await.expect("read"); assert_eq!(&buf, b"inbound"); } + + #[tokio::test] + async fn accepted_flag_tracks_the_yield_once_accept() { + let (send, _mux_recv, _demux_send, recv) = make_pair(); + let accepted = Arc::new(AtomicBool::new(false)); + let source = channel_source_with_accepted_flag(recv, send, None, Arc::clone(&accepted)); + assert!(!source.accepted(), "false before the handler accepts"); + let _bidi = source.accept_bi().await.expect("first accept"); + assert!(source.accepted(), "flips on the accept"); + assert!(matches!( + source.accept_bi().await, + Err(StreamError::ConnectionClosed) + )); + assert!(source.accepted(), "stays true after exhaustion"); + } }