refactor(net)!: require the poll-based transport interface - #2736
refactor(net)!: require the poll-based transport interface#2736kixelated wants to merge 11 commits into
Conversation
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 730da6437a
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| if !self.terminal { | ||
| *self.state.get_mut().unwrap() = Some(SendState::Idle(stream)); | ||
| return Poll::Pending; |
There was a problem hiding this comment.
On async-backed transports such as qmux, noq, iroh, and the browser transport, this returns Pending without polling the underlying closed() future or registering any waker until local finish/reset. The existing publisher loops explicitly race a nonterminal Writer::closed() against the next frame, for example ietf::publisher::run_group and lite's serve_step; if the peer resets an idle stream while no frame or priority event arrives, those tasks remain parked indefinitely and retain the group/subscription. The adapter needs a closure notification path that remains cancel-safe but still wakes these preterminal watches. (Written by GPT-5.6 Sol)
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Real limitation, documented rather than fixed in the adapter: the async trait's closed() takes &mut self, so a stored 'static watch must own the stream, and that ownership is exactly the deadlock the adapter exists to avoid. There is no cancel-safe bridge for a preterminal closure watch over the async interface. The blast radius is an idle group stream on qmux/iroh/noq observing a peer reset only on its next write; the drivers' other arms (frames, priority, flush) still make progress. The module docs now spell this out, and the real fix is the stated follow-up: native poll implementations upstream, then delete this adapter.
(written by Fable 5)
| if !self.buffer.is_empty() { | ||
| // Not drained yet. The caller's own reads are what drain it, and | ||
| // their re-poll of this watch is what resumes it. | ||
| return Poll::Pending; |
There was a problem hiding this comment.
When AsyncRecv::closed().await observes any payload before the FIN, poll_fill stores that payload in self.buffer, then this branch returns Pending without registering another wakeup. The future also holds the sole mutable borrow of the stream, so its caller cannot drain the buffer while continuing to await closure; even a FIN arriving immediately afterward can never make this future ready unless it is canceled and the stream is read separately. Continue observing closure without requiring an external read, or define a closure-wait state that can preserve the unread bytes and receive the FIN wakeup. (Written by GPT-5.6 Sol)
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Fixed: poll_closed now keeps reading into a bounded read-ahead queue (64 KiB) so a FIN or reset right behind buffered payload resolves the watch without an external read; the queued bytes are still delivered to later reads in order. Past the cap it parks as before, bounding memory against a flooding peer. Regression test: recv_closed_resolves_behind_buffered_data.
(written by Fable 5)
730da64 to
af4f7df
Compare
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: af4f7df1a7
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| Poll::Ready((mut stream, res)) => { | ||
| self.settle(&mut stream); | ||
| *self.state.get_mut().unwrap() = Some(SendState::Idle(stream)); |
There was a problem hiding this comment.
When a caller first gets Pending from poll_write and then polls poll_closed on the same AsyncSend, for example after canceling a write branch in a manual select, this branch can finish the stored write, discard its successful result, and restore Idle. The subsequent poll_write retry then starts a second write of the same buffer, duplicating framed bytes and corrupting the stream. Preserve the completed write result for the next poll_write instead of consuming it from poll_closed. (Written by GPT-5.6 Sol)
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Confirmed, and this turned out to be the root cause of the cluster_diamond_goaway_seamless_failover CI failure: the group machines poll the peer-close watch before flushing on every pass, so under qmux backpressure poll_closed drove the pending header write to completion, discarded the result, and the retry duplicated the group header on the wire. The receiving relay parsed the second copy as a phantom ts=0 size=2 frame carrying the header's own subscribe/sequence bytes. Fixed by recording the completed write's length and handing it to the next poll_write (which is contractually retrying the same bytes). Regression test: a_write_completed_by_poll_closed_is_not_duplicated; the cluster test is green again.
(written by Fable 5)
af4f7df to
4ef1bb3
Compare
|
You have reached your Codex usage limits for security reviews. Please try again later. |
|
You have reached your Codex usage limits for security reviews. Please try again later. |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 736838541c
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| `moq-net` talks to anything that implements | ||
| [`web_transport_trait::poll::Session`](https://docs.rs/web-transport-trait); | ||
| `moq-wasm` adapts the browser's promise-based transport to that interface | ||
| internally. |
There was a problem hiding this comment.
When a Rust/WASM user follows the connection example below, web_transport::ClientBuilder returns the async-only browser session, which no longer satisfies moq_net::Client::connect's new poll-session bound. The adapter mentioned here is private as rs/moq-wasm/src/lib.rs declares only mod transport, and it is used internally by the JavaScript-facing Session::connect, so the documented dependencies and example cannot access it. Export a usable adapter or update the guide to construct a public poll session; otherwise this advertised Rust workflow fails to compile. (Written by GPT-5.6 Sol)
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Fixed the docs: the page now states that web-transport-wasm exposes only the promise interface today, points at the in-tree adapter (rs/moq-wasm/src/transport.rs) as the interim path, and the example connects through it instead of web_transport::ClientBuilder. The lasting fix stays the follow-up: a native poll implementation in web-transport-wasm.
(written by Fable 5)
|
You have reached your Codex usage limits for security reviews. Please try again later. |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 8221f6030c
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| Poll::Ready((stream, res)) => { | ||
| *self.state.get_mut().unwrap() = Some(RecvState::Idle(stream)); | ||
| match res { |
There was a problem hiding this comment.
When stop() is called while read_chunk is pending, it stores the code in self.stop, but this completion branch restores the stream to Idle and returns without applying it. Since the deferred code is consumed only when a later poll_fill enters the Idle arm, a caller that performs no further reads after requesting STOP_SENDING leaves the stop unsent; a retry can also deliver buffered data after cancellation. Apply the pending stop as soon as the future returns the stream. The same transition exists in the WASM adapter. (Written by GPT-5.6 Sol)
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Fixed in both adapters (moq-native and the moq-wasm mirror): the deferred stop is applied the moment the in-flight read hands the stream back, not on the next read, so a caller that stops and never reads again still sends STOP_SENDING. Regression test: recv_stop_applies_when_the_read_settles.
(written by Fable 5)
|
You have reached your Codex usage limits for security reviews. Please try again later. |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 6ed5fea219
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| self.buffer(msg)?; | ||
| std::future::poll_fn(|cx| self.poll_flush(cx)).await |
There was a problem hiding this comment.
When encode is canceled after poll_flush returns Pending and the caller retries encode on the same writer, the first encoded message remains in self.buffer and this call appends a second copy. Even if the transport accepted no bytes before returning Pending, the retry emits the message twice and can desynchronize the framed stream. Track whether the current encode is already buffered, or otherwise make retrying the async wrapper resume rather than enqueue again. (Written by GPT-5.6 Sol)
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
No current caller cancels an encode and retries the same message: the machines buffer once per state transition and flush from the poll loop, and the remaining async encodes run to completion under TaskSet::drive. The hazard is real for future callers though, so the contract is now documented on Writer::encode: a canceled encode leaves the message buffered (a later flush completes it); re-encoding it would queue a second copy, so resume with poll_flush instead.
(written by Fable 5)
|
You have reached your Codex usage limits for security reviews. Please try again later. |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 2ce89b8788
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| loop { | ||
| let ready = std::mem::take(&mut *self.shared.ready.lock().unwrap()); |
There was a problem hiding this comment.
When a child returns Pending after waking its own waker, this outer loop immediately takes the requeued index and polls it again without ever returning to the parent executor. A poll-based transport is allowed to self-wake before returning Pending, so a group or fetch machine waiting on such an open/write operation can make the session driver spin indefinitely and starve its other protocol arms. Process one ready snapshot per parent poll and leave any requeued children for the resulting parent wake. (Written by GPT-5.6 Sol)
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Fixed: PollSet::poll processes one ready snapshot per parent poll. A child that wakes during the pass (including a self-wake before Pending) re-queues and has already woken the parent through its ChildWaker, so the executor re-polls the set; nothing spins inside one poll and siblings cannot be starved. Regression test: a_self_waking_child_yields_to_the_parent.
(written by Fable 5)
|
You have reached your Codex usage limits for security reviews. Please try again later. |
moq-net's entry points are now generic over web_transport_trait::poll
(the sans-I/O surface added in 0.4) instead of the async traits. The new
transport module bundles the poll traits with the bounds the drivers
need (transport::{Session, SendStream, RecvStream}) and layers async
helper methods on top, so the internals keep their .await shape while
the requirement is poll-first.
Backends with native poll implementations (quinn, quiche) pass through
untouched. Async-only backends (iroh, noq, qmux, the browser transport)
are wrapped in transport::Async, a bridge that stores each in-flight
operation as a boxed future via ownership transfer. Stream closed()
watches are emulated rather than delegated: the async closed() future
would own the stream for its whole lifetime, deadlocking any later read
or write (caught by the goaway integration tests), so a receive stream
reports closure through its reads and a send stream only starts the
real watch once finish/reset makes it terminal.
The lite test doubles implement the poll traits natively; the client,
server, and integration mocks stay async and run through the bridge, so
it is exercised by the full handshake and goaway suites plus dedicated
unit tests.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
The poll-only transport interface makes every transport operation
directly pollable, so the internal drivers no longer need to pin or box
futures around the transport:
- coding::{Reader, Writer, Stream} are now sans-I/O: poll_decode,
poll_decode_maybe, poll_read_chunk, poll_read_exact, poll_closed on
the reader; a buffered encode (buffer + poll_flush) plus poll_write,
poll_write_all, and poll_closed on the writer; poll_open/poll_accept
on the pair. The async methods are thin poll_fn wrappers, so the
existing tests exercise the poll paths. Buffered writes also make a
flush resumable mid-message where the old encode cleared partial
bytes, covered by new writer tests.
- Every select-style race loop polls the reader, writer, and session
directly instead of pinning closed()/decode() futures across arms,
removing the stored-future pattern behind the closed-watch ownership
deadlock this PR previously fixed in the adapters.
- TaskSet::drive takes a poll closure instead of a future, so the
accept loops poll the transport in place with nothing to cancel or
rebuild when a child finishes.
- The lite session driver is now a state machine: SendSetup and
SendGoaway replace the boxed setup/goaway tasks, the legacy session
stream is an inert poll arm, and goaway enforcement is a reusable
Enforce machine (the ietf driver uses it through the async wrapper
for now).
No wire changes and no public API changes; everything touched is
crate-private to moq-net.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
The lite publisher half no longer contains an async fn. Publisher::poll accepts control streams and drives each as a child state machine (announce, subscribe, fetch, track-info, probe, goaway), replacing the TaskSet of boxed futures. Group serving is a per-group machine that applies queue and SUBSCRIBE_UPDATE priority changes on every poll pass. The children live in a new PollSet: the poll-native counterpart of FuturesUnordered, keyed by per-child wakers so a wakeup re-polls only the child it was aimed at. A session serving hundreds of subscriptions polls one machine per event, not all of them. The lite session driver now embeds the publisher machine directly instead of boxing a future around it. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
The uni-stream accept loop, GROUP ingest, datagram receive, and PROBE feedback are now state machines driven by SubscriberDriver::poll; only the announce half still runs as a boxed async future. GROUP and FETCH payloads share one FrameIngest machine. Streaming a frame across polls needs the frame producer stored in machine state, which the borrowed frame::Producer<'_> cannot do, so the model grew a crate-private owned variant (frame::ProducerOwned via group::Producer::create_frame_owned): same shared state through a group clone, with the public borrowed API and its one-live-frame guarantee unchanged. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
…ines The announce prefixes, per-source serve loops, TrackServe (TRACK_INFO, the upstream SUBSCRIBE lifecycle, fetches), and the establish handshake are all state machines now; the lite subscriber contains no async fn outside test shims. The cross-task submission channel (Tasks/TaskSet) became a kio::Queue of source descriptors drained by the driver into a PollSet, and err_only left the lite half entirely. One timing subtlety preserved: the driver holds its own clone of the connection-progress producer until its first poll, so connect() still drives the session at least once before resolving (the old announce task's drop(connecting) did this implicitly). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
The moq-transport publisher serves each track through a TrackServe machine with one GroupServe machine per group in a PollSet, replacing the FuturesUnordered of boxed group futures. The subscriber's subgroup ingest is a GroupIngest machine sharing the owned frame producer, so no group stream on either wire protocol pins or boxes a future anymore. The last async chunk wrappers on the coding Reader/Writer went with it. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
The machine conversion conflicted with three fixes that landed on dev (#2740, #2720, and the close-consumes-writer change), so the rebase resolved those files toward the machines and this re-ports the dev semantics onto them: - The lite and ietf announce cursors register the split-horizon peer (origin.excluding), and pre-restart lite versions hold the announced consumer so the ExclusionGuard outlives the Active (#2740). Without the ietf half, unknown_publisher_does_not_flap_across_an_ietf_cluster_triangle flaps exactly as before that fix. - Group streams convert the queue rank through PriorityHandle::send_order and the ietf intake converts the wire priority via priority::from_wire, so higher-priority tracks are transmitted first (#2720). Dev's regression tests are ported onto the machines. - A cleanly finished group releases its stream once the peer acknowledges the FIN: Writer grows poll_close (the poll-native close()), so the Drop fallback cannot reset an acknowledged stream and discard bytes still retransmitting. Covered by the ported completed_group_does_not_reset. - The dev test doubles (DeadRecv, DeadStreamSession) implement the poll traits the crate now requires. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
The transitional async-to-poll adapter discarded the result of a write that its send-side poll_closed drove to completion: the state went back to Idle and the caller, told Pending earlier, retried the same bytes, putting them on the wire twice. The group machines poll the peer-close watch before flushing on every pass, so under qmux backpressure the duplicated write was the group header, which the receiving relay parsed as a phantom ts=0 size=2 frame carrying the header's own subscribe and sequence bytes. That is the frame corruption behind cluster_diamond_goaway_seamless_failover ([0, 22] instead of the payload); the whole cluster suite runs over tcp:// and therefore this adapter. The completed write's length is now recorded and handed to the next poll_write instead of being dropped. Also from the same review of the adapters (both the native one and the moq-wasm mirror): - A STOP_SENDING issued while a read is in flight is applied when that read settles, not parked until a later read that may never come. - The receive closed-watch reads through undelivered payload into a bounded (64 KiB) read-ahead queue, so a FIN right behind buffered data resolves the watch without an external read; reads drain the queue in order. - The module docs spell out the remaining send-side limitation (a peer reset of an idle, unfinished stream is only observed on the next write), which a native poll implementation upstream removes. - doc/lib/rs/env/wasm.md stops advertising web_transport::ClientBuilder as a working wasm path: the browser crate is promise-only today, so the page points at the in-tree adapter instead. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
A poll-based transport may wake its own waker before returning Pending. PollSet's inner loop re-took the ready queue until it drained, so such a child was re-polled forever inside one parent poll, never yielding to the session driver's other arms. One snapshot is processed per poll; a child woken mid-pass has already woken the parent through its ChildWaker, so the executor re-polls the set on the next turn. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
4f8a71e to
749c89a
Compare
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 749c89a608
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| if self.buffer.len() + self.queued_len >= READ_AHEAD_CAP { | ||
| // The caller's own reads are what drain the backlog, and their | ||
| // re-poll of this watch is what resumes it. | ||
| return Poll::Pending; |
There was a problem hiding this comment.
When more than 64 KiB remains unread before the peer's FIN, this branch returns Pending without registering any waker, while the async closed().await helper retains the stream's sole mutable borrow and prevents the caller from draining that backlog. The new read-ahead cap is fresh evidence beyond the earlier comment: small payloads now work, but a larger stream still parks the closure watch permanently even if the FIN has already arrived; the mirrored WASM adapter has the same transition. (Written by GPT-5.6 Sol)
AGENTS.md reference: AGENTS.md:L89-L89
Useful? React with 👍 / 👎.
| bound. This repo carries one in `rs/moq-wasm/src/transport.rs` (the adapter | ||
| behind the `@moq/wasm` package); copy it or use `moq-wasm` until | ||
| `web-transport-wasm` implements the poll interface natively. |
There was a problem hiding this comment.
The revised guide still does not provide a compilable Rust/WASM path: its dependency list contains neither moq-wasm nor the direct dependencies needed to copy the adapter, the example calls an undefined transport::connect, and moq-wasm keeps that module private as mod transport. This is fresh evidence after the earlier comment because the replacement text now explicitly recommends copying or using moq-wasm, but neither option exposes the transport identifier shown below. (Written by GPT-5.6 Sol)
AGENTS.md reference: AGENTS.md:L89-L89
Useful? React with 👍 / 👎.
…adapter Two hardening fixes from the adversarial review of the transitional adapter: - A stream dropped while its async operation is still in flight used to drop the future-owned stream, firing the backend's default code-0 reset or stop and losing the deferred code. The peer classifies aborts by that code (Old vs Cancel vs Evicted), so this destroyed error classification exactly under backpressure. Drop now finishes the pending operation on the tokio runtime and applies the deferred reset/finish/stop after it; only without a runtime does the default drop behavior remain. - The write completion recorded for a retrying poll_write now retains the sent chunk (a refcount clone, no extra copy) and debug_asserts that the retry supplies the same bytes, so a contract-violating caller fails loudly in tests instead of silently corrupting the stream. Mirrored in moq-wasm where applicable (the retry verification; wasm streams have no Drop hook, the browser handles stream teardown). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
|
CI status note: the red (written by Fable 5) |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: c844c253e0
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| handle.spawn(async move { | ||
| let (mut stream, _) = fut.await; | ||
| match reset { | ||
| Some(code) => stream.reset(code), |
There was a problem hiding this comment.
When a group is aborted while its write is flow-control blocked, this new Drop path awaits that same write before calling reset; the receive-side path similarly awaits a stalled read before stop. Those operations can remain pending until the deferred terminal action occurs, so the spawned task retains the stream indefinitely and the peer never receives the intended error code. The regression tests mask this by explicitly unblocking the operation after Drop. Apply the terminal action without waiting for the blocked I/O, and test the permanently blocked case. (Written by GPT-5.6 Sol)
AGENTS.md reference: AGENTS.md:L112-L112
Useful? React with 👍 / 👎.
|
|
||
| - [moq-native](https://crates.io/crates/moq-native): Configures QUIC (via [quinn](https://crates.io/crates/quinn) by default, with [noq](https://crates.io/crates/noq) available through the `noq` feature) and TLS (via [rustls](https://crates.io/crates/rustls)) for you. | ||
| - [moq-net](https://crates.io/crates/moq-net) — The core networking layer. Can be used directly with any `web_transport_trait::Session` implementation if you need full control over the QUIC endpoint. | ||
| - [moq-net](https://crates.io/crates/moq-net) — The core networking layer. Can be used directly with any `web_transport_trait::poll::Session` implementation if you need full control over the QUIC endpoint. |
There was a problem hiding this comment.
The modified moq-net dependency bullet retains an em-dash character, but the repository explicitly prohibits em dashes in documentation and other prose. Replace it with a colon, period, or comma before merging. (Written by GPT-5.6 Sol)
AGENTS.md reference: AGENTS.md:L81-L81
Useful? React with 👍 / 👎.
1 participant
Why
web-transport-trait 0.4 shipped a sans-I/O poll surface (
web_transport_trait::poll), implemented natively by quinn and quiche. This PR makes moq-net poll-only: every entry point requires the poll interface, and nothing in moq-net wraps futures to fake it. The second commit is the payoff: with every transport operation directly pollable, the internal drivers shed their stored/pinned futures and become plain poll logic.What
moq_net::transport::poll(new module)transport::poll::{Session, SendStream, RecvStream}(mirroringweb_transport_trait::poll): the poll traits plus the bounds the session machinery needs (Clone + MaybeSend + MaybeSync + 'static), blanket-implemented. Every entry point (Client::connect,Server::accept,Request<S>) bounds on these.accept_uni,open_bi,read_chunk,write_buf,closed, ...) are provided on the traits aspoll_fnwrappers with the same names the async trait used, which kept the internal churn to receiver mutability (&self+Clonebecame&mut self, so each concurrently pending operation gets its own session clone, per the poll contract).Backends
moq_native::transport::Async, a transitional adapter living in moq-native (where async already lives), to be deleted backend by backend as native poll implementations land upstream. moq-relay's WebSocket fallback uses it too.transport::Sessionnewtype, which is the shape a nativeweb-transport-wasmpoll implementation will eventually replace.Poll-native internals (second commit): async logic replaced with poll logic
This is what the breaking bound buys. Before, the drivers were async-first: transport waits were pinned futures (
std::pin::pin!(stream.closed()),pin!(reader.decode_maybe())) polled viawaiter.poll_futureinside everykio::waitrace, and one-shot protocol work was boxed into task sets. Now:coding::{Reader, Writer, Stream}are sans-I/O:poll_decode,poll_decode_maybe,poll_decode_peek,poll_read_chunk,poll_read_exact,poll_closedon the reader; a buffered encode (buffer+poll_flush) pluspoll_write,poll_write_all,poll_closedon the writer;poll_open/poll_accepton the pair. The async methods are thinpoll_fnwrappers, so all existing tests exercise the poll paths. Buffered writes make a flush resumable mid-message (the oldencodediscarded partial progress); new writer unit tests cover both properties.TaskSet::drivetakes a poll closure instead of a future, so accept loops pollpoll_accept_uni/poll_accept_biin place; there is no accept future whose cancel-safety needs auditing when a child completes.lite::session::Driver):SendSetupandSendGoawayreplace the boxed setup/GOAWAY tasks, the legacy session stream is an inert poll arm, and GOAWAY deadline enforcement is a reusablegoaway::Enforcemachine (the ietf driver consumes it through the async wrapper for now). Onekio::waitat the driver boundary is all that remains; everything inside isPoll.Poll-native drivers (commits 3-7): the run loops are state machines
The
kio::waitraces andFuturesUnorderedtask sets are gone from the hot paths; the drivers are hand-written state machines with onekio::waitbridge at the session boundary.poll_set::PollSet: the poll-native counterpart ofFuturesUnordered. Each child machine owns a waker that marks it ready and wakes the parent, so a wakeup re-polls only the child it was aimed at: a session serving hundreds of subscriptions polls one machine per frame, not all of them.Controldispatch machine (announce, subscribe, fetch, track-info, probe, goaway), serves each group through a per-group machine that applies priority updates on every pass, and runs the announce loop as a buffered-writer event machine. The subscriber's announce prefixes, per-source track serving, the upstream SUBSCRIBE lifecycle (TRACK_INFO, establish, updates, fetches), GROUP ingest, datagrams, and PROBE feedback are all machines; the lite half contains noasync fnoutside test shims, anderr_only/TaskSetleft it entirely.frame::Producer<'_>cannot do. The model grew a crate-private owned variant (frame::ProducerOwnedviagroup::Producer::create_frame_owned): same shared state through a group clone, with the public borrowed API and its one-live-frame guarantee unchanged.connect()must drive the session at least once before its readiness resolves, so the subscriber driver holds a clone of the connection-progress producer until its first poll (the old announce task'sdrop(connecting)did this implicitly).Internals that became poll-native in the first commit
ControlStreamAdapterand its virtual streams implement the poll traits directly over their kio queues (no more pinned futures insideaccept_bi).SendBandwidthpolls the transport close directly instead of boxing aclosed()future.MockSessionpair (rebuilt on kio queues, no tokio channels).Bugs the conversion caught
closed()future deadlocks any later read or write; the old code relied on drop-to-cancel of scopedclosed()watches. The transitional adapters therefore emulate stream closed-watches (reads report the FIN for receive streams; send streams only start the real watch oncefinish/resetmakes them terminal). Caught bygoaway_gates_new_subscribes_moq_lite_04; regression-covered by adapter unit tests in moq-native. The second commit removes the pattern from moq-net itself: no driver stores a transport future at all anymore.match state.lock().unwrap().is_some() { ... }holds the guard through the arms, and an arm took the same lock. It wedged the whole current-thread runtime (even timers), which is why four goaway tests hit the harness timeout instead of their own.Cross-package sync
js/netunaffected.doc/lib/rs/env/{index,native,wasm}.mdupdated for the poll-only requirement.Verification
just check(fmt, clippy-D warnings, nextest over changed crates + dependents): green; 790 moq-net + 253 moq-native tests pass, including new tests for the buffered writer and the PollSet wake granularity.just test smoke: green (relay + CLI end-to-end over real QUIC).cargo check -p moq-wasm --target wasm32-unknown-unknown: green.cargo check: green (not default-members, checked explicitly).Follow-ups (separate PRs)
moq_native::transport; same forweb-transport-wasm, then slimmoq-wasm/src/transport.rsto a plain newtype.run_unis/run_dispatchaccept loops, the namespace loops, the fetch stream, and theControlStreamAdapterread/write halves) to machines the same way, then deleteTaskSet/Tasks/err_onlyfromutil.rs. The lite half and both group data planes are already fully converted.Targets
dev: the public bounds onClient::connect/Server::accept/Request<S>changed, a semver break.Rebase onto dev + review fixes (second pass)
The stack is rebased onto current
devand three commits landed on top:fix(net): carry the dev regression fixes across the rebase. The machine conversion conflicted with three fixes that landed on dev while this PR was open; the rebase kept the machines and re-ported the dev semantics onto them: the split-horizon announce exclusion + held consumers (#2740, both lite and ietf), the send-order priority inversion (#2720,PriorityHandle::send_order/priority::from_wire), and close-consumes-writer (a newWriter::poll_closereleases the stream once the FIN is acknowledged so the Drop fallback cannot reset it). Dev's regression tests are ported onto the machines.fix(native): report adapter writes completed by the closure watch. This is the root cause of thecluster_diamond_goaway_seamless_failoverCI failure (and confirms the review's duplicate-write finding). The cluster suite runs overtcp://, i.e. qmux through the transitional adapter. The group machines poll the peer-close watch before flushing on every pass; under the failover burst's backpressure,AsyncSend::poll_closeddrove the pending group-header write to completion, discarded its result, and the machine's retry wrote the header a second time. The receiver parsed the duplicate as a phantomts=0 size=2frame carrying the header's own subscribe/sequence bytes, which is exactly the corrupt[0, 22]frame the test caught. The completed write's length is now recorded and reported to the retryingpoll_write. The same commit addresses the rest of the adapter review findings: deferred STOP applies when the in-flight read settles, the receive closed-watch reads through undelivered payload into a bounded (64 KiB) read-ahead queue, the send-side preterminal-closure limitation is documented, anddoc/lib/rs/env/wasm.mdstops advertising a wasm path that no longer compiles. All mirrored in moq-wasm; regression tests for each.fix(net): poll one PollSet snapshot per parent poll. A self-waking child could spin the session driver inside a single parent poll and starve its other arms; one snapshot is processed per poll and the re-queue wakes the parent instead.Verification after the rebase: full moq-net + moq-native suites (1065 tests), the full moq-relay suite (199 tests, including the previously failing
cluster_diamond_goaway_seamless_failoverandunknown_publisher_does_not_flap_across_an_ietf_cluster_triangle),cargo check -p moq-wasm --target wasm32-unknown-unknown,just check, andjust test smoke, all green.(Written by Fable 5)