fix(quic): announce stream limits that arrive after a refused open

quest/m0/ci/README.md

@@ -66,7 +66,6 @@ The historical samples locate candidates; they are not controlled benchmarks.
 - [Share local check capacity](/quest/m0/ci/local-gate.md) - gate expensive phases and bound internal fanout
 - [Dart FFI build](/quest/m0/ci/dart-ffi-build.md) - the Dart check builds release `moq-ffi` once instead of twice
 - [Windows restart drill](/quest/m0/ci/platform-windows-drill.md) - the relay restart drill passes on Windows in the merge queue
-- [macOS zero-RTT test](/quest/m0/ci/platform-macos-zero-rtt.md) - the moq-tokio 0-RTT driver test passes on macOS in the merge queue
 - [Require platform gates](/quest/m0/ci/platform-queue.md) - activate the consolidated platform checks in the main and release rulesets after the workflow lands
 - [mbx on R2](/quest/m0/ci/mbx-r2.md) - Rust builds on every OS read an R2 cache that only `main` writes, escaping the 10 GB Actions cache
 - [Nix cache](/quest/m0/ci/nix-cache.md) - `nix.moq.dev` on R2 replaces Cachix for CI and releases

quest/m0/ci/platform-queue.md

@@ -36,7 +36,6 @@ the implementation PR.
 ## Required
 
 - [Windows restart drill](/quest/m0/ci/platform-windows-drill.md) - a required Platform gate would block every merge while it fails
-- [macOS zero-RTT test](/quest/m0/ci/platform-macos-zero-rtt.md) - a required Platform gate would block every merge while it fails
 
 ## Related
 

rs/moq-quic/README.md

@@ -68,6 +68,7 @@ Changes on top of the upstream commit, besides the renames:
 - **Instants on [moq-clock](../moq-clock)**: the crate's `Instant` is `moq_clock::Instant`, so one clock steps every timer, tokio's paused clock and the simulator included. qlog stamps events relative to `QlogConfig::start_time` on that clock rather than the streamer's own `std::time::Instant`.
 - **`RESET_STREAM_AT`** ([draft-ietf-quic-reliable-stream-reset](https://datatracker.ietf.org/doc/draft-ietf-quic-reliable-stream-reset/)): the `reset_stream_at` transport parameter is always advertised, `Checkpoint::reset` (from `SendStream::checkpoint`) keeps the data written before the checkpoint reliable, holding a plain `RESET_STREAM` until that data is acknowledged when the peer lacks the extension, and a received reset surfaces only once the reliable prefix has been read. Stopping a stream after it was reset no longer credits its unread data twice.
 - **Draining endpoint event**: `EndpointEvent::is_draining` fires once per connection when it stops being active, before the drained event, so the driver's `Endpoint::wait_all_draining` can end a shutdown without sitting out every closing period.
+- **Late stream limits**: peer transport parameters that raise a stream limit after an open was refused emit `StreamEvent::Available`, as `MAX_STREAMS` does. A server accepted before the rest of a multi-packet ClientHello (a post-quantum key share is enough) otherwise leaves a 0.5-RTT `open_uni` parked until the idle timeout. Not offered upstream yet.
 
 Recheck the BBR3 changes when cherry-picking quinn#2481 updates, or if it merges.
 

rs/moq-quic/src/connection/streams/mod.rs

@@ -50,6 +50,7 @@ impl<'a> Streams<'a> {
 
         if self.state.next[dir as usize] >= self.state.max[dir as usize] {
             self.state.streams_blocked[dir as usize] = true;
+            self.state.open_refused[dir as usize] = true;
             return None;
         }
 

rs/moq-quic/src/connection/streams/state.rs

@@ -150,6 +150,8 @@ pub struct StreamsState {
     receive_window_shrink_debt: u64,
     /// Whether the locally-initiated stream limit has been hit, per direction
     pub(super) streams_blocked: [bool; 2],
+    /// Whether an open was refused since the stream limit last rose, per direction
+    pub(super) open_refused: [bool; 2],
     /// Value of `max_data` for which a `DATA_BLOCKED` frame was most recently queued
     ///
     /// A new frame is only queued once the peer has raised the limit.
@@ -203,6 +205,7 @@ impl StreamsState {
             generation: 0,
             receive_window_shrink_debt: 0,
             streams_blocked: [false, false],
+            open_refused: [false, false],
             data_blocked_limit: None,
         }
     }
@@ -212,8 +215,18 @@ impl StreamsState {
         self.initial_max_stream_data_bidi_local = params.initial_max_stream_data_bidi_local;
         self.initial_max_stream_data_bidi_remote = params.initial_max_stream_data_bidi_remote;
         self.peer_reset_stream_at = params.reset_stream_at;
-        self.max[Dir::Bi as usize] = params.initial_max_streams_bidi.into();
-        self.max[Dir::Uni as usize] = params.initial_max_streams_uni.into();
+        for (dir, count) in [
+            (Dir::Bi, params.initial_max_streams_bidi),
+            (Dir::Uni, params.initial_max_streams_uni),
+        ] {
+            let count = count.into();
+            // A 0.5-RTT server holds the connection before a ClientHello that spans several
+            // packets has arrived, so an open can be refused before the limit is known.
+            if count > self.max[dir as usize] && mem::take(&mut self.open_refused[dir as usize]) {
+                self.events.push_back(StreamEvent::Available { dir });
+            }
+            self.max[dir as usize] = count;
+        }
         self.received_max_data(params.initial_max_data);
         for (&id, slot) in self.send.iter_mut() {
             if id.initiator() != self.side
@@ -802,6 +815,7 @@ impl StreamsState {
         if count > *current {
             *current = count;
             self.streams_blocked[dir as usize] = false;
+            self.open_refused[dir as usize] = false;
             self.events.push_back(StreamEvent::Available { dir });
         }
 

rs/moq-quic/src/tests/mod.rs

@@ -1403,6 +1403,33 @@ fn alpn_mismatch() {
     );
 }
 
+/// A server accepted before the rest of a multi-packet ClientHello arrives learns the client's
+/// stream limits later, and must announce them to a 0.5-RTT opener that found none.
+#[test]
+fn late_peer_params_make_streams_available() {
+    let _guard = subscribe();
+    // Long ALPN names push the ClientHello into a second Initial, as a post-quantum key share does.
+    let alpn: Vec<Vec<u8>> = (0..4).map(|i| vec![b'a' + i; 255]).collect();
+    let server_config = ServerConfig::with_crypto(Arc::new(server_crypto_with_alpn(alpn.clone())));
+    let mut pair = Pair::new(Default::default(), server_config);
+    pair.server.handle_incoming = Box::new(|_| IncomingConnectionBehavior::Wait);
+    pair.begin_connect(ClientConfig::new(Arc::new(client_crypto_with_alpn(alpn))));
+    pair.drive_client();
+    pair.server.drive_incoming(pair.time, pair.client.addr);
+    let incoming = pair.server.pop_waiting_incoming();
+    let server_ch = pair.server.try_accept(incoming, pair.time).unwrap();
+    assert_eq!(pair.server_streams(server_ch).open(Dir::Uni), None);
+
+    pair.drive();
+    assert!(
+        iter::from_fn(|| pair.server_conn_mut(server_ch).poll()).any(|event| matches!(
+            event,
+            Event::Stream(StreamEvent::Available { dir: Dir::Uni })
+        ))
+    );
+    assert!(pair.server_streams(server_ch).open(Dir::Uni).is_some());
+}
+
 #[test]
 fn stream_id_limit() {
     let _guard = subscribe();