Repository navigation
feat(net): send the first FETCH alongside the track's info request - #7332
Conversation
Quest-Claim: 0f1c7f0952e9f4c5
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
kixelated
changed the title
|
Warning Review limit reachedYou've used all free OSS reviews for now. Wait for the free limit to reset to keep reviewing this public repository. Next included review available in 5 minutes. View limit details
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. |
Codex Review SummaryThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
![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: af8bfc8400
ℹ️ 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".
kixelated
left a comment
kixelated
left a comment
There was a problem hiding this comment.
Automated review by review (OpenAI)
Reviewed commit: af8bfc8.
P2 remains: queued FETCHes can starve their prerequisite TRACK of bidirectional stream credit (rs/moq-net/src/lite/subscriber.rs:7189–7193). The fetch tasks are polled before TRACK opens, consume credit, then wait for its info; TRACK's fail-fast open can consequently reject the track and all fetches. This matches the existing finding discussion_r4423465440, referenced rather than duplicated inline. Drive TRACK opening first, without awaiting its response, and add a regression with queued distinct-group fetches and constrained stream credit.
The overall direction is sound: stage the request early but gate delivery on verified info. Staged-copy identity, withdrawal/refetch, learned-info cancellation, and replacing fetched backfill with the live feed are coherent. No additional actionable bugs found in this pass.
Verification limits: GitHub source/test inspection and an independent static review only. I did not run tests, interop, CI, or the reported latency measurements.
# Conflicts: # rs/moq-net/src/ietf/subscriber.rs
TRACK opens its stream with `poll_open_request`, which fails fast without stream credit, while a fetch waits for credit. Polling queued fetches first could spend the last credit and fail the track with every fetch. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
|
Landing summary:
(Written by Claude Opus 5.5) |
kixelated
left a comment
kixelated
left a comment
There was a problem hiding this comment.
Automated review by review (OpenAI)
Reviewed commit: 29bfb21.
The prior P2 TRACK stream-credit starvation finding is fixed: TrackServeRun now drives TRACK opening before polling queued FETCHes, and the new regression checks that ordering. No new actionable bugs found in this update.
The direction remains sound: pipeline requests without exposing fetched objects before verified track info. The IETF merge adaptation also requests properties while the timescale is unknown, preserving timestamp delivery without adding another state machine.
Verification limits: static inspection of the fix, regression, merge adaptation and relevant surrounding code, compared with af8bfc8 and separating main-branch changes. I did not run builds, tests, interop or latency measurements.
|
quest land removed this pull request from the queue: Conflicts with the base: rs/moq-net/src/lite/subscriber.rs |
|
Codex land fixer: .encode(&mut crate::coding::Encoder::new(&mut bytes, version.into()), version)
.unwrap();
let mut w = crate::coding::Encoder::new(&mut bytes, version.into());
w.varint(0).unwrap();
w.varint(1).unwrap();
w.u8(42);
let mut setup = Vec::new();
lite::DataType::Setup
.encode(&mut crate::coding::Encoder::new(&mut setup, version.into()), version)
.unwrap();
lite::Setup::default()
.encode(&mut crate::coding::Encoder::new(&mut setup, version.into()), version)
.unwrap();
let session =
crate::lite::test_transport::ScriptedSession::eof(Vec::new()).with_incoming_unis(if pre_read {
vec![Vec::new(), vec![42], bytes, setup]
} else {
vec![bytes]
});
let log = session.log.clone();
let peer_setup = lite::PeerSetup::new(false, crate::transport::Address::Setup);
let early = if pre_read {
let accepted = lite::accept_setup(&mut crate::driver::Transport::new(session.clone()), version)
.await
.unwrap();
assert_eq!(accepted.early.len(), 2);
assert!(log.stops().is_empty());
peer_setup.set(accepted.setup);
accepted.early
} else {
Vec::new()
};
let subscriber = Subscriber::new(SubscriberConfig {
runtime: crate::time::sim(),
session: crate::driver::Transport::new(session.clone()),
origin: origin::Config::new(crate::Hop::new(1).unwrap()).produce(),
recv_bandwidth: None,
version,
peer_setup: peer_setup.clone(),
peer_hop: None,
cost: None,
going_away: Default::default(),
auth: crate::auth::Link::new(false),
});
// The track is untimed where the subscriber drops received timestamps.
let track = track::Producer::new(
std::sync::Arc::new(crate::broadcast::Info::default()),
"video",
crate::track::Info::default().with_timescale(version.has_media_time().then_some(Timescale::MILLI)),
);
let mut consumer = track.consume().subscribe(None).await.unwrap();
subscriber.subscribes.lock().insert(
7,
TrackEntry {
producer: track,
framing: lite::Framing {
timescale: Some(Timescale::MILLI),
ietf: None,
compressed: false,
},
tail: Default::default(),
},
);
let mut accept = UniAccept::new(subscriber, early);
assert!(accept.poll(&kio::Waiter::noop()).is_pending());
assert!(log.stops().is_empty());
let mut group = consumer.recv_group().await.unwrap().unwrap();
assert_eq!(group.sequence, 3);
let frame = group.read_frame().await.unwrap().unwrap();
assert_eq!(frame.payload.as_ref(), &[42]);
assert!(group.read_frame().await.unwrap().is_none());
}
}
}
/// A group at or past the end the publisher declared contradicts that end, which no later
/// stream can repair, so the whole track fails rather than ending clean without it. lite-05
/// specified an inclusive end, so there it costs only that group's stream.
#[moq_clock::test]
async fn a_group_past_the_declared_end_aborts_the_track() {
use crate::transport::poll::Session as _;
for version in [Version::Lite05, Version::Lite06, Version::Lite07] {
thread 'codex-main' (86) panicked at /rustc/48a229ceaefd4985c50990b14116b6d856af0985/library/std/src/io/stdio.rs:1166:9:
failed printing to stderr: Resource temporarily unavailable (os error 11)
stack backtrace:
0: __rustc::rust_begin_unwind
1: core::panicking::panic_fmt
2: std::io::stdio::_eprint
3: <codex_exec::event_processor_with_human_output::EventProcessorWithHumanOutput as codex_exec::event_processor::EventProcessor>::process_server_notification
4: codex_exec::run_exec_session::{closure#0}
5: <tracing::instrument::Instrumented<codex_exec::run_exec_session::{closure#0}> as core::future::future::Future>::poll
6: codex_exec::run_main::{closure#0}
7: codex::cli_main::{closure#0}
8: <tokio::runtime::park::CachedParkThread>::block_on::<codex_arg0::run_main_with_arg0_guard<codex::main::{closure#0}, codex::main::{closure#0}::{closure#0}>::{closure#0}>
9: tokio::runtime::context::runtime::enter_runtime::<<tokio::runtime::scheduler::multi_thread::MultiThread>::block_on<codex_arg0::run_main_with_arg0_guard<codex::main::{closure#0}, codex::main::{closure#0}::{closure#0}>::{closure#0}>::{closure#0}, core::result::Result<(), anyhow::Error>>
10: <tokio::runtime::runtime::Runtime>::block_on::<codex_arg0::run_main_with_arg0_guard<codex::main::{closure#0}, codex::main::{closure#0}::{closure#0}>::{closure#0}>
note: Some details are omitted, run with `RUST_BACKTRACE=full` for a verbose backtrace.
quest: fixer agent exited 101
(written by Codex) |
|
quest land removed this pull request from the queue: Conflicts with the base: rs/moq-net/src/ietf/publisher.rs, rs/moq-net/src/ietf/subscriber.rs, rs/moq-net/src/lite/subscriber.rs |
Resolve the subscriber conflicts with the pipelined first FETCH (#7332): the Info state carries both the pipelined SUBSCRIBE and the fetches sent alongside. Follow the moq_clock -> kio::clock rename. Delete the finished pipeline-requests epic and measure the first fetch over two relays. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Finishes quest/m2/pipeline-requests/fetch.md and deletes it.
A fetch-only reader's first FETCH now goes upstream together with the request for the track's info (TRACK on moq-lite, TRACK_STATUS on moq-transport) at every hop. Before, it went out one round trip later. The fetched group is still handed out only once its route's info is known and passes the origin's consistency check.
Changes
model/resume.rs,model/origin.rs,model/track.rs): the front stages the copy it queries (resume::Producer::stage). A fetch with no serving copy goes to the staged copy at once and holds its answer until that copy is spliced. If the copy is withdrawn first (info refused or mismatched, source detached, or nobody reads the track), the fetch is dropped and re-issued on the next route. The logical track gets its routes when it is assigned rather than at accept, so a fetch made before accept already reaches the staged copy. A track ended before any copy was spliced now ends its routes too, so a pending fetch fails with the front's error.model/track.rs): a live (subscription) copy of a group now takes the slot from a fetch-only backfill of the same group instead of being refused as a duplicate. Readers of the fetched copy keep their handle. Without this, a relay whose FETCH beat its SUBSCRIBE could cache group 0 invisibly to subscribers and then abort it when the fetch was abandoned. The race already existed (ServeLoopserves fills ahead of SUBSCRIBE); pipelining made it common, andcatalog_churn_lite_07caught it.TrackServeRunregisters its fetch handler up front and sends FETCH alongside TRACK. TRACK opens its stream first: it fails fast without stream credit, while a fetch waits for credit, so fetches opened first could starve it. The response stays unread in the stream until TRACK_INFO lands. A failed TRACK_INFO rejects the fetch with that error and resets its stream.resume.ts) asks the route for the group alongside#verifyand hands it out only once the check passes.Deadlinenow counts concurrent opens, so the two streams share one request deadline.util::Later/util::Learning: a value one request learns for the requests sent alongside it, shared by both Rust sessions.Measurements
Simulated link with 100 ms RTT, time to the first fetched group once the broadcast resolved (
fetch_pipeline::the_first_fetch_takes_one_round_trip_per_hop, run onorigin/mainand on this branch):The test now asserts one round trip per hop.
Tests
track_opens_before_the_fetches_sent_alongside: TRACK's stream opens before a FETCH queued ahead of the run's first poll (fails on the old order).fetch_pipeline.rs, lite-05, lite-07, moq-transport-18, moq-transport-22, direct and through a relay): the FETCH reaches the publisher before it answers the info; a refused info fails the fetch; the latency bound above.just test interop --allpasses across the Rust/JS pairs.just checkpasses. Fullmoq-netsuite: 3189 passed.Choices made without asking
fetch_rangealready widened the same way for a copy with unknown info.properties_wanted: true); once the timescale is known it asks only for a timed copy, as on main. Found while merging main.Cancel), not the front'sUnroutable, matching a fetch on a spliced copy.Impact
Public API: none. Wire: none (ordering only).
Coordination
The sibling moq-dev/moq#7328 (SUBSCRIBE with TRACK) held only its claim commit when this started. Overlap is limited to
lite/subscriber.rs(TrackServeRun/ServeLoopconstructors) and the epic's link lists; whichever lands second merges.Follow-ups
Review
mechanical fix by codex / Codex:
Merged base, preserving both conflict sides; followed clock/request renames and retained the unchanged fill-refusal helper; test execution blocked by missing local dependencies.(head 4894026)quest review --deep), gpt-6-astra, medium effort.ietf/subscriber.rs, plus the TRACK-first fix).just check origin/mainpasses on 29bfb21.(Written by Claude Opus 5.5)
🤖 Generated with Claude Code