diff --git a/dart/moq_ffi/lib/src/moq.dart b/dart/moq_ffi/lib/src/moq.dart index 87e4145ef83..8b249550d5e 100644 --- a/dart/moq_ffi/lib/src/moq.dart +++ b/dart/moq_ffi/lib/src/moq.dart @@ -20452,7 +20452,7 @@ void _checkApiChecksums() { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqmediacontainerproducer_finish() != - 25121) { + 40580) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqmediacontainerproducer_seek() != 6771) { @@ -20488,7 +20488,7 @@ void _checkApiChecksums() { 25213) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } - if (uniffi_moq_ffi_checksum_method_moqmediatrackproducer_finish() != 45426) { + if (uniffi_moq_ffi_checksum_method_moqmediatrackproducer_finish() != 22633) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqmediatrackproducer_seek() != 37244) { @@ -20503,7 +20503,7 @@ void _checkApiChecksums() { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqmediatrackstreamproducer_finish() != - 2377) { + 43608) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqmediatrackstreamproducer_write() != @@ -20577,7 +20577,7 @@ void _checkApiChecksums() { if (uniffi_moq_ffi_checksum_method_moqbroadcastproducer_announce() != 3965) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } - if (uniffi_moq_ffi_checksum_method_moqbroadcastproducer_close() != 19191) { + if (uniffi_moq_ffi_checksum_method_moqbroadcastproducer_close() != 26090) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqbroadcastproducer_consume() != 27634) { @@ -20689,7 +20689,7 @@ void _checkApiChecksums() { if (uniffi_moq_ffi_checksum_method_moqtrackrequest_abort() != 62713) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } - if (uniffi_moq_ffi_checksum_method_moqtrackrequest_accept() != 31841) { + if (uniffi_moq_ffi_checksum_method_moqtrackrequest_accept() != 26498) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqtrackrequest_demand() != 3618) { diff --git a/doc/lib/c/index.md b/doc/lib/c/index.md index 4174047f9c0..659e7b376ab 100644 --- a/doc/lib/c/index.md +++ b/doc/lib/c/index.md @@ -67,7 +67,7 @@ inside. The header is the reference: each function carries a doc comment. - **Compressed tracks.** `moq_track_info.compress` on a raw track sends each group DEFLATE-compressed to opt-in `moq-lite-07-wip` peers; other peers, and every reader here, get the original frames. `@moq/net` cannot read a compressed track yet. - **fMP4 import.** `MOQ_CONTAINER_FORMAT_FMP4` splits each fragment into one frame per sample. - **Audio needs cuts.** Video groups at its keyframes, but audio forms a group only where you call `moq_publish_media_cut`: after every frame, or at a segment cadence to align with video. -- **Ending cleanly.** When the input ends, finish every track, then call `moq_publish_catalog_finish()` to finish the catalog, which ends the content. The broadcast stays announced and keeps serving the finished tracks, so a subscriber still catching up can drain them, until `moq_publish_close()` takes it offline. How long to keep it up is yours to decide. +- **Ending cleanly.** When the input ends, finish every track, then call `moq_publish_catalog_finish()` to finish the catalog, which ends the content. The broadcast stays announced and keeps serving the finished tracks, so a subscriber still catching up can drain them, until `moq_publish_close()` takes it offline. How long to keep it up is yours to decide. A track still open at the close ends with `UNROUTABLE`, which readers take as the publisher leaving, and its handle is released. - **Timeline breaks.** On a seek or pause, call `moq_publish_media_discontinuity`, then keep timestamps moving forward and resume video on a keyframe. - **JSON values.** `moq_consume_json_value` fills `moq_json_value` with the document and its frame's `timestamp_us`, rounded down to a whole microsecond; `timestamp_present` is false on an untimed track. A snapshot consumer's callback fires for each state in order, but a group superseded before it is read is skipped. - **Decoded frames hold decoder buffers.** Free each frame id with `moq_decode_video_frame_free` promptly, or the decoder stalls. diff --git a/doc/lib/cpp/index.md b/doc/lib/cpp/index.md index 9824120a11b..6aa6c0bd368 100644 --- a/doc/lib/cpp/index.md +++ b/doc/lib/cpp/index.md @@ -227,7 +227,9 @@ To end a broadcast cleanly, finish every track, then call `catalog->finish()` on its `moq::MediaCatalogProducer` to finish the catalog, which ends the content. The broadcast stays announced and keeps serving the finished tracks, so a subscriber still catching up can drain them, until `broadcast->close()` takes it -offline. How long to keep it up is yours to decide. +offline. How long to keep it up is yours to decide. A track still open at the +close, or released without a finish, ends with `Unroutable`, which readers take +as the publisher leaving. `moq::shutdown()` stops the `moq-ffi` runtime workers, then the executor. Pending calls resolve `Cancelled`, and every object stays safe to drop. Call diff --git a/doc/lib/dart/index.md b/doc/lib/dart/index.md index 05a9fe8914f..7654e399af7 100644 --- a/doc/lib/dart/index.md +++ b/doc/lib/dart/index.md @@ -142,7 +142,7 @@ shows which module holds each. - **Timeline breaks.** On a seek or pause, call `discontinuity()`, then keep timestamps moving forward and resume video on a keyframe. - **Decoded frames hold decoder buffers.** `dispose()` each frame from a `video.Consumer` promptly rather than waiting for the garbage collector, or the decoder stalls. `resize` is best effort, so read each frame's `width()` and `height()`. - **Closing.** `await moq.close()` drains for up to one second and reports its result. `session.shutdown()` requests the drain synchronously; `await session.closed()` waits for its result. `session.cancel(code: 0)` closes at once. Finish or abort live tracks first. Awaiting cancellation of an announcement, grant, or request subscription releases its pending native read. Announcements also release their cursor; authorization and server listeners leave their shared owner usable. Application delivery requires explicitly selecting `moq-lite-07-wip` on both peers; older Lite versions (including default lite-06) and IETF wait only for transport ACKs. -- **Ending cleanly.** When the input ends, finish every track, then call `finish()` on its `CatalogProducer` to finish the catalog, which ends the content. The broadcast stays announced and keeps serving the finished tracks, so a subscriber still catching up can drain them, until `broadcast.close()` takes it offline. How long to keep it up is yours to decide. +- **Ending cleanly.** When the input ends, finish every track, then call `finish()` on its `CatalogProducer` to finish the catalog, which ends the content. The broadcast stays announced and keeps serving the finished tracks, so a subscriber still catching up can drain them, until `broadcast.close()` takes it offline. How long to keep it up is yours to decide. A track still open at the close, or released without a finish, ends with `Unroutable`, which readers take as the publisher leaving. - **Stats.** `session.stats()` reports `rtt`, `estimatedSendRateBps`, and the byte and packet counters (`bytesSent`, `bytesReceived`, `bytesLost`, `packetsSent`, `packetsReceived`, `packetsLost`). A field is `null` when the transport does not report it, which is not the same as zero. `version` names the negotiated protocol. Connect returns a connecting session; read its coherent `state()` snapshot and wait with that snapshot's revision to observe first-dial failures and reconnects. Each watcher keeps its own revision. Hold the owned `BandwidthConsumer` returned by `session.recvBandwidth()` to receive bandwidth estimates. `rate()` reads the current estimate in bits per second and `rateChanged()` waits for a new one. Release the consumer or call `cancel()` when finished; releasing the last consumer stops PROBE; the session and stats snapshots do not request it. Client consumers survive reconnects and report no estimate while disconnected. - **Authorization.** On `moq-lite-07-wip` (opt in with `ConnectOptions(versions: ...)`) and on moq-transport draft-17+ with the [MoQ Auth extension](/draft/moq-auth), listen to `moq.auth().grants()` for the `Grant` the relay sent back, `null` until it answers and while a client session reconnects. `await auth().add(token: ...)` presents a fresh token without reconnecting, returns an `AuthToken` whose grant joins the union until `cancel()`, and is presented again on every reconnect. A refusal throws with the relay's code. - **Errors.** Read diagnostic payloads from `reason` and structured protocol failures from `details`. diff --git a/doc/lib/go/index.md b/doc/lib/go/index.md index 703a20b27c7..efe33787c57 100644 --- a/doc/lib/go/index.md +++ b/doc/lib/go/index.md @@ -132,7 +132,7 @@ shows which module holds each. - **Timeline breaks.** On a seek or pause, call `Discontinuity()`, then keep timestamps moving forward and resume video on a keyframe. - **Decoded frames hold decoder buffers.** `Close` each `VideoDecodedFrame` promptly, or the decoder stalls. `Resize` is best effort, so read each frame's `Width()` and `Height()`. - **Closing.** `client.Close()` drains for up to one second and returns an error on failure. `session.Shutdown()` requests that drain immediately; `session.Closed(ctx)` waits for its result, and `session.Cancel(code)` closes at once. Finish or abort live tracks first. Application delivery requires explicitly selecting `moq-lite-07-wip` on both peers; older Lite versions (including default lite-06) and IETF wait only for transport ACKs. -- **Ending cleanly.** When the input ends, finish every track, then call `Finish()` on its `media.CatalogProducer` to finish the catalog, which ends the content. The broadcast stays announced and keeps serving the finished tracks, so a subscriber still catching up can drain them, until `broadcast.Close()` takes it offline. How long to keep it up is yours to decide. +- **Ending cleanly.** When the input ends, finish every track, then call `Finish()` on its `media.CatalogProducer` to finish the catalog, which ends the content. The broadcast stays announced and keeps serving the finished tracks, so a subscriber still catching up can drain them, until `broadcast.Close()` takes it offline. How long to keep it up is yours to decide. A track still open at the close, or released without a finish, ends with `Unroutable`, which readers take as the publisher leaving. - **Stats.** `Session().Stats()` reports `RTT`, `EstimatedSendRateBps`, and the byte and packet counters (`BytesSent`, `BytesReceived`, `BytesLost`, `PacketsSent`, `PacketsReceived`, `PacketsLost`). A field is nil when the transport does not report it, which is not the same as zero. `Version` names the negotiated protocol. `Dial` returns a connecting session; read `Session().State()` and wait with `StateChanged(ctx, state.Revision)` to observe first-dial failures and reconnects. Hold the owned `BandwidthConsumer` returned by `Session().RecvBandwidth()` to receive bandwidth estimates. `Rate()` reads the current estimate in bits per second and `RateChanged(ctx)` waits for a new one. Calling `Close()` on the last consumer stops PROBE; the session and stats snapshots do not request it. Client consumers survive reconnects and report no estimate while disconnected. - **Authorization.** On `moq-lite-07-wip` (opt in with `ClientConfig.Versions`) and on moq-transport draft-17+ with the [MoQ Auth extension](/draft/moq-auth), `Session().Auth().Grant()` is the `Grant` the relay sent back, nil until it answers and while a client session reconnects; `GrantChanged(ctx)` waits for the next one. `Add(ctx, token)` presents a fresh token without reconnecting, returns an `AuthToken` whose grant joins the union until `Close()`, and is presented again on every reconnect. A refusal returns the relay's code. - **Errors.** Read diagnostic payloads from `Reason` and structured protocol failures from `Details`. diff --git a/doc/lib/kt/index.md b/doc/lib/kt/index.md index 4155975321f..c5940ee841b 100644 --- a/doc/lib/kt/index.md +++ b/doc/lib/kt/index.md @@ -102,7 +102,7 @@ shows which module holds each. - **Timeline breaks.** On a seek or pause, call `discontinuity()`, then keep timestamps moving forward and resume video on a keyframe. - **Decoded frames hold decoder buffers.** `close()` each frame from a `dev.moq.video.Consumer` promptly (or `use { }` it), or the decoder stalls. `resize` is best effort, so read each frame's `width()` and `height()`. - **Closing.** `moq.shutdown()` requests a drain immediately; suspending `moq.session.closed()` waits up to one second and reports its result. `Moq.close()`, and so `use { }`, cancels at once. Finish or abort live tracks first. Application delivery requires explicitly selecting `moq-lite-07-wip` on both peers; older Lite versions (including default lite-06) and IETF wait only for transport ACKs. -- **Ending cleanly.** When the input ends, finish every track, then call `finish()` on its `CatalogProducer` to finish the catalog, which ends the content. The broadcast stays announced and keeps serving the finished tracks, so a subscriber still catching up can drain them, until `broadcast.end()` takes it offline. How long to keep it up is yours to decide. +- **Ending cleanly.** When the input ends, finish every track, then call `finish()` on its `CatalogProducer` to finish the catalog, which ends the content. The broadcast stays announced and keeps serving the finished tracks, so a subscriber still catching up can drain them, until `broadcast.end()` takes it offline. How long to keep it up is yours to decide. A track still open at the end, or released without a finish, ends with `Unroutable`, which readers take as the publisher leaving. - **Stats.** `session.stats()` reports `rtt`, `estimatedSendRateBps`, and the byte and packet counters (`bytesSent`, `bytesReceived`, `bytesLost`, `packetsSent`, `packetsReceived`, `packetsLost`). A field is `null` when the transport does not report it, which is not the same as zero. `version` names the negotiated protocol. Connect returns a connecting session; read its coherent `state()` snapshot and wait with that snapshot's revision to observe first-dial failures and reconnects. Each watcher keeps its own revision. Hold the owned `BandwidthConsumer` returned by `session.recvBandwidth()` to receive bandwidth estimates. `rate()` reads the current estimate in bits per second and `rateChanged()` waits for a new one. Release the consumer or call `cancel()` when finished; releasing the last consumer stops PROBE; the session and stats snapshots do not request it. Client consumers survive reconnects and report no estimate while disconnected. - **Authorization.** On `moq-lite-07-wip` (opt in with `ClientConfig(versions = ...)`) and on moq-transport draft-17+ with the [MoQ Auth extension](/draft/moq-auth), collect `moq.auth().grants()` for the `Grant` the relay sent back, `null` until it answers and while a client session reconnects. Suspending `auth().add(token)` presents a fresh token without reconnecting, returns an `AuthToken` whose grant joins the union until `cancel()`, and is presented again on every reconnect. A refusal throws with the relay's code. - **Errors.** Read diagnostic payloads from `reason` and structured protocol failures from `details`. diff --git a/doc/lib/py/index.md b/doc/lib/py/index.md index 1799e4881d2..7a31f27d7a8 100644 --- a/doc/lib/py/index.md +++ b/doc/lib/py/index.md @@ -116,7 +116,7 @@ shows which module holds each. - **Timeline breaks.** On a seek or pause, call `discontinuity()`, then keep timestamps moving forward and resume video on a keyframe. - **Decoded frames hold decoder buffers.** Drop each frame from a `moq.video.Consumer` promptly, or the decoder stalls. `resize` is best effort, so read each frame's `width()` and `height()`. - **Closing.** `session.shutdown()` requests a drain immediately; `await session.closed()` waits up to one second and reports its result. `cancel(code)` closes at once. Finish or abort live tracks first. Application delivery requires explicitly selecting `moq-lite-07-wip` on both peers; older Lite versions (including default lite-06) and IETF wait only for transport ACKs. -- **Ending cleanly.** When the input ends, finish every track, then call `finish()` on its `moq.media.CatalogProducer` to finish the catalog, which ends the content. The broadcast stays announced and keeps serving the finished tracks, so a subscriber still catching up can drain them, until `broadcast.close()` takes it offline. How long to keep it up is yours to decide. +- **Ending cleanly.** When the input ends, finish every track, then call `finish()` on its `moq.media.CatalogProducer` to finish the catalog, which ends the content. The broadcast stays announced and keeps serving the finished tracks, so a subscriber still catching up can drain them, until `broadcast.close()` takes it offline. How long to keep it up is yours to decide. A track still open at the close, or released without a finish, ends with `Unroutable`, which readers take as the publisher leaving. - **Stats.** `session.stats()` reports `rtt`, `estimated_send_rate_bps`, and the byte and packet counters (`bytes_sent`, `bytes_received`, `bytes_lost`, `packets_sent`, `packets_received`, `packets_lost`). A field is `None` when the transport does not report it, which is not the same as zero. `version` names the negotiated protocol. Connect returns a connecting session; read its coherent `state()` snapshot and wait with that snapshot's revision to observe first-dial failures and reconnects. Each watcher keeps its own revision. Hold the owned `Bandwidth.Consumer` returned by `session.recv_bandwidth()` to receive bandwidth estimates. `rate()` reads the current estimate in bits per second and `rate_changed()` waits for a new one. Release the consumer or call `cancel()` when finished; releasing the last consumer stops PROBE; the session and stats snapshots do not request it. Client consumers survive reconnects and report no estimate while disconnected. - **Authorization.** On `moq-lite-07-wip` (opt in with `versions=`) and on moq-transport draft-17+ with the [MoQ Auth extension](/draft/moq-auth), `session.auth().grant()` is the `Grant` the relay sent back, `None` until it answers and while a client session reconnects; `await auth.grant_changed()` waits for the next one. `await auth.add(token)` presents a fresh token without reconnecting, returns an `AuthToken` whose grant joins the union until `cancel()`, and is presented again on every reconnect. A refusal raises with the relay's code. - **Errors.** Read diagnostic payloads from `reason` and structured protocol failures from `details`. diff --git a/doc/lib/swift/index.md b/doc/lib/swift/index.md index efdbca0d66e..0aeaad6a9df 100644 --- a/doc/lib/swift/index.md +++ b/doc/lib/swift/index.md @@ -102,7 +102,7 @@ shows which module holds each. - **Timeline breaks.** On a seek or pause, call `discontinuity()`, then keep timestamps moving forward and resume video on a keyframe. - **Decoded frames hold decoder buffers.** Release each frame from a `Video.Consumer` promptly, or the decoder stalls. `resize` is best effort, so read each frame's `width()` and `height()`. - **Closing.** `session.shutdown()` requests a drain immediately; `try await session.closed()` waits up to one second and reports its result. `session.cancel(code:)` closes at once. Finish or abort live tracks first. Application delivery requires explicitly selecting `moq-lite-07-wip` on both peers; older Lite versions (including default lite-06) and IETF wait only for transport ACKs. -- **Ending cleanly.** When the input ends, finish every track, then call `try finish()` on its `Media.CatalogProducer` to finish the catalog, which ends the content. The broadcast stays announced and keeps serving the finished tracks, so a subscriber still catching up can drain them, until `broadcast.close()` takes it offline. How long to keep it up is yours to decide. +- **Ending cleanly.** When the input ends, finish every track, then call `try finish()` on its `Media.CatalogProducer` to finish the catalog, which ends the content. The broadcast stays announced and keeps serving the finished tracks, so a subscriber still catching up can drain them, until `broadcast.close()` takes it offline. How long to keep it up is yours to decide. A track still open at the close, or released without a finish, ends with `Unroutable`, which readers take as the publisher leaving. - **Stats.** `session.stats()` reports `rtt`, `estimatedSendRateBps`, and the byte and packet counters (`bytesSent`, `bytesReceived`, `bytesLost`, `packetsSent`, `packetsReceived`, `packetsLost`). A field is `nil` when the transport does not report it, which is not the same as zero. `version` names the negotiated protocol. Connect returns a connecting session; read its coherent `state()` snapshot and wait with that snapshot's revision to observe first-dial failures and reconnects. Each watcher keeps its own revision. Hold the owned `Bandwidth.Consumer` returned by `session.recvBandwidth()` to receive bandwidth estimates. `rate()` reads the current estimate in bits per second and `rateChanged()` waits for a new one. Release the consumer or call `cancel()` when finished; releasing the last consumer stops PROBE; the session and stats snapshots do not request it. Client consumers survive reconnects and report no estimate while disconnected. - **Authorization.** On `moq-lite-07-wip` (opt in with `Client(versions:)`) and on moq-transport draft-17+ with the [MoQ Auth extension](/draft/moq-auth), `session.auth().grant()` is the `Grant` the relay sent back, `nil` until it answers and while a client session reconnects; `try await auth.grantChanged()` waits for the next one. `try await auth.add(token:)` presents a fresh token without reconnecting, returns an `AuthToken` whose grant joins the union until `cancel()`, and is presented again on every reconnect. A refusal throws with the relay's code. - **Errors.** Diagnostic cases use the `reason:` argument label; match their associated value to inspect it. Protocol errors carry `details`. diff --git a/go/wrapper/publish.go b/go/wrapper/publish.go index 6cdd2457d5e..d03ebd52516 100644 --- a/go/wrapper/publish.go +++ b/go/wrapper/publish.go @@ -75,7 +75,9 @@ func (b *BroadcastProducer) Consume() (*BroadcastConsumer, error) { } // Close ends the broadcast for good: it retracts and serves no new tracks. -// Tracks already subscribed carry on to their own end. Closing again is a no-op. +// Tracks still open on it end with Unroutable, which readers take as the +// publisher leaving, and their later calls fail with ErrClosed; finish a track +// first to end it cleanly. Closing again is a no-op. func (b *BroadcastProducer) Close() error { return b.inner.Close() } diff --git a/py/moq-rs/moq/publish.py b/py/moq-rs/moq/publish.py index 97142f84e67..38fff546fcc 100644 --- a/py/moq-rs/moq/publish.py +++ b/py/moq-rs/moq/publish.py @@ -359,6 +359,8 @@ def consume(self) -> BroadcastConsumer: def close(self) -> None: """End the broadcast for good: retract it and serve no new tracks. - Tracks already subscribed carry on to their own end. Closing again is a no-op. + Tracks still open on it end with ``Unroutable``, which readers take as the publisher + leaving, and their later calls raise ``moq.Error.Closed``; finish a track first to end it + cleanly. Closing again is a no-op. """ self._inner.close() diff --git a/quest/m1/broadcast-lifecycle/README.md b/quest/m1/broadcast-lifecycle/README.md index dd7df0a96fa..dd2ead9f311 100644 --- a/quest/m1/broadcast-lifecycle/README.md +++ b/quest/m1/broadcast-lifecycle/README.md @@ -46,7 +46,6 @@ timer on main at 1582fdc07d: ## Required -- [Closing a binding broadcast ends its media](/quest/m1/broadcast-lifecycle/ffi-close.md) - moq-ffi and moq-c media handles end with a cause on close or drop, never a cause-less drop - [js/net close FIN](/quest/m1/broadcast-lifecycle/js-close-fin.md) - js/net finishes tracks whose FIN arrives with a graceful close ## Related diff --git a/quest/m1/broadcast-lifecycle/ffi-close.md b/quest/m1/broadcast-lifecycle/ffi-close.md deleted file mode 100644 index 06ce38923be..00000000000 --- a/quest/m1/broadcast-lifecycle/ffi-close.md +++ /dev/null @@ -1,32 +0,0 @@ -# [M] Closing a binding broadcast ends its media - -## Goal - -No moq-ffi or moq-c media track ends without a cause. Closing a broadcast -ends every media handle still open with `Unroutable`, and later calls on -those handles return `Closed`. A handle freed or dropped without `finish()` -aborts with `Unroutable` too. `finish()` stays the only clean end. Readers, -including the TS export's follower, see the publisher leave instead of a -cause-less drop. - -## Plan - -Decided with the maintainer on 2026-10-24, breaking the API: structured -ownership, no new `abort()` on media handles. - -- Found while fixing [moq-dev/moq#6656](https://github.com/moq-dev/moq/pull/6656): - `MoqBroadcastProducer::close()` and `moq_publish_close()` leave media - tracks open, and the handles have no abort. An app that drops them while - its session is up leaves readers seeing the tracks dropped unfinished, which - `left()` does not count as the publisher leaving. -- Media handles are children of their broadcast. Close ends the children - first, then the broadcast. A drop without `finish()` is an abort, never a - cause-less drop, so a garbage-collected binding cannot leak one. -- Breaking: a call on a media handle after its broadcast closes returns - `Closed`, and a dropped handle now aborts its track. Migrate each binding - wrapper (py, go, swift, kt, dart, cpp), the moq-ffi and moq-c doc - comments that say subscribed tracks carry on, and `doc/lib/*`; the PR's - Impact says how. -- Test in moq-ffi and moq-c with a paused clock: close with open handles, - drop without finish, and finish then close, each read by a subscriber that - checks the cause. diff --git a/rs/moq-c/src/api.rs b/rs/moq-c/src/api.rs index c5b2179a5b7..1f4b0b36f88 100644 --- a/rs/moq-c/src/api.rs +++ b/rs/moq-c/src/api.rs @@ -2200,16 +2200,21 @@ pub extern "C" fn moq_publish_unannounce(broadcast: u32) -> i32 { /// End a broadcast for good and release its handle. /// -/// The origin retracts the path immediately and serves no new tracks; tracks already -/// subscribed carry on to their own end. The handle is invalid afterwards, so closing -/// it again fails like any unknown handle. +/// The origin retracts the path immediately and serves no new tracks. Every media, raw track, +/// container, encoder, JSON, and binary handle still open on the broadcast ends with `UNROUTABLE`, +/// the way a reader sees a publisher leave, and is released; finish a track first to end it +/// cleanly. The +/// handles are invalid afterwards, so closing it again fails like any unknown handle, and so +/// does any later call on a released child. /// /// Returns a zero on success, or a negative code on failure. #[unsafe(no_mangle)] pub extern "C" fn moq_publish_close(broadcast: u32) -> i32 { ffi::enter(move || { let broadcast = ffi::parse_id(broadcast)?; - State::lock().publish.close(broadcast) + // Bound first, so the global lock is released before the encoders end. + let closing = State::lock().publish_close(broadcast)?; + closing.run() }) } @@ -2355,7 +2360,7 @@ pub extern "C" fn moq_publish_media_seek(media: u32, sequence: u64) -> i32 { /// Finish a media track, flushing any buffered frames. No more frames can be written. /// /// The broadcast keeps the finished track, catalog rendition included, until it closes, so a -/// late subscriber can still drain it. +/// late subscriber can still drain it. A failed finish ends the track with its error instead. /// /// Returns a zero on success, or a negative code on failure. #[unsafe(no_mangle)] @@ -2482,7 +2487,7 @@ pub extern "C" fn moq_publish_container_seek(container: u32, sequence: u64) -> i /// Finish every track the container publishes and release the handle. /// /// The broadcast keeps the finished tracks, catalog renditions included, until it closes, so a -/// late subscriber can still drain them. +/// late subscriber can still drain them. A failed finish ends the tracks with its error instead. /// /// Returns zero on success, or a negative code on failure. #[unsafe(no_mangle)] @@ -3043,7 +3048,9 @@ pub unsafe extern "C" fn moq_track_request_dynamic( /// /// Consumes the request handle. `info` is as in [moq_publish_track]: NULL for the /// microsecond default. Returns a non-zero track handle usable with every -/// `moq_publish_track_*` function, or a negative code on failure. +/// `moq_publish_track_*` function, or a negative code on failure. Once the request's +/// broadcast closed, the track ends with `UNROUTABLE` and the call fails with +/// `MOQ_ERROR_BROADCAST_NOT_FOUND`. /// /// # Safety /// - `info` must be NULL or a valid pointer to a [moq_track_info] struct. diff --git a/rs/moq-c/src/audio.rs b/rs/moq-c/src/audio.rs index d8cb10bc28f..fdbe3d20765 100644 --- a/rs/moq-c/src/audio.rs +++ b/rs/moq-c/src/audio.rs @@ -146,11 +146,19 @@ pub struct moq_audio_frame { pub(crate) struct AudioEncoder { producer: moq_audio::encode::Producer, reservation: Option>, - finished: crate::publish::Finished, + children: crate::publish::Children, } type AudioProducer = Shared; +impl crate::publish::Abort for AudioProducer { + fn abort(self, err: moq_net::Error) { + if let Some(encoder) = self.take() { + encoder.producer.abort(err); + } + } +} + #[derive(Default)] /// The audio encoder, decoder, and decoded frame handle tables. pub struct Audio { @@ -170,11 +178,14 @@ impl Audio { ) -> Result { let producer = moq_audio::encode::Producer::new(target.producer, target.catalog.clone(), input, &options)?; let reservation = reserve.then(|| Arc::new(options.bandwidth.reserve(&producer.demand(), producer.bitrate()))); - self.producers.insert(Shared::new(AudioEncoder { + let encoder = Shared::new(AudioEncoder { producer, reservation, - finished: target.finished.clone(), - })) + children: target.children.clone(), + }); + target + .children + .insert(&mut self.producers, encoder, crate::publish::Child::Audio) } pub(crate) fn reservation(&self, id: Id) -> Result>, Error> { @@ -450,17 +461,29 @@ pub unsafe extern "C" fn moq_encode_audio_frame(producer: u32, frame: *const moq #[unsafe(no_mangle)] pub extern "C" fn moq_encode_audio_finish(producer: u32) -> i32 { ffi::enter(move || { - let producer = ffi::parse_id(producer)?; - // The id is dropped first, so nothing new queues behind the flush; whatever - // is mid-encode still finishes before this takes the producer. - let producer = State::lock().audio.remove(producer)?; - let AudioEncoder { - mut producer, finished, .. - } = producer.take().ok_or(Error::MediaNotFound)?; - producer.finish()?; - // The broadcast keeps the finished track, so its finish still holds it. - finished.keep(producer); - Ok(()) + let id = ffi::parse_id(producer)?; + // The id stays until the flush ends, so a racing moq_publish_close waits it out + // instead of missing a track that still publishes. A call queued behind the flush + // finds the encoder gone. + let shared = State::lock().audio.producer(id)?; + let finished: Result<(), Error> = (|| { + let mut guard = shared.lock(); + let AudioEncoder { + mut producer, children, .. + } = guard.take().ok_or(Error::MediaNotFound)?; + if let Err(err) = producer.finish() { + // Readers see the failure rather than `Dropped`, as a failed video drain does. + children.forget(crate::publish::Child::Audio(id)); + producer.abort(moq_net::Error::Transport(err.to_string())); + return Err(err.into()); + } + // The broadcast keeps the finished track, so its finish still holds it. + children.keep(crate::publish::Child::Audio(id), producer); + Ok(()) + })(); + // A racing close may have released the id already. + let _ = State::lock().audio.remove(id); + finished }) } diff --git a/rs/moq-c/src/publish.rs b/rs/moq-c/src/publish.rs index 90ac51a32e5..a5ed46a82d6 100644 --- a/rs/moq-c/src/publish.rs +++ b/rs/moq-c/src/publish.rs @@ -1,4 +1,4 @@ -use std::collections::BTreeMap; +use std::collections::{BTreeMap, HashSet}; use std::sync::Arc; use moq_mux::catalog::hang::Extra; @@ -19,8 +19,8 @@ struct DemandEntry { /// broadcast it was made on so media can be published onto it under that broadcast's catalog. struct TrackRequest { broadcast: Id, - // Carried, not looked up, so a raw accept still works after the broadcast handle is released. - finished: Finished, + // Carried, not looked up, so a raw accept learns its broadcast closed even after the id is gone. + children: Children, request: moq_net::track::Request, } @@ -43,7 +43,7 @@ struct GroupRequest { /// What a request handler serves. enum Dynamic { /// Track requests on a broadcast, remembered so media can be published onto them. - Broadcast(moq_net::broadcast::Dynamic, Id, Finished), + Broadcast(moq_net::broadcast::Dynamic, Id, Children), /// Group requests (fetches of uncached groups) on a track. Track(moq_net::track::Dynamic), } @@ -60,51 +60,115 @@ struct Broadcast { catalog: moq_mux::catalog::Producer, video: BTreeMap, audio: BTreeMap, - finished: Finished, + children: Children, /// Set by [`Publish::catalog_finish`]: the catalog is final, so a close leaves it alone. ended: bool, } -impl Drop for Broadcast { - fn drop(&mut self) { - self.finished.release(); - } +/// A handle publishing tracks on a broadcast, by the table that holds it. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +pub(crate) enum Child { + Media(Id), + Container(Id), + Track(Id), + Audio(Id), + Video(Id), + JsonSnapshot(Id), + JsonStream(Id), + BinarySnapshot(Id), + BinaryStream(Id), +} + +/// A child's handle, which ends what it still publishes when its broadcast closes. +pub(crate) trait Abort { + /// End the handle's tracks with `err`. A track that already finished ends cleanly instead. + fn abort(self, err: moq_net::Error); } -/// The finished producers a broadcast keeps, or `None` once it ended. -type Keeping = Option>>; +/// What a broadcast holds of its children. +enum Family { + Open { + /// Handles that may still publish, ended on close. + open: HashSet, + /// Finished producers, kept so a late subscriber still finds their tracks. + kept: Vec>, + }, + /// The broadcast closed: its children ended and new ones are refused. + Closed, +} -/// Finished producers a broadcast keeps until it ends. +/// The track-publishing handles of one broadcast. /// -/// A finished track stops being served once its last producer drops. A `*_finish` call frees -/// its handle, so the broadcast keeps the finished producer here instead, and a subscriber -/// arriving after the end still finds the track until [`Publish::close`]. +/// Closing the broadcast ends every child still open with `Unroutable`, the way a reader sees a +/// publisher leave, and frees its handle. A finished track stops being served once its last +/// producer drops, and a `*_finish` call frees its handle, so the broadcast keeps the finished +/// producer here instead: a subscriber arriving after the end still finds the track until +/// [`State::publish_close`]. #[derive(Clone)] -pub struct Finished(std::sync::Arc>); +pub struct Children(Arc>); -impl Default for Finished { +impl Default for Children { fn default() -> Self { - Self(std::sync::Arc::new(std::sync::Mutex::new(Some(Vec::new())))) + Self(Arc::new(std::sync::Mutex::new(Family::Open { + open: HashSet::new(), + kept: Vec::new(), + }))) } } -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 Send + 'static) { +impl Children { + fn lock(&self) -> std::sync::MutexGuard<'_, Family> { + self.0.lock().unwrap_or_else(std::sync::PoisonError::into_inner) + } + + /// Insert a new child into `slab`, so the broadcast's close ends it. Once the broadcast + /// closed, the child is aborted and refused. + pub(crate) fn insert( + &self, + slab: &mut NonZeroSlab, + value: T, + child: fn(Id) -> Child, + ) -> Result { + let id = slab.insert(value)?; + match &mut *self.lock() { + Family::Open { open, .. } => { + open.insert(child(id)); + return Ok(id); + } + Family::Closed => {} + } + if let Some(value) = slab.remove(id) { + value.abort(moq_net::Error::Unroutable); + } + Err(Error::BroadcastNotFound) + } + + /// Keep `child`'s finished producer until the broadcast ends, or drop it now if it already has. + pub(crate) fn keep(&self, child: Child, producer: impl Send + 'static) { let producer: Box = Box::new(producer); - let mut kept = self.0.lock().unwrap_or_else(std::sync::PoisonError::into_inner); - if let Some(kept) = kept.as_mut() { + if let Family::Open { open, kept } = &mut *self.lock() { + open.remove(&child); kept.push(producer); } } - /// Stop keeping, handing back what was kept so the caller decides when it drops. - fn release(&self) -> Vec> { - self.0 - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .take() - .unwrap_or_default() + /// Stop tracking `child`, which ended on its own. + pub(crate) fn forget(&self, child: Child) { + if let Family::Open { open, .. } = &mut *self.lock() { + open.remove(&child); + } + } + + /// Refuse new children, handing back the open ones and letting go of the finished ones. + fn close(&self) -> HashSet { + let family = std::mem::replace(&mut *self.lock(), Family::Closed); + match family { + Family::Open { open, kept } => { + drop(kept); + open + } + Family::Closed => HashSet::new(), + } } } @@ -112,19 +176,73 @@ impl Finished { pub struct Target<'a> { pub producer: &'a mut moq_net::broadcast::Producer, pub catalog: &'a mut moq_mux::catalog::Producer, - /// Where the broadcast keeps the producer once it finishes. - pub finished: &'a Finished, + /// The broadcast's children, which the producer joins. + pub children: &'a Children, } -/// A producer handle, with where its broadcast keeps it once finished. +/// A producer handle, with the broadcast that keeps it once finished. struct Kept { inner: T, - finished: Finished, + children: Children, } impl Kept { - fn keep(self) { - self.finished.keep(self.inner); + fn keep(self, child: Child) { + self.children.keep(child, self.inner); + } +} + +impl Abort for Kept> { + fn abort(self, err: moq_net::Error) { + self.inner.abort(err); + } +} + +impl Abort for Kept { + fn abort(self, err: moq_net::Error) { + // A track with a declared end ends there, as when its last producer drops. + if self.inner.end().is_none() { + let _ = self.inner.abort(err); + } + } +} + +impl Abort for Container { + fn abort(self, err: moq_net::Error) { + self.inner.abort(err); + } +} + +impl Abort for JsonSnapshot { + fn abort(self, err: moq_net::Error) { + let _ = self.data.producer.abort(err); + } +} + +impl Abort for Data> { + fn abort(self, err: moq_net::Error) { + let _ = self.producer.abort(err); + } +} + +impl Abort for Data { + fn abort(self, err: moq_net::Error) { + let _ = self.producer.abort(err); + } +} + +impl Abort for Data { + fn abort(self, err: moq_net::Error) { + let _ = self.producer.abort(err); + } +} + +/// The cause a failed import ends its tracks with, so a reader sees the publisher fail rather than +/// `Dropped`. Only a moq-net error has a wire code; any other reaches readers as `Internal`. +fn cause(err: &moq_mux::Error) -> moq_net::Error { + match err { + moq_mux::Error::Moq(err) => err.clone(), + _ => moq_net::StreamError::Internal.into(), } } @@ -147,6 +265,7 @@ impl std::ops::DerefMut for Kept { struct Data { producer: T, clock: moq_mux::Clock, + children: Children, } /// A JSON snapshot producer, kept beside the signal that stops its refresh task. @@ -160,7 +279,7 @@ struct JsonSnapshot { struct Container { inner: import::Container, driver: crate::ffi::Background, - finished: Finished, + children: Children, } #[derive(Default)] @@ -240,7 +359,7 @@ impl Publish { catalog, video: BTreeMap::new(), audio: BTreeMap::new(), - finished: Finished::default(), + children: Children::default(), ended: false, })?; Ok(id) @@ -322,13 +441,12 @@ impl Publish { Ok(Target { producer: &mut broadcast.producer, catalog: &mut broadcast.catalog, - finished: &broadcast.finished, + children: &broadcast.children, }) } - /// End the broadcast for good and release it, finalizing the catalog stream. - pub fn close(&mut self, broadcast: Id) -> Result<(), Error> { - let mut broadcast = self.broadcasts.remove(broadcast).ok_or(Error::BroadcastNotFound)?; + /// End the broadcast for good, finalizing the catalog stream. Its children already ended. + fn close(mut broadcast: Broadcast) -> Result<(), Error> { if broadcast.ended { broadcast.producer.close(); return broadcast.driver.check(); @@ -355,7 +473,7 @@ impl Publish { } /// End the broadcast's content by finishing its catalog, with its entries still listed, so - /// a late subscriber finds the finished tracks until [`Self::close`]. + /// a late subscriber finds the finished tracks until [`State::publish_close`]. pub fn catalog_finish(&mut self, broadcast: Id) -> Result<(), Error> { let broadcast = self.broadcast_mut(broadcast)?; let result = broadcast.catalog.finish(); @@ -364,16 +482,25 @@ impl Publish { Ok(result?) } - /// Where a broadcast keeps the producers finished on it. - fn finished(&self, broadcast: Id) -> Result { - Ok(self.broadcast(broadcast)?.finished.clone()) + /// The children of a broadcast. + fn children(&self, broadcast: Id) -> Result { + Ok(self.broadcast(broadcast)?.children.clone()) + } + + /// Insert a media importer as a child of the broadcast `children` belongs to. + fn insert_media(&mut self, track: import::Track, children: Children) -> Result { + let media = Kept { + inner: Box::new(track), + children: children.clone(), + }; + children.insert(&mut self.media, media, Child::Media) } pub fn audio(&mut self, broadcast: Id, init: import::AudioInit) -> Result { let Broadcast { producer: broadcast, catalog, - finished, + children, .. } = self.broadcast(broadcast)?; let broadcast = broadcast.clone(); @@ -381,19 +508,15 @@ impl Publish { let request = broadcast.reserve_track(name)?; let track = import::Track::audio(request, catalog.reserve(), init)?; - let finished = finished.clone(); - let id = self.media.insert(Kept { - inner: Box::new(track), - finished, - })?; - Ok(id) + let children = children.clone(); + self.insert_media(track, children) } pub fn video(&mut self, broadcast: Id, init: import::VideoInit) -> Result { let Broadcast { producer: broadcast, catalog, - finished, + children, .. } = self.broadcast(broadcast)?; let broadcast = broadcast.clone(); @@ -401,12 +524,8 @@ impl Publish { let request = broadcast.reserve_track(name)?; let track = import::Track::video(request, catalog.reserve(), init)?; - let finished = finished.clone(); - let id = self.media.insert(Kept { - inner: Box::new(track), - finished, - })?; - Ok(id) + let children = children.clone(); + self.insert_media(track, children) } pub fn container(&mut self, broadcast: Id, init: import::ContainerInit) -> Result { @@ -414,17 +533,17 @@ impl Publish { producer: broadcast, catalog, driver, - finished, + children, .. } = self.broadcast(broadcast)?; let container = import::Container::new(broadcast.clone(), catalog.reserve(), &init)?; + let children = children.clone(); let container = Container { inner: container, driver: driver.clone(), - finished: finished.clone(), + children: children.clone(), }; - let id = self.containers.insert(container)?; - Ok(id) + children.insert(&mut self.containers, container, Child::Container) } pub fn media_frame(&mut self, media: Id, data: &[u8], timestamp: hang::container::Timestamp) -> Result<(), Error> { @@ -465,8 +584,13 @@ impl Publish { pub fn media_finish(&mut self, media: Id) -> Result<(), Error> { let mut track = self.media.remove(media).ok_or(Error::MediaNotFound)?; - track.finish()?; - track.keep(); + if let Err(err) = track.finish() { + track.children.forget(Child::Media(media)); + let cause = cause(&err); + track.abort(cause); + return Err(err.into()); + } + track.keep(Child::Media(media)); Ok(()) } @@ -500,11 +624,17 @@ impl Publish { } pub fn container_finish(&mut self, container: Id) -> Result<(), Error> { - let mut container = self.containers.remove(container).ok_or(Error::MediaNotFound)?; - let result = container.inner.finish(); + let id = container; + let mut container = self.containers.remove(id).ok_or(Error::MediaNotFound)?; + if let Err(err) = container.inner.finish() { + container.children.forget(Child::Container(id)); + let driver = container.driver.clone(); + container.abort(cause(&err)); + driver.check()?; + return Err(err.into()); + } container.driver.check()?; - result?; - container.finished.keep(container.inner); + container.children.keep(Child::Container(id), container.inner); Ok(()) } @@ -699,8 +829,8 @@ impl Publish { /// Without a live handler an unknown track name is refused, as before. pub fn dynamic(&mut self, broadcast: Id, on_request: OnStatus) -> Result { let dynamic = self.producer(broadcast)?.dynamic(); - let Broadcast { driver, finished, .. } = self.broadcast(broadcast)?; - let dynamic = Dynamic::Broadcast(dynamic, broadcast, finished.clone()); + let Broadcast { driver, children, .. } = self.broadcast(broadcast)?; + let dynamic = Dynamic::Broadcast(dynamic, broadcast, children.clone()); self.spawn_dynamic(dynamic, on_request, driver.clone()) } @@ -754,11 +884,11 @@ impl Publish { res = kio::wait(|waiter| { let _ = fills.poll(waiter); match &mut dynamic { - Dynamic::Broadcast(dynamic, broadcast, finished) => { + Dynamic::Broadcast(dynamic, broadcast, children) => { dynamic.poll_requested_track(waiter).map_ok(|request| { Request::Track(Box::new(TrackRequest { broadcast: *broadcast, - finished: finished.clone(), + children: children.clone(), request, })) }) @@ -836,41 +966,36 @@ impl Publish { /// Accept a track request as a raw track, returning a track handle like [`Self::track`]. pub fn track_request_accept(&mut self, request: Id, info: moq_net::track::Info) -> Result { - let TrackRequest { finished, request, .. } = self.track_request.remove(request).ok_or(Error::NotFound)?; - self.tracks.insert(Kept { + let TrackRequest { children, request, .. } = self.track_request.remove(request).ok_or(Error::NotFound)?; + let track = Kept { inner: request.accept(info), - finished, - }) + children: children.clone(), + }; + children.insert(&mut self.tracks, track, Child::Track) } /// Accept a track request as an audio track, returning a media handle like [`Self::audio`]. pub fn track_request_audio(&mut self, request: Id, init: import::AudioInit) -> Result { let TrackRequest { broadcast, - finished, + children, request, } = self.track_request.remove(request).ok_or(Error::NotFound)?; let catalog = self.catalog(broadcast)?; let track = import::Track::audio(request, catalog.reserve(), init)?; - self.media.insert(Kept { - inner: Box::new(track), - finished, - }) + self.insert_media(track, children) } /// Accept a track request as a video track, returning a media handle like [`Self::video`]. pub fn track_request_video(&mut self, request: Id, init: import::VideoInit) -> Result { let TrackRequest { broadcast, - finished, + children, request, } = self.track_request.remove(request).ok_or(Error::NotFound)?; let catalog = self.catalog(broadcast)?; let track = import::Track::video(request, catalog.reserve(), init)?; - self.media.insert(Kept { - inner: Box::new(track), - finished, - }) + self.insert_media(track, children) } /// Reject a track request, failing every subscriber waiting on it with `error_code`. @@ -927,10 +1052,14 @@ impl Publish { /// for non-media tracks. Pair it with [`Self::video_config`] / [`Self::audio_config`] /// if you want to describe the track in the catalog as well. pub fn track(&mut self, broadcast: Id, name: &str, info: Option) -> Result { - let finished = self.finished(broadcast)?; + let children = self.children(broadcast)?; let broadcast = self.producer(broadcast)?; let track = broadcast.create_track(name, info)?; - self.tracks.insert(Kept { inner: track, finished }) + let track = Kept { + inner: track, + children: children.clone(), + }; + children.insert(&mut self.tracks, track, Child::Track) } /// Append a new group to a raw track, returning a group producer. @@ -970,11 +1099,12 @@ impl Publish { /// [`Self::track_finish_at`] declares the boundary ahead of time, so this keeps that /// boundary and only releases the handle; the broadcast keeps the track. pub fn track_finish(&mut self, track: Id) -> Result<(), Error> { - let track = self.tracks.remove(track).ok_or(Error::TrackNotFound)?; + let id = track; + let track = self.tracks.remove(id).ok_or(Error::TrackNotFound)?; if track.end().is_none() { track.finish()?; } - track.keep(); + track.keep(Child::Track(id)); Ok(()) } @@ -987,7 +1117,9 @@ impl Publish { /// Abort a raw track with an application error code. pub fn track_abort(&mut self, track: Id, error_code: u16) -> Result<(), Error> { - let track = self.tracks.remove(track).ok_or(Error::TrackNotFound)?; + let id = track; + let track = self.tracks.remove(id).ok_or(Error::TrackNotFound)?; + track.children.forget(Child::Track(id)); track.inner.abort(moq_net::Error::App(error_code))?; Ok(()) } @@ -1011,6 +1143,7 @@ impl Publish { Ok(Data { producer, clock: broadcast.catalog.clock(), + children: broadcast.children.clone(), }) } @@ -1036,7 +1169,9 @@ impl Publish { })?; let alive = kio::Producer::new(()); let closed = alive.consume(); - let id = self.json_snapshot.insert(JsonSnapshot { data, _alive: alive })?; + let children = data.children.clone(); + let snapshot = JsonSnapshot { data, _alive: alive }; + let id = children.insert(&mut self.json_snapshot, snapshot, Child::JsonSnapshot)?; crate::ffi::Background::new(async move { Self::json_snapshot_refresh(id, closed).await; Ok(()) @@ -1079,7 +1214,9 @@ impl Publish { /// Finish a JSON snapshot track and retire its catalog entry. No more values can be published. pub fn json_snapshot_finish(&mut self, json: Id) -> Result<(), Error> { - let json = self.json_snapshot.remove(json).ok_or(Error::TrackNotFound)?; + let id = json; + let json = self.json_snapshot.remove(id).ok_or(Error::TrackNotFound)?; + json.data.children.forget(Child::JsonSnapshot(id)); json.data.producer.finish()?; Ok(()) } @@ -1092,7 +1229,8 @@ impl Publish { let producer = self.data_track(broadcast, name, None, |catalog, track| { catalog.json_stream(track, config) })?; - self.json_stream.insert(producer) + let children = producer.children.clone(); + children.insert(&mut self.json_stream, producer, Child::JsonStream) } /// Append one record to a JSON stream track. @@ -1107,6 +1245,7 @@ impl Publish { /// Finish a JSON stream track and retire its catalog entry. No more records can be appended. pub fn json_stream_finish(&mut self, stream: Id) -> Result<(), Error> { let producer = self.json_stream.remove(stream).ok_or(Error::TrackNotFound)?; + producer.children.forget(Child::JsonStream(stream)); producer.producer.finish()?; Ok(()) } @@ -1117,7 +1256,8 @@ impl Publish { let producer = self.data_track(broadcast, name, None, |catalog, track| { catalog.binary_snapshot(track, config) })?; - self.binary_snapshot.insert(producer) + let children = producer.children.clone(); + children.insert(&mut self.binary_snapshot, producer, Child::BinarySnapshot) } /// Publish a new payload to a binary snapshot track, superseding the last. @@ -1132,6 +1272,7 @@ impl Publish { /// Finish a binary snapshot track and retire its catalog entry. pub fn binary_snapshot_finish(&mut self, binary: Id) -> Result<(), Error> { let producer = self.binary_snapshot.remove(binary).ok_or(Error::TrackNotFound)?; + producer.children.forget(Child::BinarySnapshot(binary)); producer.producer.finish()?; Ok(()) } @@ -1142,7 +1283,8 @@ impl Publish { let producer = self.data_track(broadcast, name, None, |catalog, track| { catalog.binary_stream(track, config) })?; - self.binary_stream.insert(producer) + let children = producer.children.clone(); + children.insert(&mut self.binary_stream, producer, Child::BinaryStream) } /// Append one payload to a binary stream track. @@ -1157,6 +1299,7 @@ impl Publish { /// Finish a binary stream track and retire its catalog entry. pub fn binary_stream_finish(&mut self, stream: Id) -> Result<(), Error> { let producer = self.binary_stream.remove(stream).ok_or(Error::TrackNotFound)?; + producer.children.forget(Child::BinaryStream(stream)); producer.producer.finish()?; Ok(()) } @@ -1183,6 +1326,218 @@ impl Publish { } } +impl State { + /// Release a broadcast and every child still open on it, returning what is left to end it. + /// + /// The children end first, with `Unroutable` as a reader sees a publisher leave, so none + /// outlives the broadcast it publishes on. + pub fn publish_close(&mut self, broadcast: Id) -> Result { + let broadcast = self + .publish + .broadcasts + .remove(broadcast) + .ok_or(Error::BroadcastNotFound)?; + let err = moq_net::Error::Unroutable; + let mut encoders: Vec> = Vec::new(); + for child in broadcast.children.close() { + match child { + Child::Media(id) => self.publish.media.remove(id).map(|media| media.abort(err.clone())), + Child::Container(id) => self.publish.containers.remove(id).map(|media| media.abort(err.clone())), + Child::Track(id) => self.publish.tracks.remove(id).map(|track| track.abort(err.clone())), + Child::Audio(id) => self + .audio + .remove(id) + .ok() + .map(|audio| encoders.push(Box::new(|err| audio.abort(err)))), + Child::Video(id) => self + .video + .remove(id) + .ok() + .map(|video| encoders.push(Box::new(|err| video.abort(err)))), + Child::JsonSnapshot(id) => self + .publish + .json_snapshot + .remove(id) + .map(|json| json.abort(err.clone())), + Child::JsonStream(id) => self.publish.json_stream.remove(id).map(|json| json.abort(err.clone())), + Child::BinarySnapshot(id) => self + .publish + .binary_snapshot + .remove(id) + .map(|data| data.abort(err.clone())), + Child::BinaryStream(id) => self + .publish + .binary_stream + .remove(id) + .map(|data| data.abort(err.clone())), + }; + } + Ok(Closing { encoders, broadcast }) + } +} + +/// A broadcast released by [`State::publish_close`], ended by [`Closing::run`] once the global +/// lock is released. +/// +/// An encoder holds its own lock across a codec call, a drain included, so ending one under the +/// global lock would stall every unrelated call behind that codec. +#[must_use = "the broadcast ends only once run"] +pub struct Closing { + encoders: Vec>, + broadcast: Broadcast, +} + +impl Closing { + /// End the released encoders, waiting out any call in flight on them, then the broadcast. + pub fn run(self) -> Result<(), Error> { + for abort in self.encoders { + abort(moq_net::Error::Unroutable); + } + Publish::close(self.broadcast) + } +} + +#[cfg(test)] +mod children_tests { + use super::*; + + const TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); + + /// A standalone broadcast with one raw track and one media importer, each read by a subscriber. + async fn children(state: &mut State) -> (Id, Id, Id, [moq_net::track::Subscriber; 2]) { + let broadcast = state + .publish + .create((moq_net::broadcast::Info::new().produce(), Default::default())) + .unwrap(); + let consumer = state.publish.broadcast(broadcast).unwrap().producer.consume(); + let track = state.publish.track(broadcast, "raw", None).unwrap(); + let mut head = b"OpusHead".to_vec(); + head.extend_from_slice(&[1, 2, 0, 0, 0x80, 0xbb, 0, 0, 0, 0, 0]); + let media = state + .publish + .audio(broadcast, import::AudioInit::new(import::AudioFormat::Opus, head)) + .unwrap(); + let name = state.publish.media_demand(media).unwrap().name().to_string(); + let readers = [ + consumer.track("raw").unwrap().subscribe(None).await.unwrap(), + consumer.track(&name).unwrap().subscribe(None).await.unwrap(), + ]; + (broadcast, track, media, readers) + } + + /// Closing a broadcast ends every child still open on it the way a publisher leaving does, + /// and releases the child handles. + #[tokio::test(start_paused = true)] + async fn closing_a_broadcast_ends_its_open_tracks() { + let mut state = State::new(); + let (broadcast, track, media, readers) = children(&mut state).await; + state.publish_close(broadcast).unwrap().run().unwrap(); + for mut reader in readers { + let read = moq_clock::timeout(TIMEOUT, reader.recv_group()).await; + assert!(matches!( + read.expect("the track carried on"), + Err(moq_net::Error::Unroutable) + )); + } + assert!(matches!(state.publish.track_finish(track), Err(Error::TrackNotFound))); + assert!(matches!(state.publish.media_finish(media), Err(Error::MediaNotFound))); + } + + /// Closing a broadcast ends its data tracks the way it ends media ones, and releases their + /// handles. + #[tokio::test(start_paused = true)] + async fn closing_a_broadcast_ends_its_open_data_tracks() { + let mut state = State::new(); + let broadcast = state + .publish + .create((moq_net::broadcast::Info::new().produce(), Default::default())) + .unwrap(); + let consumer = state.publish.broadcast(broadcast).unwrap().producer.consume(); + let json = moq_mux::json::Config::default; + let binary = moq_mux::binary::Config::default; + let json_snapshot = state + .publish + .json_snapshot(broadcast, "json-snapshot", json(), Default::default()) + .unwrap(); + let json_stream = state.publish.json_stream(broadcast, "json-stream", json()).unwrap(); + let binary_snapshot = state + .publish + .binary_snapshot(broadcast, "binary-snapshot", binary()) + .unwrap(); + let binary_stream = state + .publish + .binary_stream(broadcast, "binary-stream", binary()) + .unwrap(); + let mut readers = Vec::new(); + for name in ["json-snapshot", "json-stream", "binary-snapshot", "binary-stream"] { + readers.push(consumer.track(name).unwrap().subscribe(None).await.unwrap()); + } + + state.publish_close(broadcast).unwrap().run().unwrap(); + for mut reader in readers { + let read = moq_clock::timeout(TIMEOUT, reader.recv_group()).await; + assert!(matches!( + read.expect("the track carried on"), + Err(moq_net::Error::Unroutable) + )); + } + let json = serde_json::Value::Null; + assert!(matches!( + state.publish.json_snapshot_update(json_snapshot, json.clone()), + Err(Error::TrackNotFound) + )); + assert!(matches!( + state.publish.json_stream_append(json_stream, json), + Err(Error::TrackNotFound) + )); + assert!(matches!( + state.publish.binary_snapshot_update(binary_snapshot, b"late"), + Err(Error::TrackNotFound) + )); + assert!(matches!( + state.publish.binary_stream_append(binary_stream, b"late"), + Err(Error::TrackNotFound) + )); + } + + /// A track finished before its broadcast closes ends cleanly. + #[tokio::test(start_paused = true)] + async fn finishing_before_close_ends_tracks_cleanly() { + let mut state = State::new(); + let (broadcast, track, media, readers) = children(&mut state).await; + state.publish.track_finish(track).unwrap(); + state.publish.media_finish(media).unwrap(); + state.publish_close(broadcast).unwrap().run().unwrap(); + for mut reader in readers { + let read = moq_clock::timeout(TIMEOUT, reader.recv_group()).await; + assert!(matches!(read.expect("the track carried on"), Ok(None))); + } + } + + /// A child made after its broadcast closed, such as a request accepted late, is refused and + /// ends the way a publisher leaving does. + #[tokio::test(start_paused = true)] + async fn a_child_of_a_closed_broadcast_is_refused() { + let children = Children::default(); + children.close(); + let track = moq_net::broadcast::Info::new() + .produce() + .create_track("late", None) + .unwrap(); + let mut reader = track.subscribe(None); + let mut tracks = NonZeroSlab::default(); + let kept = Kept { + inner: track, + children: children.clone(), + }; + assert!(matches!( + children.insert(&mut tracks, kept, Child::Track), + Err(Error::BroadcastNotFound) + )); + assert!(matches!(reader.recv_group().await, Err(moq_net::Error::Unroutable))); + } +} + #[cfg(test)] mod supervision_tests { use super::*; @@ -1211,7 +1566,7 @@ mod supervision_tests { catalog: original.catalog.clone(), video: Default::default(), audio: Default::default(), - finished: Default::default(), + children: Default::default(), ended: false, } }; @@ -1265,11 +1620,12 @@ mod supervision_tests { #[tokio::test] async fn a_finish_after_a_driver_panic_reports_it_until_the_close() { let (origin, completion, trigger) = crate::ffi::panicking_origin(); - let mut publish = Publish::default(); - let broadcast = publish + let mut state = State::new(); + let broadcast = state + .publish .create_publication((origin.create_broadcast("retained").unwrap(), completion.clone())) .unwrap(); - let producer = &publish.broadcast(broadcast).unwrap().producer; + let producer = &state.publish.broadcast(broadcast).unwrap().producer; // A dynamic handle outlives the consumed broadcast, so only an explicit close ends it. let _dynamic = producer.dynamic(); let consumer = producer.consume(); @@ -1277,10 +1633,13 @@ mod supervision_tests { .write() .unwrap_or_else(|_| unreachable!("test keeps its panic trigger open")) = true; assert!(matches!(completion.finished().await, Err(Error::Panic))); - assert!(matches!(publish.catalog_finish(broadcast), Err(Error::Panic))); + assert!(matches!(state.publish.catalog_finish(broadcast), Err(Error::Panic))); // The handle stays the caller's to close, and the close still ends the broadcast. assert!(!consumer.is_closed()); - assert!(matches!(publish.close(broadcast), Err(Error::Panic))); + assert!(matches!( + state.publish_close(broadcast).unwrap().run(), + Err(Error::Panic) + )); assert!(consumer.is_closed()); } } diff --git a/rs/moq-c/src/test.rs b/rs/moq-c/src/test.rs index 5a0c5e9c29d..86527a0815c 100644 --- a/rs/moq-c/src/test.rs +++ b/rs/moq-c/src/test.rs @@ -3223,9 +3223,10 @@ fn dynamic_track_request_publishes_media() { assert_eq!(moq_origin_close(origin), 0); } -/// A raw track request already delivered stays acceptable after its broadcast handle is closed. +/// A raw track request accepted after its broadcast closed is refused, and its subscriber sees the +/// publisher leave instead of a track that outlives the broadcast. #[test] -fn track_request_accepts_after_the_broadcast_closes() { +fn track_request_accepted_after_the_broadcast_closes_is_refused() { let origin = id(moq_origin_create()); let broadcast = publish_broadcast(origin, b"dynamic-closed"); let request_cb = Callback::new(); @@ -3237,10 +3238,11 @@ fn track_request_accepts_after_the_broadcast_closes() { let request = id(request_cb.recv()); assert_eq!(moq_publish_close(broadcast), 0); - let track = id(unsafe { moq_track_request_accept(request, std::ptr::null()) }); - assert_eq!(moq_publish_track_finish(track), 0); - // The subscription ends with the finished track, before its callback's channel is freed. - assert_eq!(frame_cb.recv_terminal(), 0); + assert_eq!( + unsafe { moq_track_request_accept(request, std::ptr::null()) }, + MOQ_ERROR_BROADCAST_NOT_FOUND + ); + assert!(frame_cb.recv_terminal() < 0, "the subscription ended cleanly"); assert_eq!(request_cb.recv_terminal(), 0); assert_eq!(moq_consume_close(consume), 0); assert_eq!(moq_origin_close(origin), 0); @@ -4672,6 +4674,60 @@ fn a_stalled_encode_does_not_block_unrelated_calls() { assert_eq!(moq_origin_close(origin), 0); } +/// Closing a broadcast whose encoder is mid-encode waits for that encode without holding the +/// global lock, so unrelated calls go on meanwhile. +#[test] +fn closing_a_broadcast_does_not_block_unrelated_calls_behind_its_encoder() { + let origin = id(moq_origin_create()); + let broadcast = publish_broadcast(origin, b"video-close-stall-test"); + + let input = moq_video_encoder_input { + format: moq_video_pixel_format::MOQ_VIDEO_PIXEL_FORMAT_RGBA as u32, + width: 320, + height: 240, + framerate: moq_video_frame_rate { + numerator: 30, + denominator: 1, + }, + }; + let output = moq_video_encoder_output { + codec: moq_video_codec::MOQ_VIDEO_CODEC_H264 as u32, + bitrate: 0, + gop: 0, + kind: moq_video_encoder_kind::MOQ_VIDEO_ENCODER_KIND_SOFTWARE as u32, + encoder: std::ptr::null(), + encoder_len: 0, + }; + let stalled = id(unsafe { moq_encode_video(broadcast, &input, &output, 0) }); + + // Hold the lock an encode takes for the duration of its codec call. + let handle = State::lock().video.producer(Id::try_from(stalled).unwrap()).unwrap(); + let held = handle.lock(); + let closing = std::thread::spawn(move || moq_publish_close(broadcast)); + + // Once the close released the broadcast, unrelated calls must not wait on the encoder. On + // its own thread so a regression fails this assertion rather than hanging the test. + let (tx, rx) = mpsc::channel(); + let unrelated = std::thread::spawn(move || { + let broadcast = Id::try_from(broadcast).unwrap(); + while State::lock().publish.catalog_snapshot(broadcast).is_ok() { + std::thread::yield_now(); + } + let _ = tx.send(moq_origin_create()); + }); + let created = rx + .recv_timeout(TIMEOUT) + .expect("an unrelated call was waiting on the closing encoder"); + assert!(created > 0, "creating an origin failed while a close was waiting"); + unrelated.join().unwrap(); + + drop(held); + assert_eq!(closing.join().unwrap(), 0); + assert!(moq_encode_video_finish(stalled) < 0, "the close released the encoder"); + assert_eq!(moq_origin_close(id(created)), 0); + assert_eq!(moq_origin_close(origin), 0); +} + /// A raw video producer rejects a buffer that isn't one picture at the /// configured resolution, rather than reinterpreting it. #[test] diff --git a/rs/moq-c/src/video.rs b/rs/moq-c/src/video.rs index ced9a90b0b3..a878240f5dd 100644 --- a/rs/moq-c/src/video.rs +++ b/rs/moq-c/src/video.rs @@ -319,7 +319,16 @@ pub(crate) struct VideoEncoder { /// grant. Behind the encoder's lock, so the follower never reads an old, higher /// one and retunes over a newer cap. ceiling: moq_net::bandwidth::Rate, - finished: crate::publish::Finished, + children: crate::publish::Children, +} + +impl crate::publish::Abort for Shared { + fn abort(self, err: moq_net::Error) { + // Dropping the follow sender stops the rate follower; the encoder closes with it. + if let Some(encoder) = self.take() { + encoder.producer.abort(err); + } + } } /// A delivered frame: the decoded picture, retained until its id is freed, and @@ -393,16 +402,18 @@ pub struct DecoderOutput { fn finalize( mut producer: moq_video::encode::Producer, drained: Result<(), moq_video::Error>, - finished: &crate::publish::Finished, + children: &crate::publish::Children, + child: crate::publish::Child, ) -> Result<(), Error> { match drained { Ok(()) => { producer.finish()?; // The broadcast keeps the finished track, so its finish still holds it. - finished.keep(producer); + children.keep(child, producer); Ok(()) } Err(err) => { + children.forget(child); producer.abort(moq_net::Error::Transport(err.to_string())); Err(err.into()) } @@ -454,18 +465,19 @@ impl VideoEncoder { Ok(()) } - fn publish_finish(self) -> Result<(), Error> { + /// Drain and finish the encoder behind handle `id`. + fn publish_finish(self, id: Id) -> Result<(), Error> { let VideoEncoder { encoder, mut producer, followed, - finished, + children, .. } = self; // Drain the codec into the track before ending it, so the last frames land // in it rather than being dropped with the encoder. let drained = block_on(encoder.finish()).and_then(|encoded| producer.publish(&encoded)); - finalize(producer, drained, &finished)?; + finalize(producer, drained, &children, crate::publish::Child::Video(id))?; // A follower panic only stops rate adaptation, so the track still ends // cleanly before the panic is reported. followed.map_or(Ok(()), |followed| followed.check()) @@ -492,8 +504,7 @@ impl Video { .with_timeline_hint(config.gop.duration(config.framerate)); let ceiling = video_ceiling(config, &rendition); let producer = moq_video::encode::Producer::new(target.producer.clone(), reserved, rendition)?; - let finished = target.finished.clone(); - self.producers.insert(Shared::new(VideoEncoder { + let encoder = Shared::new(VideoEncoder { encoder, producer, format, @@ -502,8 +513,11 @@ impl Video { follow: None, followed: None, ceiling, - finished, - })) + children: target.children.clone(), + }); + target + .children + .insert(&mut self.producers, encoder, crate::publish::Child::Video) } /// Reserve the encoder's ceiling against `allocator` and follow the grant. @@ -898,11 +912,18 @@ pub extern "C" fn moq_encode_video_bitrate(producer: u32, bitrate: u64) -> i32 { #[unsafe(no_mangle)] pub extern "C" fn moq_encode_video_finish(producer: u32) -> i32 { ffi::enter(move || { - let producer = ffi::parse_id(producer)?; - // The id is dropped first, so nothing new queues behind the drain; whatever - // is mid-encode still finishes before this takes the encoder. - let producer = State::lock().video.remove(producer)?; - producer.take().ok_or(Error::MediaNotFound)?.publish_finish() + let id = ffi::parse_id(producer)?; + // The id stays until the drain ends, so a racing moq_publish_close waits it out + // instead of missing a track that still publishes. A call queued behind the drain + // finds the encoder gone. + let shared = State::lock().video.producer(id)?; + let finished = { + let mut guard = shared.lock(); + guard.take().ok_or(Error::MediaNotFound)?.publish_finish(id) + }; + // A racing close may have released the id already. + let _ = State::lock().video.remove(id); + finished }) } @@ -1060,7 +1081,13 @@ mod tests { #[tokio::test] async fn a_successful_drain_ends_the_track_cleanly() { let (producer, mut track) = track_under_test().await; - finalize(producer, Ok(()), &Default::default()).unwrap(); + finalize( + producer, + Ok(()), + &Default::default(), + crate::publish::Child::Video(Id::try_from(1u32).unwrap()), + ) + .unwrap(); assert!(matches!(track.recv_group().await, Ok(None)), "expected a clean end"); } @@ -1071,7 +1098,13 @@ mod tests { async fn a_failed_drain_aborts_the_track() { let (producer, mut track) = track_under_test().await; let err = moq_video::Error::Codec(anyhow::anyhow!("the codec lost the tail")); - finalize(producer, Err(err), &Default::default()).unwrap_err(); + finalize( + producer, + Err(err), + &Default::default(), + crate::publish::Child::Video(Id::try_from(1u32).unwrap()), + ) + .unwrap_err(); let Err(err) = track.recv_group().await else { panic!("expected an abort, not a clean end"); @@ -1101,10 +1134,11 @@ mod tests { follow: None, followed: Some(followed), ceiling: moq_net::bandwidth::Rate::from_mbps(1), - finished: Default::default(), + children: Default::default(), }; - let result = tokio::task::spawn_blocking(move || video.publish_finish()) + let id = Id::try_from(1u32).unwrap(); + let result = tokio::task::spawn_blocking(move || video.publish_finish(id)) .await .unwrap(); assert!(matches!(result, Err(Error::Panic)), "{result:?}"); diff --git a/rs/moq-ffi/src/audio.rs b/rs/moq-ffi/src/audio.rs index b12f3e5ae12..49e9e95e5aa 100644 --- a/rs/moq-ffi/src/audio.rs +++ b/rs/moq-ffi/src/audio.rs @@ -338,13 +338,33 @@ impl TryFrom for moq_audio::Frame { /// [`write`](Self::write) accepts an [`MoqAudioFrame`] whose `data` /// is PCM in the format declared by the [`MoqAudioEncoderInput`] /// passed at publish time. +/// +/// Released before its finish, or still open when its broadcast closes, it ends its track with +/// `Unroutable`, which readers take as the publisher leaving. #[derive(uniffi::Object)] pub struct MoqAudioProducer { inner: std::sync::Mutex>, reservation: std::sync::Mutex>>, /// Held so the reservation's registry outlives extra bandwidth handles. _bandwidth: Option>, - finished: crate::producer::Finished, + children: crate::producer::Children, +} + +// Dropped or closed before its finish, the track ends the way a publisher leaving does. +impl crate::producer::Child for MoqAudioProducer { + fn abort(&self, err: moq_net::Error) { + let _guard = crate::ffi::runtime().enter(); + if let Some(producer) = self.inner.lock().unwrap().take() { + self.reservation.lock().unwrap().take(); + producer.abort(err); + } + } +} + +impl Drop for MoqAudioProducer { + fn drop(&mut self) { + crate::producer::Child::abort(self, moq_net::Error::Unroutable); + } } impl MoqAudioProducer { @@ -396,10 +416,17 @@ impl MoqAudioProducer { /// The broadcast keeps the finished track until it ends, so its finish still holds it. pub fn finish(&self) -> Result<(), MoqError> { let _guard = crate::ffi::runtime().enter(); - let mut producer = self.inner.lock().unwrap().take().ok_or(MoqError::Closed)?; + // Held through the flush, so a racing broadcast close waits it out instead of + // missing a track that still publishes. + let mut inner = self.inner.lock().unwrap(); + let mut producer = inner.take().ok_or(MoqError::Closed)?; self.reservation.lock().unwrap().take(); - producer.finish()?; - self.finished.keep(producer); + if let Err(err) = producer.finish() { + // Readers see the failure rather than `Dropped`, as a failed video drain does. + producer.abort(moq_net::Error::Transport(err.to_string())); + return Err(err.into()); + } + self.children.keep(producer); Ok(()) } } @@ -457,10 +484,10 @@ impl MoqAudioProducer { options.bandwidth = bandwidth.allocator().clone(); } - let (producer, finished) = broadcast.with_state(|state| { + let (producer, children) = broadcast.with_state(|state| { let producer = moq_audio::encode::Producer::new(&mut state.broadcast, state.catalog.clone(), input, &options)?; - Ok((producer, state.finished.clone())) + Ok((producer, state.children.clone())) })?; // Producer::new does not reserve today (only capture does), so the @@ -470,11 +497,11 @@ impl MoqAudioProducer { .as_ref() .map(|bandwidth| bandwidth.reserve_demand(&producer.demand(), producer.bitrate().as_bps())); - Ok(Arc::new(MoqAudioProducer { + children.adopt(Arc::new(MoqAudioProducer { inner: std::sync::Mutex::new(Some(producer)), reservation: std::sync::Mutex::new(reservation), _bandwidth: bandwidth, - finished, + children: children.clone(), })) } } diff --git a/rs/moq-ffi/src/flate.rs b/rs/moq-ffi/src/flate.rs index 289c4874488..2340e5805db 100644 --- a/rs/moq-ffi/src/flate.rs +++ b/rs/moq-ffi/src/flate.rs @@ -11,7 +11,7 @@ use std::sync::Arc; use moq_mux::binary::Config; use crate::error::MoqError; -use crate::producer::{MoqBroadcastProducer, MoqTrackProducer}; +use crate::producer::{Child, MoqBroadcastProducer, MoqTrackProducer}; /// Options for an opaque data track, in either mode (the mode is fixed by the constructor). #[derive(Clone, uniffi::Record)] @@ -36,11 +36,30 @@ impl From for Config { } /// Publishes opaque payloads that consumers see as a single latest value. +/// +/// Released before its finish, or still open when its broadcast closes, it ends its track with +/// `Unroutable`, which readers take as the publisher leaving. #[derive(uniffi::Object)] pub struct MoqFlateSnapshotProducer { inner: std::sync::Mutex>, } +impl Child for MoqFlateSnapshotProducer { + fn abort(&self, err: moq_net::Error) { + let _guard = crate::ffi::enter(); + let producer = self.inner.lock().unwrap().take(); + if let Some(producer) = producer { + let _ = producer.abort(err); + } + } +} + +impl Drop for MoqFlateSnapshotProducer { + fn drop(&mut self) { + self.abort(moq_net::Error::Unroutable); + } +} + #[uniffi::export] impl MoqFlateSnapshotProducer { /// Publish `track` as an opaque snapshot track (lossy latest-value), advertised in `broadcast`'s catalog. @@ -58,11 +77,12 @@ impl MoqFlateSnapshotProducer { ) -> Result, MoqError> { let _guard = crate::ffi::enter(); let config = Config::from(config); - let producer = - track.adopt(|track| broadcast.with_state(|state| Ok(state.catalog.binary_snapshot(track, config)?)))?; - Ok(Arc::new(Self { - inner: std::sync::Mutex::new(Some(producer)), - })) + track.adopt(|track| { + let producer = broadcast.with_state(|state| Ok(state.catalog.binary_snapshot(track, config)?))?; + Ok(Self { + inner: std::sync::Mutex::new(Some(producer)), + }) + }) } /// Publish a new payload, superseding the last. @@ -88,11 +108,30 @@ impl MoqFlateSnapshotProducer { } /// Publishes an ordered log of opaque payloads, one per append. +/// +/// Released before its finish, or still open when its broadcast closes, it ends its track with +/// `Unroutable`, which readers take as the publisher leaving. #[derive(uniffi::Object)] pub struct MoqFlateStreamProducer { inner: std::sync::Mutex>, } +impl Child for MoqFlateStreamProducer { + fn abort(&self, err: moq_net::Error) { + let _guard = crate::ffi::enter(); + let producer = self.inner.lock().unwrap().take(); + if let Some(producer) = producer { + let _ = producer.abort(err); + } + } +} + +impl Drop for MoqFlateStreamProducer { + fn drop(&mut self) { + self.abort(moq_net::Error::Unroutable); + } +} + #[uniffi::export] impl MoqFlateStreamProducer { /// Publish `track` as an opaque stream track (lossless append-log), advertised in `broadcast`'s catalog. @@ -110,11 +149,12 @@ impl MoqFlateStreamProducer { ) -> Result, MoqError> { let _guard = crate::ffi::enter(); let config = Config::from(config); - let producer = - track.adopt(|track| broadcast.with_state(|state| Ok(state.catalog.binary_stream(track, config)?)))?; - Ok(Arc::new(Self { - inner: std::sync::Mutex::new(Some(producer)), - })) + track.adopt(|track| { + let producer = broadcast.with_state(|state| Ok(state.catalog.binary_stream(track, config)?))?; + Ok(Self { + inner: std::sync::Mutex::new(Some(producer)), + }) + }) } /// Append one payload to the log. diff --git a/rs/moq-ffi/src/json.rs b/rs/moq-ffi/src/json.rs index 08a3d85cb8a..9e09f9bce76 100644 --- a/rs/moq-ffi/src/json.rs +++ b/rs/moq-ffi/src/json.rs @@ -15,7 +15,7 @@ use crate::consumer::{MoqTrackConsumer, timestamp_us}; use crate::demand::MoqTrackDemand; use crate::error::MoqError; use crate::ffi::Task; -use crate::producer::{MoqBroadcastProducer, MoqTrackProducer}; +use crate::producer::{Child, MoqBroadcastProducer, MoqTrackProducer}; /// Options for a JSON snapshot track (lossy latest-value mode). /// @@ -158,6 +158,9 @@ mod tests { /// A join that finds more deltas than the track's `refresh` ratio allows gets the value restated /// in a fresh group, by a task this producer runs until it is finished or dropped. That joiner /// still reads the cached group first. +/// +/// Released before its finish, or still open when its broadcast closes, it ends its track with +/// `Unroutable`, which readers take as the publisher leaving. #[derive(uniffi::Object)] pub struct MoqJsonSnapshotProducer { inner: Arc>>>, @@ -190,11 +193,20 @@ impl MoqJsonSnapshotProducer { } } +impl Child for MoqJsonSnapshotProducer { + fn abort(&self, err: moq_net::Error) { + let _guard = crate::ffi::enter(); + let producer = self.inner.lock().unwrap().take(); + let _ = self.alive.close(); + if let Some(producer) = producer { + let _ = producer.abort(err); + } + } +} + impl Drop for MoqJsonSnapshotProducer { fn drop(&mut self) { - // The refresh task shares `inner`, so drop the producer here rather than whenever the task - // next wakes: dropping it retires the catalog entry. - self.inner.lock().unwrap().take(); + self.abort(moq_net::Error::Unroutable); } } @@ -218,13 +230,16 @@ impl MoqJsonSnapshotProducer { let _guard = crate::ffi::enter(); let options = moq_mux::json::Options::default().with_deltas(config.deltas); let config = moq_mux::json::Config::default().with_compression(config.compression); - let producer = track.adopt(|track| { - broadcast.with_state(|state| Ok(state.catalog.json_snapshot::(track, config, options)?)) + let this = track.adopt(|track| { + let producer = + broadcast.with_state(|state| Ok(state.catalog.json_snapshot::(track, config, options)?))?; + Ok(Self { + inner: Arc::new(std::sync::Mutex::new(Some(producer))), + alive: kio::Producer::new(()), + }) })?; - let inner = Arc::new(std::sync::Mutex::new(Some(producer))); - let alive = kio::Producer::new(()); - crate::ffi::spawn(Self::refresh(inner.clone(), alive.consume())); - Ok(Arc::new(Self { inner, alive })) + crate::ffi::spawn(Self::refresh(this.inner.clone(), this.alive.consume())); + Ok(this) } /// Publish a new value, encoded as a snapshot or delta automatically. `value` is a JSON @@ -336,11 +351,30 @@ impl MoqJsonSnapshotConsumer { // ---- Stream ---- /// Publishes an ordered log of JSON records, one record per append. +/// +/// Released before its finish, or still open when its broadcast closes, it ends its track with +/// `Unroutable`, which readers take as the publisher leaving. #[derive(uniffi::Object)] pub struct MoqJsonStreamProducer { inner: std::sync::Mutex>>, } +impl Child for MoqJsonStreamProducer { + fn abort(&self, err: moq_net::Error) { + let _guard = crate::ffi::enter(); + let producer = self.inner.lock().unwrap().take(); + if let Some(producer) = producer { + let _ = producer.abort(err); + } + } +} + +impl Drop for MoqJsonStreamProducer { + fn drop(&mut self) { + self.abort(moq_net::Error::Unroutable); + } +} + #[uniffi::export] impl MoqJsonStreamProducer { /// Publish `track` as a JSON stream track (lossless append-log), advertised in `broadcast`'s catalog. @@ -359,11 +393,12 @@ impl MoqJsonStreamProducer { ) -> Result, MoqError> { let _guard = crate::ffi::enter(); let config = moq_mux::json::Config::default().with_compression(config.compression); - let producer = track - .adopt(|track| broadcast.with_state(|state| Ok(state.catalog.json_stream::(track, config)?)))?; - Ok(Arc::new(Self { - inner: std::sync::Mutex::new(Some(producer)), - })) + track.adopt(|track| { + let producer = broadcast.with_state(|state| Ok(state.catalog.json_stream::(track, config)?))?; + Ok(Self { + inner: std::sync::Mutex::new(Some(producer)), + }) + }) } /// Append one record to the log. `value` is a JSON document. @@ -447,6 +482,9 @@ impl MoqJsonStreamConsumer { /// /// A join that finds the open group's edits outweigh the track's `refresh` ratio gets the window /// restated in a fresh group, by a task this producer runs until it is finished or dropped. +/// +/// Released before its finish, or still open when its broadcast closes, it ends its track with +/// `Unroutable`, which readers take as the publisher leaving. #[derive(uniffi::Object)] pub struct MoqJsonWindowProducer { inner: Arc>>>, @@ -492,11 +530,20 @@ impl MoqJsonWindowProducer { } } +impl Child for MoqJsonWindowProducer { + fn abort(&self, err: moq_net::Error) { + let _guard = crate::ffi::enter(); + let producer = self.inner.lock().unwrap().take(); + let _ = self.alive.close(); + if let Some(producer) = producer { + let _ = producer.abort(err); + } + } +} + impl Drop for MoqJsonWindowProducer { fn drop(&mut self) { - // The refresh task shares `inner`, so release the track here rather than whenever the task - // next wakes. - self.inner.lock().unwrap().take(); + self.abort(moq_net::Error::Unroutable); } } @@ -511,14 +558,17 @@ impl MoqJsonWindowProducer { pub fn new(track: &MoqTrackProducer, config: MoqJsonWindowConfig) -> Result, MoqError> { let _guard = crate::ffi::enter(); let config = config.try_into()?; - let (producer, demand) = track.adopt(|track| { - let demand = track.demand(); - Ok((moq_json::window::Producer::new(track, config), demand)) + let this = track.adopt(|track| { + Ok(Self { + demand: track.demand(), + inner: Arc::new(std::sync::Mutex::new(Some(moq_json::window::Producer::new( + track, config, + )))), + alive: kio::Producer::new(()), + }) })?; - let inner = Arc::new(std::sync::Mutex::new(Some(producer))); - let alive = kio::Producer::new(()); - crate::ffi::spawn(Self::refresh(inner.clone(), alive.consume())); - Ok(Arc::new(Self { inner, demand, alive })) + crate::ffi::spawn(Self::refresh(this.inner.clone(), this.alive.consume())); + Ok(this) } /// Append one record to the back of the window. `value` is a JSON document. diff --git a/rs/moq-ffi/src/media.rs b/rs/moq-ffi/src/media.rs index 29d17f806dc..a7203d1e192 100644 --- a/rs/moq-ffi/src/media.rs +++ b/rs/moq-ffi/src/media.rs @@ -12,7 +12,7 @@ use crate::consumer::{ use crate::demand::MoqTrackDemand; use crate::error::MoqError; use crate::ffi::Task; -use crate::producer::{MoqBroadcastProducer, MoqTrackRequest}; +use crate::producer::{Child, Children, MoqBroadcastProducer, MoqTrackRequest}; /// A width and height in pixels. #[derive(Clone, uniffi::Record)] @@ -399,7 +399,7 @@ fn reserve_container( moq_net::broadcast::Producer, moq_mux::catalog::Reserved, crate::ffi::Background, - crate::producer::Finished, + crate::producer::Children, ), MoqError, > { @@ -412,7 +412,7 @@ fn reserve_container( state.broadcast.clone(), reserved, state.driver.clone(), - state.finished.clone(), + state.children.clone(), )) }) } @@ -814,25 +814,81 @@ struct MediaStreamProducer { } /// Imports a container into a broadcast, from chunks split anywhere such as a raw byte stream. +/// +/// Released before its finish, or still open when its broadcast closes, it ends its tracks with +/// `Unroutable`, which readers take as the publisher leaving. #[derive(uniffi::Object)] pub struct MoqMediaContainerProducer { driver: crate::ffi::Background, inner: std::sync::Mutex>, - finished: crate::producer::Finished, + children: Children, } /// Imports complete encoded frames onto one track. +/// +/// Released before its finish, or still open when its broadcast closes, it ends its track with +/// `Unroutable`, which readers take as the publisher leaving. #[derive(uniffi::Object)] pub struct MoqMediaTrackProducer { inner: std::sync::Mutex>, - finished: crate::producer::Finished, + children: Children, } /// Imports a video byte stream onto one track. +/// +/// Released before its finish, or still open when its broadcast closes, it ends its track with +/// `Unroutable`, which readers take as the publisher leaving. #[derive(uniffi::Object)] pub struct MoqMediaTrackStreamProducer { inner: std::sync::Mutex>, - finished: crate::producer::Finished, + children: Children, +} + +// A media handle dropped or closed before its finish ends its tracks the way a publisher +// leaving does, never as a cause-less drop. +impl Child for MoqMediaContainerProducer { + fn abort(&self, err: moq_net::Error) { + let _guard = crate::ffi::enter(); + if let Some(container) = self.inner.lock().unwrap().take() { + container.import.abort(err); + } + } +} + +impl Drop for MoqMediaContainerProducer { + fn drop(&mut self) { + self.abort(moq_net::Error::Unroutable); + } +} + +impl Child for MoqMediaTrackProducer { + fn abort(&self, err: moq_net::Error) { + let _guard = crate::ffi::enter(); + if let Some(media) = self.inner.lock().unwrap().take() { + media.import.abort(err); + } + } +} + +impl Drop for MoqMediaTrackProducer { + fn drop(&mut self) { + self.abort(moq_net::Error::Unroutable); + } +} + +impl Child for MoqMediaTrackStreamProducer { + fn abort(&self, err: moq_net::Error) { + let _guard = crate::ffi::enter(); + if let Some(media) = self.inner.lock().unwrap().take() { + media.import.abort(err); + } + } +} + +impl Drop for MoqMediaTrackStreamProducer { + fn drop(&mut self) { + self.abort(moq_net::Error::Unroutable); + } } #[uniffi::export] @@ -856,7 +912,7 @@ impl MoqMediaTrackProducer { reason: format!("init failed: {err}"), } })?; - Ok(MoqMediaTrackProducer::from_import(import, state.finished.clone())) + MoqMediaTrackProducer::from_import(import, &state.children) }) } } @@ -883,7 +939,7 @@ impl MoqMediaTrackProducer { reason: format!("init failed: {err}"), } })?; - Ok(MoqMediaTrackProducer::from_import(import, state.finished.clone())) + MoqMediaTrackProducer::from_import(import, &state.children) }) } } @@ -900,15 +956,15 @@ impl MoqMediaContainerProducer { ) -> Result, MoqError> { let _guard = crate::ffi::enter(); let native = moq_mux::import::ContainerInit::new(init.format.into(), init.data); - let (target, reserved, driver, finished) = reserve_container(broadcast, init.timebase.as_deref())?; + let (target, reserved, driver, children) = reserve_container(broadcast, init.timebase.as_deref())?; let import = moq_mux::import::Container::new(target, reserved, &native).map_err(|err| MoqError::Codec { reason: format!("init failed: {err}"), })?; - Ok(Arc::new(MoqMediaContainerProducer { + children.adopt(Arc::new(MoqMediaContainerProducer { driver, inner: std::sync::Mutex::new(Some(ContainerProducer { import })), - finished, + children: children.clone(), })) } } @@ -937,11 +993,11 @@ impl MoqMediaTrackStreamProducer { } })?; - Ok(Arc::new(MoqMediaTrackStreamProducer { + state.children.adopt(Arc::new(MoqMediaTrackStreamProducer { inner: std::sync::Mutex::new(Some(MediaStreamProducer { import: Box::new(import), })), - finished: state.finished.clone(), + children: state.children.clone(), })) }) } @@ -949,15 +1005,15 @@ impl MoqMediaTrackStreamProducer { impl MoqMediaTrackProducer { /// Wrap a single-codec importer, capturing the demand handle its track exposes. - fn from_import(import: moq_mux::import::Track, finished: crate::producer::Finished) -> Arc { + fn from_import(import: moq_mux::import::Track, children: &Children) -> Result, MoqError> { let demand = import.demand(); - Arc::new(Self { + children.adopt(Arc::new(Self { inner: std::sync::Mutex::new(Some(MediaProducer { import: Box::new(import), demand, })), - finished, - }) + children: children.clone(), + })) } } @@ -1038,15 +1094,19 @@ impl MoqMediaTrackProducer { /// Finish this track and finalize encoding. /// - /// The broadcast keeps the finished track until it ends, so its finish still holds it. + /// The broadcast keeps the finished track until it ends, so its finish still holds it. A + /// failed finish ends the track with its error instead. pub fn finish(&self) -> Result<(), MoqError> { let _guard = crate::ffi::enter(); let mut guard = self.inner.lock().unwrap(); let mut media = guard.take().ok_or(MoqError::Closed)?; - media.import.finish().map_err(|err| MoqError::Codec { - reason: format!("finish failed: {err}"), - })?; - self.finished.keep(media); + if let Err(err) = media.import.finish() { + media.import.abort(cause(&err)); + return Err(MoqError::Codec { + reason: format!("finish failed: {err}"), + }); + } + self.children.keep(media); Ok(()) } } @@ -1112,16 +1172,20 @@ impl MoqMediaContainerProducer { /// Finish every track this container publishes. /// - /// The broadcast keeps the finished tracks until it ends, so its finish still holds them. + /// The broadcast keeps the finished tracks until it ends, so its finish still holds them. A + /// failed finish ends the tracks with its error instead. pub fn finish(&self) -> Result<(), MoqError> { let result = (|| { let _guard = crate::ffi::enter(); let mut guard = self.inner.lock().unwrap(); let mut container = guard.take().ok_or(MoqError::Closed)?; - container.import.finish().map_err(|err| MoqError::Codec { - reason: format!("finish failed: {err}"), - })?; - self.finished.keep(container); + if let Err(err) = container.import.finish() { + container.import.abort(cause(&err)); + return Err(MoqError::Codec { + reason: format!("finish failed: {err}"), + }); + } + self.children.keep(container); Ok(()) })(); self.driver.check()?; @@ -1154,19 +1218,32 @@ impl MoqMediaTrackStreamProducer { /// Finalize the track, publishing the last access unit still waiting for a following start code. /// - /// The broadcast keeps the finished track until it ends, so its finish still holds it. + /// The broadcast keeps the finished track until it ends, so its finish still holds it. A + /// failed finish ends the track with its error instead. pub fn finish(&self) -> Result<(), MoqError> { let _guard = crate::ffi::enter(); let mut guard = self.inner.lock().unwrap(); let mut media = guard.take().ok_or(MoqError::Closed)?; - media.import.finish().map_err(|err| MoqError::Codec { - reason: format!("finish failed: {err}"), - })?; - self.finished.keep(media); + if let Err(err) = media.import.finish() { + media.import.abort(cause(&err)); + return Err(MoqError::Codec { + reason: format!("finish failed: {err}"), + }); + } + self.children.keep(media); Ok(()) } } +/// The cause a failed import ends its tracks with, so a reader sees the publisher fail rather than +/// `Dropped`. Only a moq-net error has a wire code; any other reaches readers as `Internal`. +fn cause(err: &moq_mux::Error) -> moq_net::Error { + match err { + moq_mux::Error::Moq(err) => err.clone(), + _ => moq_net::StreamError::Internal.into(), + } +} + /// Reserve the named track, or a unique one named after the format. fn reserve_track( broadcast: &moq_net::broadcast::Producer, diff --git a/rs/moq-ffi/src/producer.rs b/rs/moq-ffi/src/producer.rs index 8a573ef1315..778e35acac9 100644 --- a/rs/moq-ffi/src/producer.rs +++ b/rs/moq-ffi/src/producer.rs @@ -88,7 +88,7 @@ pub(crate) struct BroadcastProducer { // Core timebases own catalog outputs, so the broadcast holds them and binding handles stay weak. pub(crate) timebases: std::collections::HashMap>, pub(crate) driver: Background, - pub(crate) finished: Finished, + pub(crate) children: Children, /// Set by [`MoqMediaCatalogProducer::finish`](crate::media::MoqMediaCatalogProducer::finish): the /// catalog is final, so a close leaves it alone. pub(crate) ended: bool, @@ -96,41 +96,115 @@ pub(crate) struct BroadcastProducer { impl Drop for BroadcastProducer { fn drop(&mut self) { - self.finished.release(); + self.children.release(); } } -/// The finished importers a broadcast keeps, or `None` once it ended. -type Keeping = Option>>; +/// `Send + Sync` natively, where any thread may close a broadcast; nothing on single-threaded wasm. +#[cfg(not(target_family = "wasm"))] +pub(crate) trait Shared: Send + Sync {} +#[cfg(not(target_family = "wasm"))] +impl Shared for T {} +#[cfg(target_family = "wasm")] +pub(crate) trait Shared {} +#[cfg(target_family = "wasm")] +impl Shared for T {} -/// Finished producers a broadcast keeps until it ends. +/// A handle publishing tracks on a broadcast, which its broadcast ends on close. +pub(crate) trait Child: Shared { + /// End whatever this handle still publishes with `err`, so later calls fail with `Closed`. + /// + /// A track that already finished ends cleanly instead. + fn abort(&self, err: moq_net::Error); +} + +/// What a broadcast holds of its children. +enum Family { + Open { + /// Handles that may still publish, ended on close. + open: Vec>, + /// Finished producers, kept so a late subscriber still finds their tracks. + kept: Vec>, + }, + /// The broadcast handle was dropped without a close: each child ends on its own. + Released, + /// The broadcast closed: its children ended and new ones are refused. + Closed, +} + +/// The track-publishing handles of one broadcast. /// -/// A finished track stops being served once its last producer drops. An importer lets go of -/// its producer when it finishes, and a binding may release a raw track's handle whenever its -/// garbage collector runs, so the broadcast keeps them here instead, and a subscriber arriving -/// after the end still finds the track until the broadcast closes. +/// A child that drops without finishing aborts itself, so a garbage-collected binding never +/// leaves a track to drop without a cause, and closing the broadcast aborts every child still +/// open. A finished track stops being served once its last producer drops, and an importer lets +/// go of its producer when it finishes, so the broadcast keeps finished producers here instead: +/// a subscriber arriving after the end still finds the track until the broadcast closes. #[derive(Clone)] -pub(crate) struct Finished(Arc>); +pub(crate) struct Children(Arc>); -impl Default for Finished { +impl Default for Children { fn default() -> Self { - Self(Arc::new(std::sync::Mutex::new(Some(Vec::new())))) + Self(Arc::new(std::sync::Mutex::new(Family::Open { + open: Vec::new(), + kept: Vec::new(), + }))) + } +} + +impl Children { + /// Register a new child so a close ends it. Once the broadcast closed, the child is aborted + /// and refused with `Closed`. + pub(crate) fn adopt(&self, child: Arc) -> Result, MoqError> { + let mut family = self.0.lock().unwrap(); + match &mut *family { + Family::Open { open, .. } => { + open.retain(|child| child.strong_count() > 0); + let weak: std::sync::Weak = Arc::downgrade(&child) as _; + open.push(weak); + Ok(child) + } + Family::Released => Ok(child), + Family::Closed => { + drop(family); + child.abort(moq_net::Error::Unroutable); + Err(MoqError::Closed) + } + } + } + + /// Whether the broadcast closed, ending its children. + fn is_closed(&self) -> bool { + matches!(*self.0.lock().unwrap(), Family::Closed) } -} -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 = Box::new(producer); - let mut kept = self.0.lock().unwrap(); - if let Some(kept) = kept.as_mut() { + if let Family::Open { kept, .. } = &mut *self.0.lock().unwrap() { kept.push(producer); } } - /// Stop keeping, handing back what was kept so the caller decides when it drops. - fn release(&self) -> Vec> { - self.0.lock().unwrap().take().unwrap_or_default() + /// End every open child with `Unroutable`, then let go of the finished ones. + fn close(&self) { + let family = std::mem::replace(&mut *self.0.lock().unwrap(), Family::Closed); + if let Family::Open { open, kept } = family { + // Outside the lock: a child's own finish keeps into it while holding the child's lock. + for child in open.iter().filter_map(std::sync::Weak::upgrade) { + child.abort(moq_net::Error::Unroutable); + } + drop(kept); + } + } + + /// Let go of the finished producers, leaving open children to end on their own. + fn release(&self) { + let mut family = self.0.lock().unwrap(); + if matches!(*family, Family::Open { .. }) { + let released = std::mem::replace(&mut *family, Family::Released); + drop(family); + drop(released); + } } } @@ -162,7 +236,7 @@ impl MoqTrackDynamic { struct DynamicProducer { inner: moq_net::broadcast::Dynamic, - finished: Finished, + children: Children, } struct TrackDynamicProducer { @@ -175,7 +249,7 @@ impl DynamicProducer { // accepts it (raw, at a chosen timescale) or publishes media onto it (importer accepts). // The subscriber's subscribe stays pending until then. let request = self.inner.requested_track().await?; - Ok(Arc::new(MoqTrackRequest::new(request, self.finished.clone()))) + Ok(Arc::new(MoqTrackRequest::new(request, self.children.clone()))) } } @@ -203,7 +277,7 @@ impl MoqBroadcastProducer { catalog, timebases: Default::default(), driver, - finished: Default::default(), + children: Default::default(), ended: false, })), }) @@ -267,7 +341,7 @@ impl MoqBroadcastProducer { Ok(Arc::new(MoqBroadcastDynamic { task: Task::new(DynamicProducer { inner: state.broadcast.dynamic(), - finished: state.finished.clone(), + children: state.children.clone(), }) .supervised(state.driver.clone()), })) @@ -328,17 +402,15 @@ impl MoqBroadcastProducer { // Clone the broadcast handle (shared Arc internally) to get &mut access. let broadcast = state.broadcast.clone(); let producer = broadcast.create_track(name, Some(info))?; - Ok(Arc::new(MoqTrackProducer { - inner: std::sync::Mutex::new(Some(producer)), - finished: std::sync::Mutex::new(Some(state.finished.clone())), - })) + MoqTrackProducer::child(producer, &state.children) }) } /// End the broadcast for good: retract it, serve no new tracks, and finalize the catalog. /// - /// Tracks already subscribed carry on to their own end. Every later call on this - /// producer fails with `Closed`; closing again is a no-op. + /// Every track handle still open on it ends with `Unroutable`, the way a reader sees a + /// publisher leave, and its later calls fail with `Closed`. Finish a track first to end it + /// cleanly. Every later call on this producer fails with `Closed`; closing again is a no-op. pub fn close(&self) -> Result<(), MoqError> { let _guard = crate::ffi::enter(); // Hold the lock through shutdown so a concurrent close() returns only once it is done. @@ -346,7 +418,9 @@ impl MoqBroadcastProducer { let Some(mut state) = guard.take() else { return Ok(()); }; - // Close the broadcast first so it ends even if finalizing the catalog fails. + // The children first, so none outlives the broadcast it publishes on, and the broadcast + // before the catalog so it ends even if finalizing the catalog fails. + state.children.close(); state.broadcast.close(); let result = match state.ended { true => Ok(()), @@ -516,14 +590,14 @@ impl MoqFill { #[derive(uniffi::Object)] pub struct MoqTrackRequest { inner: std::sync::Mutex>, - finished: Finished, + children: Children, } impl MoqTrackRequest { - pub(crate) fn new(request: moq_net::track::Request, finished: Finished) -> Self { + pub(crate) fn new(request: moq_net::track::Request, children: Children) -> Self { Self { inner: std::sync::Mutex::new(Some(request)), - finished, + children, } } @@ -563,16 +637,15 @@ impl MoqTrackRequest { /// Accept the request as a raw track, fixing its [`MoqTrackInfo`] (timescale, etc.). /// + /// Once the broadcast closed, the track ends with `Unroutable` and this fails with `Closed`. + /// /// For media use [`crate::media::MoqMediaTrackProducer::audio`] instead, which lets /// the importer pick the timescale. pub fn accept(&self, info: Option) -> Result, MoqError> { let _guard = crate::ffi::enter(); let info = raw_track_info(info)?; let request = self.take()?; - Ok(Arc::new(MoqTrackProducer { - inner: std::sync::Mutex::new(Some(request.accept(Some(info)))), - finished: std::sync::Mutex::new(Some(self.finished.clone())), - })) + MoqTrackProducer::child(request.accept(Some(info)), &self.children) } /// Reject the request with an application error code, failing the waiting subscriber. @@ -587,14 +660,47 @@ impl MoqTrackRequest { // ---- Track Producer ---- /// A track to which groups, frames, or datagrams can be published. +/// +/// Released before its finish, or still open when its broadcast closes, it ends its track with +/// `Unroutable`, which readers take as the publisher leaving. #[derive(uniffi::Object)] pub struct MoqTrackProducer { inner: std::sync::Mutex>, // Taken by the first finish, so repeating it keeps the track only once. - finished: std::sync::Mutex>, + keep: std::sync::Mutex>, +} + +impl Child for MoqTrackProducer { + fn abort(&self, err: moq_net::Error) { + let _guard = crate::ffi::enter(); + // A finished track ends at its declared end, as when its handle drops. + let track = self.inner.lock().unwrap().take(); + if let Some(track) = track.filter(|track| track.end().is_none()) { + let _ = track.abort(err); + } + } +} + +impl Drop for MoqTrackProducer { + fn drop(&mut self) { + let _guard = crate::ffi::enter(); + // A finished track lives on in its broadcast, and may still take groups below its end. + let track = self.inner.get_mut().unwrap().take(); + if let Some(track) = track.filter(|track| track.end().is_none()) { + let _ = track.abort(moq_net::Error::Unroutable); + } + } } impl MoqTrackProducer { + /// Wrap `track` as a child of the broadcast `children` belongs to. + fn child(track: moq_net::track::Producer, children: &Children) -> Result, MoqError> { + children.adopt(Arc::new(Self { + inner: std::sync::Mutex::new(Some(track)), + keep: std::sync::Mutex::new(Some(children.clone())), + })) + } + pub(crate) fn track_demand(&self) -> Result { let guard = self.inner.lock().unwrap(); let track = guard.as_ref().ok_or(MoqError::Closed)?; @@ -602,16 +708,33 @@ impl MoqTrackProducer { } /// Hand the track to a typed writer that `f` builds (a JSON producer, say), closing this - /// handle once `f` succeeds so the writer is the track's only producer. On failure the - /// handle stays open. - pub(crate) fn adopt( + /// handle once `f` succeeds so the writer is the track's only producer. The writer takes this + /// handle's place among its broadcast's children. On failure the handle stays open. + pub(crate) fn adopt( &self, - f: impl FnOnce(moq_net::track::Producer) -> Result, - ) -> Result { - let mut guard = self.inner.lock().unwrap(); - let out = f(guard.as_ref().ok_or(MoqError::Closed)?.clone())?; - guard.take(); - Ok(out) + f: impl FnOnce(moq_net::track::Producer) -> Result, + ) -> Result, MoqError> { + // Taken for the whole construction, so a racing adopt finds the handle closed. `f` runs + // outside the lock: it may take the broadcast's, which a close holds while it takes this one. + let track = self.inner.lock().unwrap().take().ok_or(MoqError::Closed)?; + let writer = match f(track.clone()) { + Ok(writer) => Arc::new(writer), + Err(err) => { + *self.inner.lock().unwrap() = Some(track); + // A close during `f` found nothing to end, so the restored track ends here. + let children = self.keep.lock().unwrap().clone(); + if children.is_some_and(|children| children.is_closed()) { + Child::abort(self, moq_net::Error::Unroutable); + } + return Err(err); + } + }; + // A finished track has no family left to join, and ends at its declared end. + let children = self.keep.lock().unwrap().clone(); + match children { + Some(children) => children.adopt(writer), + None => Ok(writer), + } } } @@ -723,8 +846,8 @@ impl MoqTrackProducer { if track.end().is_none() { track.finish()?; } - if let Some(finished) = self.finished.lock().unwrap().take() { - finished.keep(track.clone()); + if let Some(children) = self.keep.lock().unwrap().take() { + children.keep(track.clone()); } Ok(()) } @@ -838,10 +961,12 @@ mod metadata_tests { track.finish().unwrap(); } let state = broadcast.state.lock().unwrap(); - let finished = &state.as_ref().unwrap().finished; - let kept = finished.0.lock().unwrap().as_ref().map(Vec::len); + let kept = match &*state.as_ref().unwrap().children.0.lock().unwrap() { + Family::Open { kept, .. } => kept.len(), + _ => unreachable!("the broadcast is still open"), + }; drop(state); - assert_eq!(kept, Some(1)); + assert_eq!(kept, 1); } #[test] diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index 171ed31eaee..0211c489fa1 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -1832,9 +1832,17 @@ async fn track_request_demand_survives_accept() { demand.until_wanted().await.unwrap(); consumer.cancel(); demand.until_unwanted().await.unwrap(); + // Released unfinished, the track aborts, and its watcher keeps the cause. drop(track); broadcast.close().unwrap(); - assert!(matches!(demand.until_wanted().await, Err(MoqError::Closed))); + let Err(err) = demand.until_wanted().await else { + panic!("the watcher outlived its track"); + }; + assert_protocol( + &err, + crate::error::MoqErrorScope::Stream, + crate::error::MoqProtocolKind::Unroutable, + ); } #[tokio::test] @@ -4135,6 +4143,250 @@ async fn finish_serves_finished_tracks_until_close() { assert!(matches!(catalog.finish(), Err(MoqError::Closed))); } +/// One raw and one media handle on `broadcast`, each with a subscriber reading it. +async fn children( + broadcast: &MoqBroadcastProducer, +) -> ( + Arc, + Arc, + [Arc; 2], +) { + let consumer = broadcast.consume().unwrap(); + let raw = broadcast.publish_track("raw".into(), Some(timed_info())).unwrap(); + let media = MoqMediaTrackProducer::audio( + broadcast, + MoqMediaTarget::Named { + name: Some("media".into()), + }, + audio_init(MoqAudioFormat::Opus, opus_head()), + ) + .unwrap(); + let readers = [ + consumer.subscribe_track("raw".into(), None).await.unwrap(), + consumer.subscribe_track("media".into(), None).await.unwrap(), + ]; + (raw, media, readers) +} + +/// Each reader ends with `Unroutable`, the cause of a publisher leaving. +async fn assert_left(readers: [Arc; N]) { + for reader in readers { + let read = tokio::time::timeout(TIMEOUT, reader.read_frame()).await; + let Err(err) = read.expect("the track carried on") else { + panic!("the track ended without a cause"); + }; + assert_protocol( + &err, + crate::error::MoqErrorScope::Stream, + crate::error::MoqProtocolKind::Unroutable, + ); + } +} + +/// Closing a broadcast ends every track handle still open on it the way a publisher leaving +/// does, and the handles refuse later calls. +#[tokio::test(start_paused = true)] +async fn closing_a_broadcast_ends_its_open_tracks() { + let broadcast = MoqBroadcastProducer::new().unwrap(); + let (raw, media, readers) = children(&broadcast).await; + broadcast.close().unwrap(); + assert_left(readers).await; + + let frame = MoqFrame { + payload: b"late".to_vec(), + timestamp_us: Some(0), + }; + assert!(matches!(raw.write_frame(frame.clone()), Err(MoqError::Closed))); + assert!(matches!(raw.finish(), Err(MoqError::Closed))); + assert!(matches!(media.write_frame(frame), Err(MoqError::Closed))); + assert!(matches!(media.finish(), Err(MoqError::Closed))); +} + +/// A track handle released without a finish, as a garbage collector does, ends its track as a +/// publisher leaving rather than a cause-less drop. +#[tokio::test(start_paused = true)] +async fn dropping_an_unfinished_track_handle_ends_it() { + let broadcast = MoqBroadcastProducer::new().unwrap(); + let (raw, media, readers) = children(&broadcast).await; + drop((raw, media)); + assert_left(readers).await; + broadcast.close().unwrap(); +} + +/// A track finished before its broadcast closes ends cleanly, even once its handle is released. +#[tokio::test(start_paused = true)] +async fn finishing_before_close_ends_tracks_cleanly() { + let broadcast = MoqBroadcastProducer::new().unwrap(); + let (raw, media, readers) = children(&broadcast).await; + raw.finish().unwrap(); + media.finish().unwrap(); + drop(raw); + broadcast.close().unwrap(); + for reader in readers { + let read = tokio::time::timeout(TIMEOUT, reader.read_frame()).await; + assert!(read.expect("the track carried on").unwrap().is_none()); + } +} + +/// One writer of each data kind on `broadcast`, each with a subscriber reading its track. +async fn data_children( + broadcast: &MoqBroadcastProducer, +) -> ( + ( + Arc, + Arc, + Arc, + Arc, + Arc, + ), + [Arc; 5], +) { + let consumer = broadcast.consume().unwrap(); + let flate = || MoqFlateConfig { + compression: false, + mime: None, + }; + let track = |name: &str| broadcast.publish_track(name.into(), None).unwrap(); + let writers = ( + MoqJsonSnapshotProducer::new( + broadcast, + &track("json-snapshot"), + MoqJsonSnapshotConfig { + deltas: true, + compression: false, + }, + ) + .unwrap(), + MoqJsonStreamProducer::new( + broadcast, + &track("json-stream"), + MoqJsonStreamConfig { compression: false }, + ) + .unwrap(), + MoqJsonWindowProducer::new( + &track("json-window"), + MoqJsonWindowConfig { + compression: false, + checkpoint_records: None, + }, + ) + .unwrap(), + MoqFlateSnapshotProducer::new(broadcast, &track("flate-snapshot"), flate()).unwrap(), + MoqFlateStreamProducer::new(broadcast, &track("flate-stream"), flate()).unwrap(), + ); + let readers = [ + consumer.subscribe_track("json-snapshot".into(), None).await.unwrap(), + consumer.subscribe_track("json-stream".into(), None).await.unwrap(), + consumer.subscribe_track("json-window".into(), None).await.unwrap(), + consumer.subscribe_track("flate-snapshot".into(), None).await.unwrap(), + consumer.subscribe_track("flate-stream".into(), None).await.unwrap(), + ]; + (writers, readers) +} + +/// Closing a broadcast ends its data tracks the way it ends media ones, and their writers +/// refuse later calls. +#[tokio::test(start_paused = true)] +async fn closing_a_broadcast_ends_its_open_data_tracks() { + let broadcast = MoqBroadcastProducer::new().unwrap(); + let ((json_snapshot, json_stream, json_window, flate_snapshot, flate_stream), readers) = + data_children(&broadcast).await; + broadcast.close().unwrap(); + assert_left(readers).await; + + assert!(matches!(json_snapshot.update("1".into(), None), Err(MoqError::Closed))); + assert!(matches!(json_stream.append("1".into(), None), Err(MoqError::Closed))); + assert!(matches!(json_window.push("1".into(), None), Err(MoqError::Closed))); + assert!(matches!(flate_snapshot.update(vec![1], None), Err(MoqError::Closed))); + assert!(matches!(flate_stream.append(vec![1], None), Err(MoqError::Closed))); +} + +/// A data writer released without a finish ends its track as a publisher leaving. +#[tokio::test(start_paused = true)] +async fn dropping_an_unfinished_data_writer_ends_it() { + let broadcast = MoqBroadcastProducer::new().unwrap(); + let (writers, readers) = data_children(&broadcast).await; + drop(writers); + assert_left(readers).await; + broadcast.close().unwrap(); +} + +/// A track handle goes to one writer: while one is built from it, another is refused, and a +/// construction that fails hands the handle back. +#[tokio::test(start_paused = true)] +async fn a_track_is_adopted_by_one_writer() { + let broadcast = MoqBroadcastProducer::new().unwrap(); + let track = broadcast.publish_track("json".into(), None).unwrap(); + let config = || MoqJsonSnapshotConfig { + deltas: false, + compression: false, + }; + let failed = track.adopt::(|_| { + assert!(matches!( + MoqJsonSnapshotProducer::new(&broadcast, &track, config()), + Err(MoqError::Closed) + )); + Err(MoqError::JsonTrack { + reason: "refused".into(), + }) + }); + assert!(matches!(failed, Err(MoqError::JsonTrack { .. }))); + let writer = MoqJsonSnapshotProducer::new(&broadcast, &track, config()).unwrap(); + assert!(matches!( + MoqJsonSnapshotProducer::new(&broadcast, &track, config()), + Err(MoqError::Closed) + )); + drop(writer); + broadcast.close().unwrap(); +} + +/// A media finish that fails ends the track with that failure, rather than dropping it without a +/// cause. +#[tokio::test(start_paused = true)] +async fn a_failed_finish_ends_the_track_with_its_cause() { + let broadcast = MoqBroadcastProducer::new().unwrap(); + let consumer = broadcast.consume().unwrap(); + let stream = MoqMediaTrackStreamProducer::video( + &broadcast, + MoqMediaTarget::Named { + name: Some("video".into()), + }, + video_init(MoqVideoFormat::Avc3, h264_init()), + ) + .unwrap(); + let reader = consumer.subscribe_track("video".into(), None).await.unwrap(); + + // A delta slice with no keyframe before it, held by the splitter until the finish flushes it. + stream.write(vec![0, 0, 0, 1, 0x61, 0xe0, 0x12, 0x34]).unwrap(); + assert!(matches!(stream.finish(), Err(MoqError::Codec { .. }))); + + let read = tokio::time::timeout(TIMEOUT, reader.read_frame()).await; + let Err(err) = read.expect("the track carried on") else { + panic!("the track ended without a cause"); + }; + assert_protocol( + &err, + crate::error::MoqErrorScope::Stream, + crate::error::MoqProtocolKind::Internal, + ); + broadcast.close().unwrap(); +} + +/// A request accepted after its broadcast closed is refused, and its subscriber sees the +/// publisher leave. +#[tokio::test(start_paused = true)] +async fn accepting_a_request_after_close_is_refused() { + let broadcast = MoqBroadcastProducer::new().unwrap(); + let dynamic = broadcast.dynamic().unwrap(); + let consumer = broadcast.consume().unwrap(); + let subscribe = tokio::spawn(async move { consumer.subscribe_track("late".into(), None).await }); + let request = dynamic.requested_track().await.unwrap(); + broadcast.close().unwrap(); + assert!(matches!(request.accept(None), Err(MoqError::Closed))); + let reader = subscribe.await.unwrap().unwrap(); + assert_left([reader]).await; +} + #[tokio::test] async fn announced_broadcast() { let origin = MoqOriginProducer::new(MoqOriginConfig::default()); diff --git a/rs/moq-ffi/src/video.rs b/rs/moq-ffi/src/video.rs index b69b9762346..489a90f4937 100644 --- a/rs/moq-ffi/src/video.rs +++ b/rs/moq-ffi/src/video.rs @@ -305,12 +305,12 @@ impl MoqVideoEncoder { fn finalize( mut producer: moq_video::encode::Producer, drained: Result<(), moq_video::Error>, - finished: &crate::producer::Finished, + children: &crate::producer::Children, ) -> Result<(), MoqError> { match drained { Ok(()) => { producer.finish()?; - finished.keep(producer); + children.keep(producer); Ok(()) } Err(err) => { @@ -408,14 +408,14 @@ impl VideoProducer { Ok(()) } - fn finish(self, finished: &crate::producer::Finished) -> Result<(), MoqError> { + fn finish(self, children: &crate::producer::Children) -> Result<(), MoqError> { let Self { encoder, mut producer, .. } = self; // Drain the codec into the track before ending it, so the last frames land // in it rather than being dropped with the encoder. let drained = block_on(encoder.finish()).and_then(|encoded| producer.publish(&encoded)); - finalize(producer, drained, finished) + finalize(producer, drained, children) } } @@ -424,6 +424,9 @@ impl VideoProducer { /// Constructed from a broadcast's media catalog. Each /// [`write`](Self::write) accepts a [`MoqVideoFrame`] whose `data` is in the /// pixel format declared by the [`MoqVideoEncoderInput`] passed at publish time. +/// +/// Released before its finish, or still open when its broadcast closes, it ends its track with +/// `Unroutable`, which readers take as the publisher leaving. #[derive(uniffi::Object)] pub struct MoqVideoProducer { inner: Arc>>, @@ -433,7 +436,33 @@ pub struct MoqVideoProducer { applied: Arc, /// Held so the reservation's registry outlives extra bandwidth handles. _bandwidth: Option>, - finished: crate::producer::Finished, + children: crate::producer::Children, +} + +// Dropped or closed before its finish, the track ends the way a publisher leaving does. +impl crate::producer::Child for MoqVideoProducer { + fn abort(&self, err: moq_net::Error) { + let _guard = crate::ffi::runtime().enter(); + let Some(producer) = self + .inner + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .take() + else { + return; + }; + if let Some(mut follow) = self.follow.lock().unwrap().take() { + let _ = follow.stop(); + } + self.reservation.lock().unwrap().take(); + producer.producer.abort(err); + } +} + +impl Drop for MoqVideoProducer { + fn drop(&mut self) { + crate::producer::Child::abort(self, moq_net::Error::Unroutable); + } } impl MoqVideoProducer { @@ -548,14 +577,9 @@ impl MoqVideoProducer { /// The broadcast keeps the finished track until it ends, so its finish still holds it. pub fn finish(&self) -> Result<(), MoqError> { let _guard = crate::ffi::runtime().enter(); - let producer = self - .inner - .lock() - .map_err(|_| panic_error())? - .take() - .ok_or(MoqError::Closed)?; // Stop following and release the share before draining, so siblings can - // take the room while the last frames go out. + // take the room while the last frames go out. The follower takes the + // producer's lock, so it stops before the drain holds it. let followed = self .follow .lock() @@ -563,9 +587,13 @@ impl MoqVideoProducer { .take() .map_or(Ok(()), |mut follow| follow.stop()); self.reservation.lock().unwrap().take(); + // Held through the drain, so a racing broadcast close waits it out instead of + // missing a track that still publishes. + let mut inner = self.inner.lock().map_err(|_| panic_error())?; + let producer = inner.take().ok_or(MoqError::Closed)?; // A follower panic only stops rate adaptation, so the encoded frames still // end the track cleanly before the panic is reported. - producer.finish(&self.finished)?; + producer.finish(&self.children)?; followed } } @@ -608,7 +636,7 @@ impl MoqVideoProducer { let encoder = block_on(moq_video::encode::Sink::open(&config))?; let gop = config.gop.duration(config.framerate); let ceiling = video_ceiling(&config, &rendition); - let finished = broadcast.with_state(|state| Ok(state.finished.clone()))?; + let children = broadcast.with_state(|state| Ok(state.children.clone()))?; let producer = broadcast.with_state(|state| match track { Some(name) => { let track = state @@ -649,13 +677,13 @@ impl MoqVideoProducer { .as_ref() .map(|reservation| spawn_follow(inner.clone(), reservation.consumer(), ceiling, applied.clone())); - Ok(Arc::new(MoqVideoProducer { + children.adopt(Arc::new(MoqVideoProducer { inner, reservation: std::sync::Mutex::new(reserved), follow: std::sync::Mutex::new(follow), applied, _bandwidth: bandwidth, - finished, + children: children.clone(), })) } } diff --git a/rs/moq-flate/src/snapshot/producer.rs b/rs/moq-flate/src/snapshot/producer.rs index 5515d4df39a..7d6194c0749 100644 --- a/rs/moq-flate/src/snapshot/producer.rs +++ b/rs/moq-flate/src/snapshot/producer.rs @@ -58,6 +58,17 @@ impl Producer { pub fn finish(&mut self) -> Result<()> { self.inner.lock().unwrap().finish() } + + /// Abort the track with the given error, which consumers see in place of any further value. + /// + /// Consumes the handle, since nothing can be published afterwards. Clones sharing the track see + /// it closed: their next [`update`](Self::update) fails. + pub fn abort(self, err: moq_net::Error) -> Result<()> { + // Through a clone, since aborting consumes a handle and the state is shared. Idempotent: a + // track that is already closed keeps its first reason. + let _ = self.inner.lock().unwrap().track.clone().abort(err); + Ok(()) + } } /// Shared publishing state behind [`Producer`]'s `Arc`. diff --git a/rs/moq-flate/src/stream/producer.rs b/rs/moq-flate/src/stream/producer.rs index 678a0035c5a..0c1376e31b7 100644 --- a/rs/moq-flate/src/stream/producer.rs +++ b/rs/moq-flate/src/stream/producer.rs @@ -72,6 +72,15 @@ impl Producer { pub fn finish(&mut self) -> Result<()> { self.inner.lock().unwrap().finish() } + + /// Abort the track with the given error, which consumers see in place of a clean end. + /// + /// Consumes the handle, since nothing can be appended afterwards. Clones sharing the track see + /// it closed: their next [`append`](Self::append) fails. + pub fn abort(self, err: moq_net::Error) -> Result<()> { + self.inner.lock().unwrap().abort(err); + Ok(()) + } } /// Shared publishing state behind [`Producer`]'s `Arc`. diff --git a/rs/moq-json/src/stream/producer.rs b/rs/moq-json/src/stream/producer.rs index 305df0f888d..ec48c004f44 100644 --- a/rs/moq-json/src/stream/producer.rs +++ b/rs/moq-json/src/stream/producer.rs @@ -92,6 +92,15 @@ impl Producer { pub fn finish(&mut self) -> Result<()> { self.inner.lock().unwrap().finish() } + + /// Abort the track with the given error, which consumers see in place of a clean end. + /// + /// Consumes the handle, since nothing can be appended afterwards. Clones sharing the track see + /// it closed: their next [`append`](Self::append) fails. + pub fn abort(self, err: moq_net::Error) -> Result<()> { + self.inner.lock().unwrap().track.abort(err); + Ok(()) + } } /// Shared publishing state behind [`Producer`]'s `Arc`. diff --git a/rs/moq-json/src/window/producer.rs b/rs/moq-json/src/window/producer.rs index 8a086a60382..5c1e5b82c94 100644 --- a/rs/moq-json/src/window/producer.rs +++ b/rs/moq-json/src/window/producer.rs @@ -114,6 +114,15 @@ impl Producer { pub fn finish(&mut self) -> Result<()> { self.inner.lock().unwrap().finish() } + + /// Abort the track with the given error, which consumers see in place of a clean end. + /// + /// Consumes the handle, since nothing can be published afterwards. Clones sharing the track see + /// it closed: their next edit fails with [`moq_net::Error::Closed`]. + pub fn abort(self, err: moq_net::Error) -> Result<()> { + self.inner.lock().unwrap().abort(err); + Ok(()) + } } impl Producer { @@ -194,6 +203,16 @@ impl Inner { self.track.finish() } + fn abort(&mut self, err: moq_net::Error) { + self.finished = true; + // The group first: aborting the track leaves an already-pulled group consumer independent. + if let Some(group) = self.track.group.take() { + let _ = group.abort(err.clone()); + } + // Idempotent: a track that is already closed keeps its first reason. + let _ = self.track.inner.clone().abort(err); + } + fn ensure_open(&self) -> Result<()> { if self.finished { return Err(moq_net::Error::Closed.into()); diff --git a/rs/moq-mux/src/binary.rs b/rs/moq-mux/src/binary.rs index 54aa92d1109..35561654795 100644 --- a/rs/moq-mux/src/binary.rs +++ b/rs/moq-mux/src/binary.rs @@ -191,6 +191,12 @@ impl Snapshot { pub fn finish(mut self) -> crate::Result<()> { Ok(self.inner.finish()?) } + + /// Abort the track with `err`, which consumers see in place of a clean end, and retire its + /// catalog entry. + pub fn abort(self, err: moq_net::Error) -> crate::Result<()> { + Ok(self.inner.abort(err)?) + } } /// Publishes an append-log binary track, advertised in the catalog for as long as this handle lives. @@ -280,6 +286,12 @@ impl Stream { pub fn finish(mut self) -> crate::Result<()> { Ok(self.inner.finish()?) } + + /// Abort the track with `err`, which consumers see in place of a clean end, and retire its + /// catalog entry. + pub fn abort(self, err: moq_net::Error) -> crate::Result<()> { + Ok(self.inner.abort(err)?) + } } /// Reads a binary track, in whichever mode its catalog entry declares. diff --git a/rs/moq-mux/src/json.rs b/rs/moq-mux/src/json.rs index 1eafafe8587..98f44d19f7c 100644 --- a/rs/moq-mux/src/json.rs +++ b/rs/moq-mux/src/json.rs @@ -247,6 +247,12 @@ impl Snapshot { pub fn finish(mut self) -> crate::Result<()> { Ok(self.inner.finish()?) } + + /// Abort the track with `err`, which consumers see in place of a clean end, and retire its + /// catalog entry. + pub fn abort(self, err: moq_net::Error) -> crate::Result<()> { + Ok(self.inner.abort(err)?) + } } /// Publishes an append-log JSON track, advertised in the catalog for as long as this handle lives. @@ -343,6 +349,12 @@ impl Stream { pub fn finish(mut self) -> crate::Result<()> { Ok(self.inner.finish()?) } + + /// Abort the track with `err`, which consumers see in place of a clean end, and retire its + /// catalog entry. + pub fn abort(self, err: moq_net::Error) -> crate::Result<()> { + Ok(self.inner.abort(err)?) + } } /// Reads a JSON track, in whichever mode its catalog entry declares. diff --git a/swift/Sources/Moq/Broadcast.swift b/swift/Sources/Moq/Broadcast.swift index 9f963e88303..c04dd1a3c8c 100644 --- a/swift/Sources/Moq/Broadcast.swift +++ b/swift/Sources/Moq/Broadcast.swift @@ -95,7 +95,9 @@ public final class BroadcastProducer: Sendable { /// End the broadcast for good: retract it and serve no new tracks. /// - /// Tracks already subscribed carry on to their own end. Closing again is a no-op. + /// Tracks still open on it end with `Unroutable`, which readers take as the publisher + /// leaving, and their later calls throw `MoqError.Closed`; finish a track first to end it + /// cleanly. Closing again is a no-op. public func close() throws { try ffi.close() }