Expose shared backend references through stream wrappers

PATCHES.md

@@ -169,3 +169,25 @@ Exact tracked-source validation passes under `target/pr1-tracked-alignment/`. Fr
 The original ignored Quinn 0.11.12/proto 0.11.19 lock was restored byte for byte in `finally`, with SHA256 `b79696fb5e7e3ea6eb626a062448c1f17359c5fadea93bec5d79b3d68bfc772e`. On that original lock, `08-workspace-default.log` and `09-workspace-all-features.log` each pass 347 tests with zero failures/ignored tests. Formatting passes in `10-fmt.log`; full all-target/all-feature Clippy matches exactly the 16 historical messages and primary file/line/column locations in `11-clippy-all.log` and `12-clippy-baseline.log`. Strict workspace-library and native-test Clippy pass in `13-clippy-libraries-strict.log` and `14-clippy-native-strict.log`. Strict h3-quinn rustdoc passes in `15-rustdoc-strict.log`; workspace docs pass in `16-workspace-doc.log` with only the six existing bare-URL warnings in unchanged h3/src/config.rs. `git diff --check` passes in `17-diff-check.log`. These checks ran once on the final Rust/manifest source, with two build jobs and 300-second per-command bounds.
 
 `validated-sources.json` fingerprints all 89 tracked Rust/manifest files used in the final checks. `preserved-controls.sha256.json` verifies that the true-base pass, unaligned-89ed failure, temporary-copy alignment pass and their natural locks remain unchanged; all earlier failed logs are retained. The prior inherited-baseline characterization remains withdrawn. No runtime, API, test, resolver, other dependency or Runewarp change is added by this final alignment. Original source-only reviews cover c854 only; both manifest repairs are post-review. Parent owns the fresh targeted review, CI and readiness disposition, and PR2 remains held. No merge, additional reviewer delegation, benchmark or downstream-ready claim is included.
+
+## Shared backend references through stream wrappers
+
+Patch 2 of the completion API series is based on PR #5, `xenorune/quinn-send-stopped` at `7b9ea7c1dc7e61ec20960344a112bfc9514bbd0e`, descending from integration base `df27106fb243baaf867d11441d11cf671ec9da7e`. It adds inherent `get_ref(&self) -> &S` accessors to `h3::stream::BufRecvStream` and `h3_webtransport::stream::SendStream`. The WebTransport accessor returns the underlying backend through the buffered wrapper. Callers can capture a backend-owned completion observer while retaining the wrapper and its session guard. Neither accessor consumes buffered data or changes framing, full session IDs, writes, finish, reset, cancellation or teardown. No mutable accessor, consuming escape, required QUIC trait method, task, callback, queue, unsafe code, dependency/MSRV change or Runewarp policy is added.
+
+Backend observations keep their native contract. With h3-quinn, `stopped()` returning `Ok(None)` establishes local FIN/data acknowledgment only on a caller-known normal-FIN path without a prior reset. Absent send state after RESET acknowledgment can also produce `None`; peer STOP returns `Some(code)` before RESET acknowledgment, and a pending observer has no reliable RESET-ACK wakeup. This is not remote application consumption, receive-half release, session completion or general stream-free/abort recovery. Dropping an observer releases its task waker, but Quinn may retain its per-stream notification entry until a stream event or connection cleanup. This patch does not resolve Runewarp's general Drop recovery blocker or unexplained burst losses.
+
+Five public real-Quinn regressions in `examples/tests/webtransport_server.rs` cover partial buffered-byte preservation and EOF, full session ID four after an ordinary request, exact bidirectional payloads and independent FIN with an observer outliving its send wrapper, observer cancellation releasing its task waker while writes remain usable, session-guard rejection after close despite a live observer, and connection/session teardown with an independently owned live observer. Every new asynchronous test has a five-second internal deadline; pending-state checks poll directly without guessed sleeps or detached observation tasks.
+
+Scoped TDD evidence is retained in `target/pr2-scoped/`. `cargo test --locked -p examples --test webtransport_server buffered_get_ref_preserves_partial_bytes_and_eof` first fails E0599 for the missing buffered accessor in `01-buffered-red.log`. An unrelated fixture generic annotation error is preserved in `02-buffered-green.log`; after correction the test passes in `03-buffered-green.log`. The analogous `send_get_ref_observer_preserves_session_id_bytes_and_fin` command fails E0599 for the missing WebTransport accessor in `04-send-red.log` and passes in `05-send-green.log`. `cargo test --locked -p examples --test webtransport_server get_ref` passes all five regressions in `06-accessor-behavior.log`. The final formatted-source rerun also passes all five tests in `07-formatted-accessor-behavior.log`. The seven commands total 30.502 seconds, use two build jobs and are bounded by the 180-second cumulative scoped budget and a 300-second per-command maximum. They use the copied ignored Quinn 0.11.12/proto 0.11.19 lock, SHA256 `b79696fb5e7e3ea6eb626a062448c1f17359c5fadea93bec5d79b3d68bfc772e`. Full required validation, independent review and CI remain pending; these scoped results do not replace them.
+
+### Patch 2 full validation
+
+The exact final Rust source passes both `cargo test --locked --workspace` and `cargo test --locked --workspace --all-features`, each with 352 tests, zero failures and zero ignored tests, in `target/pr2-validation/10-workspace-default.log` and `11-workspace-all-features.log`. The scoped missing-method E0599 failures establish API absence, not a preexisting behavioral bug. The unrelated fixture annotation failure remains separate. Initial full Clippy reports one new warning for the cancellation test's no-op waker; `01-clippy-all.log` and `02-clippy-baseline.json` preserve it. A narrow test-only correction records wakeups and asserts none before cancellation. Final full-suite runs include that correction.
+
+Final all-target/all-feature Clippy matches the 16 inherited warnings exactly by message and primary file/line/column in `05-clippy-all-final.log` and `06-clippy-baseline-final.json`. Strict workspace-library and changed-test Clippy pass in `03-clippy-libraries-strict.log` and `07-clippy-test-strict.log`. Formatting passes in `04-fmt.log`; strict WebTransport API rustdoc passes in `08-rustdoc-webtransport-strict.log`; workspace docs, including both new accessors, pass in `09-workspace-doc.log` with only the six inherited bare-URL warnings in unchanged `h3/src/config.rs`, verified in `09-doc-baseline.json`. No upstream warning is blanket-fixed.
+
+Fresh natural workspace fallback, without a supplied lock, precise selection or dependency override, selects Quinn 0.11.9/proto 0.11.14/UDP 0.5.14. Its lock SHA256 is `972b79527137449fb0920d72708b56339c81396bc266c4cc947aab0990d01333`, matching PR #5's aligned resolution. Actual Rust 1.74.1 checks pass for h3 and h3-quinn in `15-msrv-h3.log` and `16-msrv-h3-quinn.log`. The optional h3-webtransport check fails on 16 associated-type-bound E0658 errors in unchanged detached Server files. An independent clean archive of exact base `7b9ea7c1dc7e61ec20960344a112bfc9514bbd0e`, resolving its own fresh natural lock, produces the identical lock and the same 16 errors at identical source locations. `17-msrv-webtransport.log`, `18-base-natural-resolution.log`, `19-base-msrv-webtransport.log` and `20-msrv-webtransport-control.json` preserve this matched control. WebTransport does not declare Rust 1.74.1 support; this check is not claimed as passing, and its inherited compatibility limitation is not repaired here. All five accessor tests and all four native completion tests pass on the naturally selected minimum dependencies in `21-natural-accessor-tests.log` and `22-natural-native-tests.log`. The original ignored Quinn 0.11.12/proto 0.11.19 lock is restored byte for byte after each temporary-lock phase.
+
+The WebTransport Server admission example test and all example builds pass in `23-example-webtransport-budget.log` and `24-build-examples.log`. The ordinary client/server exchange passes in `25-example-exchange.log`. Native h3spec on localhost port 4434 passes 43 examples with the five existing exclusions in `26-h3spec.log` and `h3spec.log`. No Docker, capacity benchmark, performance measurement, browser interoperability or CI result is claimed. Source/lock fingerprints, command vectors, exits and durations are recorded under `target/pr2-validation/`. Independent two-axis review and CI disposition remain owned by the main orchestrator; this candidate is not yet reviewed or approved for downstream use.
+
+The full-validation window used 155.145 seconds of conservatively counted command time, including 0.529 seconds for native h3spec, within the separately approved five-minute budget. Commands used two build jobs and individual 300-second bounds. `27-final-verification.log` verifies all 89 Rust/manifest fingerprints, restored lock, exact four-file scope, warning baselines, matched optional-MSRV control and protected checkout state. `process-release-verification.json` confirms every owned Cargo/compiler/documentation/formatter/test/example/h3spec process and recorded process group exited and UDP ports 4433/4434 were released. Explicit release was sent to both parents and the diagnostic thread before Git-only delivery; no further Cargo command ran after release.

examples/tests/webtransport_server.rs

@@ -71,6 +71,41 @@ async fn connections_with_credit(
     (server, client, server_conn, client_conn)
 }
 
+#[tokio::test]
+async fn buffered_get_ref_preserves_partial_bytes_and_eof() {
+    use h3::quic::RecvStream as _;
+
+    timeout(Duration::from_secs(5), async {
+        let (_se, _ce, sq, cq) = connections().await;
+        let mut send = sq.open_uni().await.unwrap();
+        send.write_all(b"buffered tail").await.unwrap();
+        send.finish().unwrap();
+        let mut adapter = h3_quinn::Connection::new(cq);
+        let recv = futures::future::poll_fn(|cx| {
+            <h3_quinn::Connection as h3::quic::Connection<Bytes>>::poll_accept_recv(
+                &mut adapter,
+                cx,
+            )
+        })
+        .await
+        .unwrap();
+        let mut recv = h3::stream::BufRecvStream::<_, Bytes>::new(recv);
+        assert!(!futures::future::poll_fn(|cx| recv.poll_read(cx))
+            .await
+            .unwrap());
+        assert_eq!(recv.take_chunk(3).unwrap(), b"buf"[..]);
+        assert_eq!(recv.get_ref().recv_id().into_inner(), u64::from(send.id()));
+        assert_eq!(recv.take_chunk(32).unwrap(), b"fered tail"[..]);
+        assert!(futures::future::poll_fn(|cx| recv.poll_data(cx))
+            .await
+            .unwrap()
+            .is_none());
+        assert!(recv.is_eos());
+    })
+    .await
+    .expect("buffered get_ref test timed out");
+}
+
 #[tokio::test]
 async fn detached_session_opens_while_get_and_put_continue() {
     timeout(Duration::from_secs(5), async {
@@ -352,6 +387,165 @@ fn connect_request() -> Request<()> {
     req
 }
 
+fn completion_observer(
+    future: impl std::future::Future<Output = Result<Option<quinn::VarInt>, quinn::StoppedError>>
+        + Send
+        + Sync
+        + 'static,
+) -> impl std::future::Future<Output = Result<Option<quinn::VarInt>, quinn::StoppedError>>
+       + Send
+       + Sync
+       + 'static {
+    future
+}
+
+#[tokio::test]
+async fn send_get_ref_observer_preserves_session_id_bytes_and_fin() {
+    use h3::quic::BidiStream as _;
+
+    timeout(Duration::from_secs(5), async {
+        let mut pair = Pair::default().await;
+        pair.ordinary().await;
+        let (session, client) = pair.session().await;
+        assert_eq!(
+            session.session_id(),
+            h3_webtransport::SessionId::try_from(4).unwrap()
+        );
+        assert_eq!(
+            client.session_id(),
+            h3_webtransport::SessionId::try_from(4).unwrap()
+        );
+        let (mut send, mut recv) = session.opener().open_bi().await.unwrap().split();
+        let stopped = completion_observer(send.get_ref().stopped());
+        send.write_all(b"observed tail").await.unwrap();
+        send.shutdown().await.unwrap();
+        drop(send);
+        let mut peer = client.accept_bi().await.unwrap();
+        let mut bytes = Vec::new();
+        peer.read_to_end(&mut bytes).await.unwrap();
+        assert_eq!(bytes, b"observed tail");
+        assert_eq!(stopped.await.unwrap(), None);
+        peer.write_all(b"independent reply").await.unwrap();
+        peer.shutdown().await.unwrap();
+        bytes.clear();
+        recv.read_to_end(&mut bytes).await.unwrap();
+        assert_eq!(bytes, b"independent reply");
+        pair.ordinary().await;
+    })
+    .await
+    .expect("send get_ref FIN test timed out");
+}
+
+#[tokio::test]
+async fn send_get_ref_observer_cancellation_releases_waker_and_preserves_writes() {
+    use h3::quic::BidiStream as _;
+    use std::sync::atomic::{AtomicBool, Ordering};
+    use std::task::{Context, Wake, Waker};
+
+    struct ObserverTask(AtomicBool);
+    impl Wake for ObserverTask {
+        fn wake(self: Arc<Self>) {
+            self.0.store(true, Ordering::SeqCst);
+        }
+    }
+
+    timeout(Duration::from_secs(5), async {
+        let mut pair = Pair::default().await;
+        let (session, client) = pair.session().await;
+        let (mut send, _recv) = session.opener().open_bi().await.unwrap().split();
+        let mut stopped = Box::pin(completion_observer(send.get_ref().stopped()));
+        let task = Arc::new(ObserverTask(AtomicBool::new(false)));
+        let weak = Arc::downgrade(&task);
+        {
+            let waker = Waker::from(task);
+            let mut cx = Context::from_waker(&waker);
+            assert!(std::future::Future::poll(stopped.as_mut(), &mut cx).is_pending());
+        }
+        assert!(weak.upgrade().is_some());
+        assert!(!weak.upgrade().unwrap().0.load(Ordering::SeqCst));
+        drop(stopped);
+        assert!(
+            weak.upgrade().is_none(),
+            "cancelled observer retained its task waker"
+        );
+        // Quinn may retain its Notify entry. Cancellation must leave the guarded
+        // stream usable and permit a new, independently owned observation.
+        let stopped = completion_observer(send.get_ref().stopped());
+        send.write_all(b"after cancellation").await.unwrap();
+        send.shutdown().await.unwrap();
+        let mut peer = client.accept_bi().await.unwrap();
+        let mut bytes = Vec::new();
+        peer.read_to_end(&mut bytes).await.unwrap();
+        assert_eq!(bytes, b"after cancellation");
+        assert_eq!(stopped.await.unwrap(), None);
+    })
+    .await
+    .expect("send get_ref cancellation test timed out");
+}
+
+#[tokio::test]
+async fn send_get_ref_live_observer_keeps_session_guard_effective() {
+    use futures::{future::poll_fn, FutureExt};
+    use h3::quic::{BidiStream as _, SendStream as _, SendStreamUnframed as _};
+
+    timeout(Duration::from_secs(5), async {
+        let mut pair = Pair::default().await;
+        let (session, client) = pair.session().await;
+        let (mut send, _recv) = session.opener().open_bi().await.unwrap().split();
+        let mut peer = client.accept_bi().await.unwrap();
+        let mut stopped = Box::pin(completion_observer(send.get_ref().stopped()));
+        assert!(stopped.as_mut().now_or_never().is_none());
+        session.close(7, "observer still alive").await.unwrap();
+        let (local, remote) = tokio::join!(session.closed(), client.closed());
+        assert_eq!(local.unwrap().code, 7);
+        assert_eq!(remote.unwrap().code, 7);
+        let mut bytes = Bytes::from_static(b"forbidden after close");
+        assert!(matches!(poll_fn(|cx| send.poll_send(cx, &mut bytes)).await,
+            Err(h3::quic::StreamErrorIncoming::StreamTerminated { error_code })
+                if error_code == h3_webtransport::WT_SESSION_GONE));
+        assert_eq!(bytes, b"forbidden after close"[..]);
+        assert!(matches!(poll_fn(|cx| send.poll_finish(cx)).await,
+            Err(h3::quic::StreamErrorIncoming::StreamTerminated { error_code })
+                if error_code == h3_webtransport::WT_SESSION_GONE));
+        assert!(send.write_all(b"also forbidden").await.is_err());
+        let mut bytes = Vec::new();
+        assert!(peer.read_to_end(&mut bytes).await.is_err());
+        assert!(bytes.is_empty());
+        // Do not await RESET acknowledgment: native stopped() has no reliable
+        // RESET-ACK wakeup. A live observer must not hold up session teardown.
+        drop(stopped);
+        pair.ordinary().await;
+    })
+    .await
+    .expect("send get_ref guard test timed out");
+}
+
+#[tokio::test]
+async fn send_get_ref_live_observer_outlives_wrapper_and_connection_teardown() {
+    use futures::FutureExt;
+    use h3::quic::BidiStream as _;
+
+    timeout(Duration::from_secs(5), async {
+        let mut pair = Pair::default().await;
+        let (session, client) = pair.session().await;
+        let (send, recv) = session.opener().open_bi().await.unwrap().split();
+        let mut stopped = Box::pin(completion_observer(send.get_ref().stopped()));
+        assert!(stopped.as_mut().now_or_never().is_none());
+        pair.sq.close(0_u32.into(), b"observer teardown");
+        assert!(session.closed().await.is_err());
+        assert!(client.closed().await.is_err());
+        drop((send, recv, session, client, pair));
+        assert!(matches!(
+            stopped.await,
+            Err(quinn::StoppedError::ConnectionLost(
+                quinn::ConnectionError::LocallyClosed
+            ))
+        ));
+    })
+    .await
+    .expect("send get_ref connection teardown test timed out");
+}
+
 #[tokio::test]
 async fn invalid_connects_reject_without_a_session_and_preserve_put() {
     timeout(Duration::from_secs(5), async {

h3-webtransport/src/stream.rs

@@ -212,6 +212,18 @@ pub struct SendStream<S, B> {
     guard: SessionGuard<S, B>,
 }
 
+impl<S, B> SendStream<S, B> {
+    /// Returns a shared reference to the underlying QUIC send stream.
+    ///
+    /// This allows backend-specific observation without consuming the wrapper.
+    /// Continue writing, finishing and resetting through this wrapper to retain
+    /// session lifecycle handling. Observations retain the backend's semantics
+    /// and do not establish receive-side or session completion.
+    pub fn get_ref(&self) -> &S {
+        self.stream.get_ref()
+    }
+}
+
 impl<S, B> std::fmt::Debug for SendStream<S, B> {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("SendStream")

h3/src/stream.rs

@@ -407,6 +407,15 @@ impl<S, B> BufRecvStream<S, B> {
             _marker: PhantomData,
         }
     }
+
+    /// Returns a shared reference to the underlying stream.
+    ///
+    /// This does not consume buffered bytes or change the stream's read position.
+    /// Backend-specific observation through this reference retains the backend's
+    /// semantics; it does not account for bytes still buffered by this wrapper.
+    pub fn get_ref(&self) -> &S {
+        &self.stream
+    }
 }
 
 impl<S, B> BufRecvStream<S, B>