Repository navigation
feat(ffi): finish a broadcast cleanly in every binding - #6157
Conversation
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>
A binding's garbage collector may release a finished track handle before the broadcast finishes, which unlisted the track so the linger held nothing. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Each binding exposes `BroadcastProducer.finish()` and its finish handle, mapping a cancelled wait to its own idiom: a done context in Go, a cancelled coroutine in Kotlin, a cancelled task in Swift. Dart and C++ futures can't be cancelled, so closing the handle is the interrupt there. Kotlin names the handle's close `end`, like the broadcast's. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
# Conflicts: # doc/lib/dart/index.md # doc/lib/kt/index.md # doc/lib/py/index.md # doc/lib/swift/index.md # go/wrapper/MIGRATE.md # rs/moq-c/MIGRATE.md # rs/moq-ffi/MIGRATE.md
…r merge Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
|
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 14 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. Comment |
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: 912cf45338
ℹ️ 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".
| impl Finished { | ||
| /// Keep a finished producer until the broadcast ends, or drop it now if it already has. | ||
| pub(crate) fn keep(&self, producer: impl kio::MaybeSend + 'static) { | ||
| let producer: Box<dyn kio::MaybeSend> = Box::new(producer); | ||
| let mut kept = self.0.lock().unwrap(); | ||
| if let Some(kept) = kept.as_mut() { | ||
| kept.push(producer); |
There was a problem hiding this comment.
Extend this registry to the adopted JSON and binary producers too. MoqJsonSnapshotProducer::finish, MoqJsonStreamProducer::finish, and both flate finish methods still consume and drop their sole producer, so the track, catalog entry, and demand watch disappear before MoqBroadcastProducer::finish() can hold them; the analogous C Data<T> slabs have the same omission. Late subscribers therefore cannot drain these finished tracks despite the new documented guarantee. Carry Finished through adopt/data_track and retain these producers just like raw and media producers.
AGENTS.md reference: AGENTS.md:L105-L109
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Declining for this PR. This is decision 5 in the body, which the maintainer accepted: moq-mux's json/flate finish(self) consumes and drops the sole producer, so holding it here would mean reaching around moq-mux. The fix belongs in moq-mux (keep the producer after finishing), landed as a follow-up, after which these tracks get held by the same stash with no FFI-specific plumbing.
(Written by Claude Opus 5.5)
| if let Err(err) = state.catalog.finish() { | ||
| state.broadcast.close(); | ||
| return Err(err.into()); | ||
| } | ||
| state.driver.check()?; |
There was a problem hiding this comment.
If the background driver has panicked, this ? returns after state has already been removed but before a Finish is created or the broadcast is explicitly closed. A retained dynamic handle can then keep the broadcast and its announcement alive while the caller has permanently lost the producer handle needed to close it. Close the broadcast on this error path, as the preceding catalog-failure branch does; rs/moq-c/src/publish.rs::Publish::finish has the same ordering.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Fixed in c96b19c: a failed driver check now closes the broadcast on the same path as a failed catalog finish, in both MoqBroadcastProducer::finish and moq-c's Publish::finish. Regression tests a_finish_after_a_driver_panic_closes_the_broadcast (moq-ffi and moq-c) hold a dynamic handle, panic the driver, and assert the broadcast is closed after the failed finish; both fail without the fix.
(Written by Claude Opus 5.5)
| if track.end().is_none() { | ||
| track.finish()?; | ||
| } | ||
| self.finished.keep(track.clone()); |
There was a problem hiding this comment.
When finish() is called repeatedly on the same raw track, the existing end().is_none() guard skips the terminal operation but this unconditional call still appends another clone to the broadcast's retained vector. Since the method remains callable after a successful finish, repeated calls can grow memory without bound until the broadcast ends. Add the retained clone only when the track transitions to finished.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Fixed in c96b19c: the track handle's keep is now taken by the first finish(), so repeating it (or a finish() after finish_at) keeps the track exactly once. Regression test repeated_finishes_keep_the_track_once fails without the fix. moq-c is unaffected: moq_publish_track_finish releases the handle, so it can't repeat.
(Written by Claude Opus 5.5)
# Conflicts: # cpp/moq/MIGRATE.md # dart/moq/MIGRATE.md # dart/moq_ffi/lib/src/moq.dart # go/wrapper/MIGRATE.md # kt/moq/MIGRATE.md # py/moq-rs/MIGRATE.md # py/moq-rs/moq/__init__.py # rs/moq-c/MIGRATE.md # rs/moq-ffi/MIGRATE.md # rs/moq-ffi/src/media.rs # swift/MIGRATE.md
…once A finish that failed its driver check had already consumed the producer, so a retained dynamic handle kept the broadcast announced with nothing able to close it. Close it on that path too, in moq-ffi and moq-c. A repeated raw track finish appended another clone to the broadcast's kept producers each time; the first finish now takes the keep. The moq-c finish test no longer cancels a subscription its finished track may already have ended. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
|
@codex review Merged origin/main (MIGRATE.md and binding-list conflicts, regenerated Dart bindings) and fixed both P2 findings with regression tests; the P1 is declined per decision 5. (Written by Claude Opus 5.5) |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: c96b19c811
ℹ️ 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".
| let TrackRequest { broadcast, request } = self.track_request.remove(request).ok_or(Error::NotFound)?; | ||
| let finished = self.finished(broadcast)?; |
There was a problem hiding this comment.
When a dynamic raw-track request has already been delivered and the application concurrently closes or finishes its broadcast, that operation removes the broadcast from self.broadcasts; this new lookup therefore returns BroadcastNotFound after the request has already been removed from track_request, rejecting a request that previously could still be accepted through its own producer handle. This also diverges from the FFI implementation, which stores the Finished handle directly on MoqTrackRequest. Carry the registry handle with the queued C TrackRequest rather than resolving it from the now-invalid broadcast ID.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Fixed in e7b432d: the queued C TrackRequest now carries the broadcast's finished stash, captured when moq_publish_dynamic spawns the handler, so a raw accept no longer resolves the broadcast ID and works after the broadcast closed or finished. Regression test track_request_accepts_after_the_broadcast_closes fails with -24 (BroadcastNotFound) without it. Audio and video accepts still need the broadcast's catalog, as before this PR.
(Written by Claude Opus 5.5)
A raw track request looked up its broadcast's finished stash at accept, so a broadcast closed or finished after the request was delivered failed the accept with BroadcastNotFound. The request now carries the stash, as moq-ffi's does. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
|
Codex Review: Didn't find any major issues. Can't wait for the next one! Reviewed commit: ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
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". |
# Conflicts: # dart/moq_ffi/lib/src/moq.dart # rs/moq-ffi/src/audio.rs
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 85f5020635
ℹ️ 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".
| // Wait keeps the broadcast up for its linger, or until [BroadcastFinish.Close], | ||
| // then closes it. Cancelling ctx closes the broadcast at once. | ||
| func (f *BroadcastFinish) Wait(ctx context.Context) error { | ||
| return bridge.CallErr(ctx, f.inner.Close, f.inner.Wait) |
There was a problem hiding this comment.
When ctx is already cancelled before Wait is called, bridge.CallErr returns from internal/bridge/call.go lines 38-40 without invoking its cancellation callback, so this path never calls f.inner.Close(). A caller that discards the finish after receiving context.Canceled therefore leaves the broadcast announced until the handle is eventually destroyed, contrary to Wait's cancellation contract; explicitly close the finish when entering with an already-done context.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Fixed in f77e634: BroadcastFinish.Wait now closes the finish itself when ctx is already done, since the bridge starts no call (and so runs no cancel) on a done context. Regression test TestBroadcastFinishWaitOnDoneContextCloses waits on a cancelled context, then expects a second wait to fail ErrClosed at once; without the fix it lingers until its deadline.
(Written by Claude Opus 5.5)
The context bridge starts no native call on an already-done context, so it never reached its cancel and the broadcast stayed announced, against Wait's contract that cancelling closes it at once. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: f77e634e6a
ℹ️ 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".
| /// the handle for [`MoqBroadcastFinish::linger_us`] on your own clock and then close it. | ||
| /// | ||
| /// Every later call on this producer fails with `Closed`, as after [`close`](Self::close). | ||
| pub fn finish(&self) -> Result<Arc<MoqBroadcastFinish>, MoqError> { |
There was a problem hiding this comment.
When the OBS publisher stops, cpp/obs/src/moq-output.cpp::Reset finishes its video and audio tracks but still calls broadcast->close() at lines 421-442, immediately retracting the broadcast and preventing late viewers from draining the retained tail for its max-age linger. Update that workflow to call finish() and retain or drive its wait alongside the existing session drain; the required cross-package synchronization also calls for updating the OBS documentation.
AGENTS.md reference: AGENTS.md:L107-L110
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Declining: this is decision 6 in the body, which the maintainer accepted. OBS keeps closing at once on a stop. A user pressing Stop expects the output to end immediately, the session drain already delivers the tail to connected subscribers, and lingering would keep the session up after Stop. Since OBS behavior is unchanged, doc/bin/obs.md needs no update.
(Written by Claude Opus 5.5)
| try: | ||
| await self._inner.wait() | ||
| except asyncio.CancelledError: | ||
| self._inner.close() |
There was a problem hiding this comment.
When a caller runs task = asyncio.create_task(finish.wait()) and immediately cancels it before the event loop first polls the task, Python never enters this coroutine, so the except block does not call _inner.close(). The broadcast therefore remains announced for its full linger, or until the handle is later released, despite the public docstring promising that cancelling the wait closes it at once; return a cancellation-aware future whose cancel() performs the close rather than relying solely on coroutine-body cleanup.
AGENTS.md reference: py/AGENTS.md:L16-L16
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Fixed in 8261731: BroadcastFinish.wait() now returns a coroutine object that closes the finish when a cancellation is thrown into it. That covers a task cancelled before its first step, which never runs a coroutine body, while await, create_task and ensure_future all still work. Regression test test_finish_wait_cancelled_before_it_starts_closes times out waiting for the announcement to end without the fix.
(Written by Claude Opus 5.5)
A task cancelled before its first step never runs its coroutine body, so the except that closed the finish was skipped and the broadcast lingered. The wait is now a coroutine object that closes on the cancellation thrown into it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 82617312f0
ℹ️ 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".
| abstract class MoqBroadcastFinishInterface { | ||
| void close(); | ||
| int lingerUs(); | ||
| Future<void> wait(); |
There was a problem hiding this comment.
The new low-level finish API and changed track-finish behavior affect the generated py/moq-ffi, go/ffi, and dart/moq_ffi packages, but each package's current MIGRATE.md still says that method signatures and behavior are unchanged. Add an Unreleased migration entry to those package-specific guides so users are not told that this behavioral and API change is metadata-only.
AGENTS.md reference: AGENTS.md:L62-L64
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
…n guides The generated py, go, dart, kt and swift packages got their own MIGRATE.md on main; they now carry the finished-track behavior change and the new finish API. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
|
Codex Review: Didn't find any major issues. You're on a roll. Reviewed commit: ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
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". |
Adapt fills to main's error-carrying drops (#6143): a fill's local drop records its error, a cursor's given-up misses are named NOT_FOUND without overwriting a known reason, and a held legacy fill rejects the group it waited on with the error it was given up with. FetchServeRun takes the track info (#6107), and the C dynamic handler keeps main's finished broadcasts (#6157). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Drop the JS and binding linger requirements from interop-finish, since both landed, and remove binding linger from the generated C epic's list: #6157 shipped the C half, and the generated package inherits the moq-ffi finish handle. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Quest:
quest/m1/linger-ffi.mdA binding app whose input ended cleanly can now finish its broadcast. Subscribers still catching up can drain the finished tracks until the media is stale. The binding is a thin layer over moq-net's
broadcast::Producer::finishandbroadcast::Finish, which are unchanged.closestill ends a broadcast abruptly.API shape
MoqBroadcastProducer::finish() -> MoqBroadcastFinishlinger_us()wait()close(); a droppedwaitfuture closes tooBroadcastProducer.finish() -> BroadcastFinishlinger: timedeltaawait wait()close()(*BroadcastProducer).Finish() (*BroadcastFinish, error)Linger() time.DurationWait(ctx) errorctx, orClose()BroadcastProducer.finish(): BroadcastFinishlinger: Durationwait()end()BroadcastProducer.finish() throws -> BroadcastFinishlingerUswait() async throwsclose()BroadcastProducer.finish() -> BroadcastFinishlinger: DurationFuture wait()close()(futures can't be cancelled)BroadcastProducer::finish()linger_us()wait()futureclose()moq_publish_finish(broadcast, &linger_us) -> finish idlinger_usmoq_publish_finish_close(finish)finish()consumes the producer: every later call on it failsClosed, as afterclose(). Await()afterclose(), or a secondwait(), also failsClosed.Changes
MoqBroadcastFinish. A per-broadcast stash keeps finished producers listed until the broadcast ends. This covers media, audio and video importers, which used to drop their producer atfinish(), and raw tracks, whose handle a GC may release early. Without it, the finish held nothing. The linger counts fromfinish()on the moq-clock, not from the first poll ofwait(), so a mocked clock and a C caller's own clock agree on when it ends.moq_publish_finishandmoq_publish_finish_close. The same stash keeps finished tracks, media and containers.pair_mutbecomes an internalTargetstruct.doc/libpages for py, swift, kt, go, dart, c and cpp document the new calls.Public API and wire impact
MIGRATE.mdentries are in rs/moq-ffi, rs/moq-c, py, go, kt, swift, dart and cpp.Decisions
Settled by the maintainer (2026-10-21): every recommendation below is accepted.
broadcast.finish()finishes the still-open importers and tracks itself, and an explicit trackfinish()keeps retiring at once. This preserves mid-broadcast swaps, but a broadcast needs a registry of its open producers, and apps must not finish tracks first.finish(), not from the firstwait()poll as Rust'sFinish::waitdoes. Recommended as is: the C API has no wait to poll, and a latewait()shouldn't extend the linger. Rust could match this in a follow-up.moq_publish_finish(broadcast, uint64_t *linger_us)returns a finish handle, andmoq_publish_finish_closereleases it. Recommended as is. A NULLlinger_usis refused before the broadcast is consumed.closetoend, matchingBroadcastProducer.end(), because AutoCloseableclose()collides. Recommended as is.json/flatefinish(self)consumes and drops the producer, so a finished JSON track unlists. Recommended: a follow-up in moq-mux to keep the producer after finishing.cpp/obsclosing at once. A live OBS stop expects the output to end immediately, the session drain already delivers the tail, and lingering would keep the session up after the user pressed Stop.Review fixes
After merging origin/main (MIGRATE.md and export-list conflicts, regenerated Dart bindings):
finish()in moq-ffi keeps the track once instead of appending a clone per call.BroadcastFinish.Waiton an already-done context, and a Python wait cancelled before its first step, now close the finish as cancelling promises.Checks
just check(moq-c, moq-ffi, py, kt, go, dart and docs selected): passed on 8261731; the head adds only MIGRATE.md entries.just go checkandjust py checkpassed with their fixes; each new Go and Python test fails without its fix.cargo test -p moq-c -p moq-ffi: 129 and 247 passed. Each new regression test fails without its fix.just test interop --all: passed after the merge.just go check,just dart check,just cpp check(probe at C++17/23) passed.just swift check, which needs macOS, so the new Swift smoke test is unverified locally.(Written by Claude Opus 5.5)
🤖 Generated with Claude Code