diff --git a/dart/moq/lib/src/client.dart b/dart/moq/lib/src/client.dart index 146f818c87d..7c24a23f724 100644 --- a/dart/moq/lib/src/client.dart +++ b/dart/moq/lib/src/client.dart @@ -228,7 +228,8 @@ final class Moq { AnnounceOptions options = const AnnounceOptions(), }) => session.consume().announced(config: options._ffi); - /// Wait for a broadcast announced at exactly [path]. + /// Follow the broadcast at exactly [path]: each `available()` resolves the + /// next publisher instance serving it, such as a restarted publisher. AnnouncedBroadcast announcedBroadcast(String path) => session.consume().announcedBroadcast(path: path); diff --git a/dart/moq_ffi/lib/src/moq.dart b/dart/moq_ffi/lib/src/moq.dart index 8a6c75df9f7..e40b0114252 100644 --- a/dart/moq_ffi/lib/src/moq.dart +++ b/dart/moq_ffi/lib/src/moq.dart @@ -8,9 +8,7 @@ import "dart:ffi"; import "dart:io" show Platform, File, Directory; import "dart:isolate"; import "dart:typed_data"; - import "package:ffi/ffi.dart"; - import "uniffi_runtime.dart"; export "uniffi_runtime.dart"; @@ -18188,7 +18186,7 @@ void _checkApiChecksums() { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqannouncedbroadcast_available() != - 17719) { + 47724) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqannouncedbroadcast_cancel() != 63175) { @@ -18207,7 +18205,7 @@ void _checkApiChecksums() { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqoriginconsumer_announced_broadcast() != - 8509) { + 56171) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqoriginconsumer_request_broadcast() != diff --git a/doc/lib/cpp/index.md b/doc/lib/cpp/index.md index 86b10779410..fbc267fcc72 100644 --- a/doc/lib/cpp/index.md +++ b/doc/lib/cpp/index.md @@ -112,6 +112,10 @@ if (auto catalog = ok(catalogs->next().get())) { // std::nullopt once the cata } ``` +Call `available()` again to wait for the next publisher instance at that path, such as a +restarted publisher, and drop what was read from the previous one then: its tracks end +when its session does. + `MediaTrackConsumer` reads in decode order. For arrival-order reads, decode each group from `TrackConsumer::recv_group` with `moq::MediaGroupConsumer::init(group, container)`; frames carry container media time as they land. diff --git a/doc/setup/upgrade.md b/doc/setup/upgrade.md index 806bb5a1fa1..86b09986825 100644 --- a/doc/setup/upgrade.md +++ b/doc/setup/upgrade.md @@ -467,6 +467,12 @@ These land with the next breaking release, not the 2026-09-23 train. decode time (RTMP, a browser) trails by the delay plus the video's send-ahead: 2 s at the new default, where `--delay 500ms` keeps the 1 s it had. moq-srt egress runs at its SRT latency or 1 s, whichever is longer. +- **`moq_origin_announced_broadcast` delivers every publisher instance.** Its + callback gets a new broadcast handle each time another instance takes over + the path or the broadcast returns, and its terminal `0` only on cancel. Free + each handle with `moq_consume_close` once its successor arrives. A binding's + `AnnouncedBroadcast.available()` works the same way, and its first call is + unchanged. ## Wire diff --git a/go/wrapper/client.go b/go/wrapper/client.go index 288eb68b663..596ef3bd49e 100644 --- a/go/wrapper/client.go +++ b/go/wrapper/client.go @@ -367,8 +367,8 @@ func (c *Client) Announced(options AnnounceOptions) (*AnnounceConsumer, error) { return c.consumer.Announced(options) } -// AnnouncedBroadcast waits for a route covering path, then resolves the -// broadcast there. +// AnnouncedBroadcast follows path, resolving each publisher instance that serves +// it. func (c *Client) AnnouncedBroadcast(path string) (*AnnouncedBroadcast, error) { return c.consumer.AnnouncedBroadcast(path) } diff --git a/go/wrapper/origin.go b/go/wrapper/origin.go index 73bd7815be9..088e539b6be 100644 --- a/go/wrapper/origin.go +++ b/go/wrapper/origin.go @@ -147,8 +147,8 @@ func (o *OriginConsumer) Announced(options AnnounceOptions) (*AnnounceConsumer, return &AnnounceConsumer{inner: inner}, nil } -// AnnouncedBroadcast waits for a route covering an exact path, then resolves -// the broadcast there. +// AnnouncedBroadcast follows an exact path, resolving each publisher instance +// that serves it. func (o *OriginConsumer) AnnouncedBroadcast(path string) (*AnnouncedBroadcast, error) { inner, err := o.inner.AnnouncedBroadcast(path) if err != nil { @@ -195,12 +195,17 @@ func (a *AnnounceConsumer) Cancel() { a.inner.Cancel() } -// AnnouncedBroadcast awaits a specific broadcast becoming available. +// AnnouncedBroadcast follows a specific broadcast across its publisher instances. type AnnouncedBroadcast struct { inner *ffi.MoqAnnouncedBroadcast } -// Available blocks until the broadcast is available and returns its consumer. +// Available blocks until the next publisher instance serves the broadcast and +// returns its consumer. The first call resolves the instance serving the path once +// one is announced. Each later call waits for another to take over (a restarted +// publisher, or a replacement session) or for the broadcast to return after it went +// offline. Drop what was read from the previous instance then: its tracks end when +// its session does. func (a *AnnouncedBroadcast) Available(ctx context.Context) (*BroadcastConsumer, error) { inner, err := bridge.CallHandle(ctx, a.inner.Cancel, a.inner.Available) if err != nil { diff --git a/js/net/src/goaway-requests.test.ts b/js/net/src/goaway-requests.test.ts index f68978ef047..1e98339604e 100644 --- a/js/net/src/goaway-requests.test.ts +++ b/js/net/src/goaway-requests.test.ts @@ -283,3 +283,109 @@ test("announcing and subscribing after GOAWAY still open streams, on the adapter subscriber.close(); } }); + +/** + * A route change alone is never fatal, but a dead route is: once the replacement wins, the + * subscription already open on the draining session keeps going while the old publisher + * serves it, and that session closing ends the subscription with an error, never cleanly. + * The old session closes either before the `restart` reaches the viewer or after. + */ +async function routeDeath(kind: "lite" | "ietf", after: boolean): Promise { + const protocol = kind === "lite" ? ALPN_06 : ALPN.DRAFT_17; + const pair = createMockTransportPair(protocol); + const inject = kind === "ietf" ? captureFirstUni(pair.server) : undefined; + const goawayBytes = kind === "ietf" ? await encodeIetfGoaway() : undefined; + + const consume = new OriginProducer(); + const publish = new OriginProducer(); + const broadcast = publish.createBroadcast(Path.from("room")); + broadcast.announce(); + + const served: TrackProducer[] = []; + const serving = (async () => { + for (;;) { + const req = await wireOf(broadcast).requested(); + if (!req) break; + const track = req.accept({ timescale: Timescale.MILLI }); + track.writeFrame(textFrame("one")); + served.push(track); + } + })(); + + const [client, server] = await Promise.all([ + connect({ url, transport: pair.client, consume }), + accept({ transport: pair.server, url, publish: publish.consume() }), + ]); + + const watching = consume.announced(); + const request = consume.request(Path.from("room")); + let routes: ReturnType | undefined; + let replacement: Awaited> | undefined; + let replacementServer: Awaited> | undefined; + + try { + expect((await watching.next())?.kind).toBe("start"); + await waitUntil(() => request.active.peek() !== undefined); + const front = request.active.peek(); + if (!front) throw new Error("no route"); + const video = front.track("video").subscribe().ordered(); + expect(await video.readString()).toBe("one"); + + if (inject && goawayBytes) await inject.inject(goawayBytes); + else await sendLiteGoaway(pair.server); + await wireOf(client).goaway; + + routes = consume.announced(); + expect((await routes.next())?.kind).toBe("start"); + const pair2 = createMockTransportPair(protocol); + [replacement, replacementServer] = await Promise.all([ + connect({ url, transport: pair2.client, consume }), + accept({ transport: pair2.server, url, publish: publish.consume() }), + ]); + if (after) { + // Skip the drain cost reaching the route, if it lands after the start. + let event = await routes.next(); + while (event?.kind === "update") event = await routes.next(); + expect(event).toMatchObject({ prefix: Path.from("room"), kind: "restart" }); + } + + // The old publisher still serves the old instance, so its subscription carries on. + for (const producer of served) { + if (producer.closed.peek() === undefined) producer.writeFrame(textFrame("two")); + } + expect(await video.readString()).toBe("two"); + + // A draining close of a session still serving a live track gives up after its drain. + await server.close().catch(() => {}); + await expect( + (async () => { + for (;;) { + if ((await video.readString()) === undefined) return "ended cleanly"; + } + })(), + ).rejects.toThrow(); + video.close(); + } finally { + watching.close(); + routes?.close(); + request.close(); + replacement?.abort(); + replacementServer?.abort(); + client.abort(); + server.abort(); + broadcast.close(); + await serving; + publish.close(); + consume.close(); + } +} + +for (const kind of ["lite", "ietf"] as const) { + test(`${kind}: the draining session closing before the restart errors its subscription`, async () => { + await routeDeath(kind, false); + }); + + test(`${kind}: the draining session closing after the restart errors its subscription`, async () => { + await routeDeath(kind, true); + }); +} diff --git a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Moq.kt b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Moq.kt index e8bf96cd90a..21d7628e621 100644 --- a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Moq.kt +++ b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Moq.kt @@ -33,10 +33,11 @@ class Moq internal constructor( session.consume().announced(config) /** - * Await a route covering exactly [path], then resolve the broadcast there. + * Follow exactly [path], resolving each publisher instance that serves it. * * Unlike [requestBroadcast] this waits indefinitely for a future - * announcement. Cancel the returned handle to stop waiting. + * announcement, and each later `available()` waits for the next instance, + * such as a restarted publisher. Cancel the returned handle to stop waiting. */ fun announcedBroadcast(path: String): AnnouncedBroadcast = session.consume().announcedBroadcast(path) diff --git a/py/moq-rs/moq/client.py b/py/moq-rs/moq/client.py index 7ec963ad4c4..ce913c44f84 100644 --- a/py/moq-rs/moq/client.py +++ b/py/moq-rs/moq/client.py @@ -197,7 +197,7 @@ def announced(self, prefix: str = "", *, filter: str | None = None, hidden: bool return self._require_consumer().announced(prefix, filter=filter, hidden=hidden) def announced_broadcast(self, path: str) -> AnnouncedBroadcast: - """Await announcement of the broadcast at exactly ``path``. + """Follow the broadcast at exactly ``path`` across its publisher instances. See :meth:`OriginConsumer.announced_broadcast`. """ diff --git a/py/moq-rs/moq/origin.py b/py/moq-rs/moq/origin.py index c5bdd575d6b..00e83ba2e18 100644 --- a/py/moq-rs/moq/origin.py +++ b/py/moq-rs/moq/origin.py @@ -81,10 +81,11 @@ def cancel(self) -> None: class AnnouncedBroadcast: - """Awaitable that resolves once a route covers a specific path. + """Follows a specific path across the publisher instances that serve it. ``await`` it (or call :meth:`available`) to get the :class:`BroadcastConsumer` - once the broadcast becomes available. Usable as an async context manager. + once the broadcast becomes available, and again for each instance that takes over. + Usable as an async context manager. """ def __init__(self, inner: MoqAnnouncedBroadcast) -> None: @@ -97,7 +98,13 @@ async def __aexit__(self, *exc) -> None: self.cancel() async def available(self) -> BroadcastConsumer: - """Await a covering route and return the resolved broadcast consumer.""" + """Await the next publisher instance serving the path and return its consumer. + + The first call resolves the instance serving the path once one is announced. Each + later call waits for another to take over (a restarted publisher, or a replacement + session) or for the broadcast to return after it went offline. Drop what was read + from the previous instance then: its tracks end when its session does. + """ return BroadcastConsumer(await self._inner.available()) def __await__(self): @@ -181,7 +188,7 @@ def announced(self, prefix: str = "", *, filter: str | None = None, hidden: bool return AnnounceConsumer(self._inner.announced(MoqAnnounceConfig(prefix=prefix, filter=filter, hidden=hidden))) def announced_broadcast(self, path: str) -> AnnouncedBroadcast: - """Await the broadcast at ``path``, resolving once something can serve it. + """Follow the broadcast at ``path``, resolving each publisher instance that serves it. A broadcast created on this origin resolves once it is announced, like a remote one. diff --git a/quest/m0/broadcast-epoch/catalog-references.md b/quest/m0/broadcast-epoch/catalog-references.md index f5999c8feec..4e6a259a6da 100644 --- a/quest/m0/broadcast-epoch/catalog-references.md +++ b/quest/m0/broadcast-epoch/catalog-references.md @@ -56,7 +56,7 @@ following does not authorize substituting a different requested epoch. restarts its pump. - Exporters: a late first announcement starts the rendition and End ends it. Restart behavior belongs to - [restart followers](/quest/m1/restart-followers.md). + [sink restart follow](/quest/m1/sink-restart-follow.md). - Share this in `moq_mux` (a `Source::follow`-style API; the shape is open) rather than per reader. `Source::bind` binds once today, as `failed_binding_does_not_retry_a_later_publisher` asserts; that changes. diff --git a/quest/m1/README.md b/quest/m1/README.md index be9c7ab5c2a..e0af98c2417 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -52,7 +52,8 @@ lands before the cut. - [qmux credit](/quest/m0/qmux-credit.md) - dropped/stopped streams return credit and close is delivered - [Per-package MIGRATE.md](/quest/m1/migrate.md) - every published package documents how to migrate each breaking change, and breaking PRs add their entry -- [Restart followers](/quest/m1/restart-followers.md) - first-party consumers follow an epochless `Restart`, from a GOAWAY drain or a publisher replacement, by resolving the path again +- [Sink restart follow](/quest/m1/sink-restart-follow.md) - moq-cli's server and push sinks follow an epochless `Restart` where their output can mark it, and fail loud otherwise +- [OBS restart follow](/quest/m1/obs-restart-follow.md) - the OBS source keeps playing across a GOAWAY drain or a publisher replacement - [Cluster routing](/quest/m1/cluster-routing/README.md) - prove shared routing contracts, implement per-origin routes, and connect outbound local clusters - [Routing guide](/quest/m1/routing-guide.md) - concepts and practical outbound federation setup - [Controlled time](/quest/m1/time/README.md) - every crate and package reads time through moq-clock or @moq/clock, with regression tests for the changed contracts diff --git a/quest/m1/obs-restart-follow.md b/quest/m1/obs-restart-follow.md new file mode 100644 index 00000000000..fe98a8f1123 --- /dev/null +++ b/quest/m1/obs-restart-follow.md @@ -0,0 +1,22 @@ +# [S] The OBS source follows a broadcast's restarts + +## Goal + +The OBS MoQ source keeps playing across an epochless GOAWAY drain or a +publisher replacement. When `AnnouncedBroadcast::available()` hands out the +next publisher instance, the source drops its catalog and track readers for +the old one and subscribes to the new one, without blanking for longer than +the new instance takes to deliver a keyframe. + +## Plan + +`cpp/obs/src/moq-source.cpp` calls `available()` once (`moq_source_on_connect`) +and stops when its catalog ends, so a replaced instance leaves the source +stopped once the old route dies and its tracks end with an error. Since moq-ffi's `available()` now +resolves each instance in turn, re-arm it after every instance and rebuild the +`Connection`'s consumer, catalog and tracks on each one, reusing the decoders +when the codec config is unchanged. + +Cover it in `cpp/obs/test` with a replacement while the source plays. + +Public API: none. Wire: none. diff --git a/quest/m1/restart-followers.md b/quest/m1/restart-followers.md deleted file mode 100644 index a3947fa899c..00000000000 --- a/quest/m1/restart-followers.md +++ /dev/null @@ -1,62 +0,0 @@ -# [L] First-party consumers follow an epochless Restart - -## Goal - -Every first-party consumer that resolves a broadcast and keeps reading holds -the announce stream for its path, and on `Restart` drops its handles and -resolves the path again. On `Unroutable` it fails; it never retries a -request. A GOAWAY drain or a publisher replacement on an epochless route -then costs a long-lived consumer nothing but the restart. - -## Plan - -Moved to m1 on 2026-10-13 (maintainer): the WebSocket-to-QUIC upgrade is -removed, so an epochless Restart no longer follows -every connect; it remains for GOAWAY drains and publisher replacement. - -The origin side is done: a handle resolved through a peer keeps serving new -requests while the peer's epochless replacement is held, and refuses them -with `Unroutable` only once the `Restart` is deliverable (`Release` in `rs/moq-net/src/model/origin.rs`, -`rs/moq-net/tests/integration/replacement.rs`). So a consumer that never follows `Restart` -works through the hold (300 ms by default), and its open subscriptions stay -sticky on the old session until it closes, where they end cleanly. A -consumer that opens a request after the hold, or reads past the old -session's close, needs this quest. - -Audit and fix: - -- Rust: `moq export` and `moq subscribe` (the interop subscriber), and any - other moq-cli command that resolves a broadcast and keeps reading. The TS - export's `Follower` (`rs/moq-mux/src/container/ts/follower.rs`) already - watches the path, but treats any epochless return as a replacement: - sticky until the old instance ends, then `Error::Replaced`, unless - stitching. The fmp4, mkv, h264, h265, and flv exports don't follow at all. - Decide whether an epochless `Restart` restarts the output (a new init - segment mid-stream), stitches like TS, or ends the export. - `moq play` already follows `Restart` (`rs/moq-cli/src/play/media.rs`). -- Bindings: moq-ffi's catalog and container consumers, as Go, C++, and - Python reach them, and moq-c's (`moq_origin_announced_broadcast`, - `moq_consume_catalog`). Prefer following once in moq-ffi and moq-c over - duplicating it in every language wrapper. New API goes through the - Cross-Package Sync table. -- JS: `js/net`'s origin owns restart following (`js/watch/src/broadcast.ts` - mirrors its current handle). Confirm the js-native interop subscriber, - which requests with `announced: true`, follows a Rust relay's `Restart`. -- The interop clients under `test/interop/clients`, where they resolve a - broadcast themselves. - -Regression: extend `rs/moq-net/tests/integration/replacement.rs` (or the consumer's own -tests) so each follower reads past the old session's close after an -epochless replacement. Then run `just test interop --all`. - -If a failing client still leaves its connection to idle out, fix that -client's error path to close its session, as a separate commit. The C -client's `fail()` (`_exit` without `moq_session_close`) and the C++ client -on a consumer error both look suspect. - -Public API: consumer-follow API in moq-ffi and moq-c, if the audit needs it. -Wire: none. - -## Related - -- [Broadcast epochs](/quest/m0/broadcast-epoch/README.md) - the players that already follow `Restart` diff --git a/quest/m1/sink-restart-follow.md b/quest/m1/sink-restart-follow.md new file mode 100644 index 00000000000..06746a63e75 --- /dev/null +++ b/quest/m1/sink-restart-follow.md @@ -0,0 +1,36 @@ +# [M] Server and push sinks handle a broadcast's restarts + +## Goal + +Every moq-cli sink that resolves a broadcast and keeps serving it handles an +epochless `Restart` on purpose: it follows the new publisher instance where +its output can mark the break. A route change alone is never fatal: a sink +that cannot follow stays on the old instance while its publisher serves it, +and fails once that route dies, since its subscriptions then end with an +error. + +## Plan + +The stdout exports already do this: TS follows with `--stitch`, and the other +formats ride the old instance until its route dies and its tracks error. Audit +and fix the rest the same way: + +- `export hls`: `moq-hls`'s server resolves per request; check that a restart + moves new requests to the new instance and how a playlist marks the break + (`EXT-X-DISCONTINUITY`). Coordinate with + [HLS export failure signal](/quest/m1/hls-failure-signal.md). +- `export rtmp`, `export srt`, `export rtc`, both `--listen` (per-viewer + streams in `moq-rtmp`, `moq-srt`, `moq-rtc`) and `--connect` (one push). + SRT already rides the TS follower (`--linger`, `--stitch`); confirm its + listen side does too. FLV and WebRTC need a decision. +- `export archive`: a restart already fails a continued store under a new + epoch; decide what a mid-recording restart does. +- `transcode`: its source broadcast; following means restarting the ladder + on the new instance. + +Share one helper (a `moq_mux` `Source::follow`-style API per +[catalog references](/quest/m0/broadcast-epoch/catalog-references.md)) +rather than one rule per sink. Each fix lands with a test that replaces the +publisher while the sink runs. + +Public API: possibly a shared follow helper in `moq-mux`. Wire: none. diff --git a/rs/moq-c/src/api.rs b/rs/moq-c/src/api.rs index 650018fe0c4..0e11668680c 100644 --- a/rs/moq-c/src/api.rs +++ b/rs/moq-c/src/api.rs @@ -1841,7 +1841,7 @@ pub extern "C" fn moq_origin_announced_cancel(announced: u32) -> i32 { }) } -/// Consume a broadcast from an origin by path, waiting until something can serve it. +/// Follow a broadcast on an origin by path, delivering each publisher instance that serves it. /// /// Resolves against future announcements: it waits for the announcement to arrive (e.g. over the /// network) and then delivers the broadcast handle via `on_broadcast`. Use it right after @@ -1849,9 +1849,12 @@ pub extern "C" fn moq_origin_announced_cancel(announced: u32) -> i32 { /// reachable now, use [moq_origin_request] instead. A broadcast created on this origin /// resolves once it is announced, like a remote one. /// -/// `on_broadcast` is invoked with a positive broadcast handle once announced, then exactly once -/// more with a terminal code: `0` (the wait finished, including after -/// [moq_origin_announced_broadcast_cancel]) or a negative error. After the terminal (`<= 0`) callback, +/// `on_broadcast` is invoked with a positive broadcast handle once announced, and again with a +/// new handle each time another publisher instance takes over (a restarted publisher, or a +/// replacement session) or the broadcast returns after it went offline. Drop what was read +/// from the previous handle then: its tracks end when its session does. The wait ends with +/// exactly one terminal code: `0` (after [moq_origin_announced_broadcast_cancel], or the origin +/// closing once a handle was delivered) or a negative error. After the terminal (`<= 0`) callback, /// `on_broadcast` is never called again and `user_data` is never touched again, so release /// `user_data` there. The broadcast handle is usable with [moq_consume_catalog] / [moq_consume_track] /// and must be freed separately with [moq_consume_close]. diff --git a/rs/moq-c/src/consume.rs b/rs/moq-c/src/consume.rs index 48e2c6acbf4..733ace6302c 100644 --- a/rs/moq-c/src/consume.rs +++ b/rs/moq-c/src/consume.rs @@ -176,6 +176,13 @@ impl Consume { self.broadcast.insert(ConsumeBroadcast { broadcast, origin }) } + /// Whether the caller still holds `id` as a handle to `broadcast`. + pub fn holds(&self, id: Id, broadcast: &moq_net::broadcast::Consumer) -> bool { + self.broadcast + .get(id) + .is_some_and(|held| held.broadcast.is_clone(broadcast)) + } + pub fn catalog(&mut self, broadcast: Id, on_catalog: OnStatus) -> Result { let ConsumeBroadcast { broadcast, origin } = self.broadcast.get(broadcast).ok_or(Error::BroadcastNotFound)?.clone(); diff --git a/rs/moq-c/src/origin.rs b/rs/moq-c/src/origin.rs index a4109492dd7..51c796f5b41 100644 --- a/rs/moq-c/src/origin.rs +++ b/rs/moq-c/src/origin.rs @@ -242,11 +242,11 @@ impl Origin { Ok(()) } - /// Wait until the broadcast at `path` is announced, then deliver its handle via the callback. + /// Follow the broadcast at `path`, delivering each publisher instance's handle via the callback. /// - /// The callback fires the broadcast handle (> 0) once announced, then a terminal `0`. On error - /// or cancellation it fires a single terminal code (`0` on close, negative on error). Returns a - /// task handle for cancellation via [`Self::consume_announced_close`]. + /// The callback fires a broadcast handle (> 0) for each instance serving the path, until a + /// single terminal code (`0` on close, negative on error). Returns a task handle for + /// cancellation via [`Self::consume_announced_close`]. pub fn consume_announced(&mut self, origin: Id, path: String, on_broadcast: OnStatus) -> Result { let origin = self.active.get(origin).ok_or(Error::OriginNotFound)?; origin.driver.check()?; @@ -282,27 +282,80 @@ impl Origin { path: String, mut close: oneshot::Receiver<()>, ) -> Result<(), Error> { - // `routed_broadcast` rides out the churn between something covering the path - // and the path actually resolving (failover, an advertise-only announce - // racing its handler). `biased` so a pending close always wins. - let broadcast = tokio::select! { - biased; - _ = &mut close => return Ok(()), - resolved = consumer.routed_broadcast(path.as_str()) => match resolved { - Ok(broadcast) => broadcast, - // An unreachable path and a closed origin both mean no broadcast - // can ever arrive here. - Err(moq_net::Error::Unauthorized | moq_net::Error::Closed) => { - return Err(Error::BroadcastNotFound); - } - Err(err) => return Err(err.into()), - }, + // An unreachable path and a closed origin both mean no broadcast can ever arrive here. + let mut follow = match consumer.follow(path.as_str()) { + Ok(follow) => follow, + Err(moq_net::Error::Unauthorized) => return Err(Error::BroadcastNotFound), + Err(err) => return Err(err.into()), + }; + let mut unseen = false; + let mut delivered = false; + // The handle delivered last, until the caller frees it or the path goes offline. An id + // rather than the broadcast, so it never keeps an instance the caller freed from retiring. + let mut last: Option = None; + // The origin closing ends the wait, cleanly once a handle was delivered. + let closed = |delivered: bool| match delivered { + true => Ok(()), + false => Err(Error::BroadcastNotFound), }; - // Hold the lock only to buffer the broadcast; release it before the callback. - let broadcast_id = State::lock().consume.start(broadcast, Some(consumer))?; - callback.call(broadcast_id); - Ok(()) + // Deliver each publisher instance serving the path: the first, then each one that takes + // over or returns. `biased` so a pending close always wins. + loop { + // Apply every change on hand before resolving, so a burst of them delivers only the + // instance serving the path now. + if settle(&mut follow, &mut unseen, &mut last).is_none() { + return closed(delivered); + } + if !unseen { + let event = tokio::select! { + biased; + _ = &mut close => return Ok(()), + event = follow.next() => event, + }; + let Some(event) = event else { + return closed(delivered); + }; + apply(&mut unseen, &mut last, event); + continue; + } + + // `routed_broadcast` rides out the churn between something covering the path + // and the path actually resolving (failover, an advertise-only announce + // racing its handler). + let broadcast = tokio::select! { + biased; + _ = &mut close => return Ok(()), + resolved = consumer.routed_broadcast(path.as_str()) => match resolved { + Ok(broadcast) => broadcast, + Err(moq_net::Error::Closed) => return closed(delivered), + Err(moq_net::Error::Unauthorized) => return Err(Error::BroadcastNotFound), + Err(err) => return Err(err.into()), + }, + }; + + // A change that came through while resolving may predate the answer, such as an end + // and a start for the instance it resolved. Resolve afresh once caught up. + match settle(&mut follow, &mut unseen, &mut last) { + None => return closed(delivered), + Some(true) => continue, + Some(false) => unseen = false, + } + + // Hold the lock only to buffer the broadcast; release it before the callback. + let broadcast_id = { + let mut state = State::lock(); + // Resolving can run ahead of the follower: a takeover resolves at once while its + // restart is held, and that restart then names the instance already delivered. + if last.is_some_and(|last| state.consume.holds(last, &broadcast)) { + continue; + } + state.consume.start(broadcast, Some(consumer.clone()))? + }; + last = Some(broadcast_id); + delivered = true; + callback.call(broadcast_id); + } } /// Request the broadcast at `path`, delivering its handle once it can be served. @@ -498,6 +551,32 @@ impl Origin { } } +/// Apply every change on hand, returning whether any came through, or `None` once the origin +/// closed. +fn settle(follow: &mut moq_net::announce::Follow, unseen: &mut bool, last: &mut Option) -> Option { + let mut changed = false; + while let std::task::Poll::Ready(event) = follow.poll_next(&kio::Waiter::noop()) { + changed |= apply(unseen, last, event?); + } + Some(changed) +} + +/// Apply one change to the path, returning whether it is one. +fn apply(unseen: &mut bool, last: &mut Option, event: moq_net::announce::Event) -> bool { + match event { + moq_net::announce::Event::Start(_) | moq_net::announce::Event::Restart(_) => *unseen = true, + // An end is a gap that ended every request on the old route, so whatever serves the path + // next is delivered, even the same instance coming back. + moq_net::announce::Event::End(_) => { + *unseen = false; + *last = None; + } + // The same instance over another route, which its subscriptions ride out. + moq_net::announce::Event::Update(_) => return false, + } + true +} + #[cfg(test)] mod supervision_tests { use super::*; diff --git a/rs/moq-c/src/test.rs b/rs/moq-c/src/test.rs index 330a8f81dba..5dbd9aa1963 100644 --- a/rs/moq-c/src/test.rs +++ b/rs/moq-c/src/test.rs @@ -2938,7 +2938,7 @@ fn consume_announced_local() { // Start waiting before the broadcast exists: the announcement arrives afterwards. let cb = Callback::new(); let path = b"live"; - let _task = id(unsafe { + let task = id(unsafe { moq_origin_announced_broadcast( origin, path.as_ptr() as *const c_char, @@ -2953,9 +2953,10 @@ fn consume_announced_local() { let format = moq_audio_format::MOQ_AUDIO_FORMAT_OPUS; let media = id(publish_audio(broadcast, format, &init, None)); - // First the broadcast handle, then a terminal 0 once the wait finishes. + // The broadcast handle; the wait carries on for the next instance until cancelled. let consume = id(cb.recv()); - assert_eq!(cb.recv_terminal(), 0, "wait delivers terminal 0 after the handle"); + assert_eq!(moq_origin_announced_broadcast_cancel(task), 0); + assert_eq!(cb.recv_terminal(), 0, "cancel delivers terminal 0"); // The delivered handle behaves like one from moq_origin_request. let catalog_cb = Callback::new(); @@ -3119,6 +3120,219 @@ fn consume_announced_close_cancels() { assert_eq!(moq_origin_close(origin), 0); } +/// Publish `path` on `origin` through `route`, which the C ABI cannot set. +fn publish(origin: u32, path: &str, route: moq_net::origin::Route) -> moq_net::broadcast::Producer { + let _runtime = crate::ffi::RUNTIME.enter(); + let origin = crate::Id::try_from(origin).unwrap(); + State::lock().origin.get(origin).unwrap().publish(path, route).unwrap() +} + +/// Publish `path` on `origin` under a fresh epoch, as each run of a publisher does. +fn publish_run(origin: u32, path: &str) -> moq_net::broadcast::Producer { + publish( + origin, + path, + moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()), + ) +} + +/// Another publisher instance taking the path delivers a fresh handle on the same wait. +#[test] +fn consume_announced_follows_a_restart() { + let origin = id(moq_origin_create()); + let cb = Callback::new(); + let path = b"live"; + let task = id(unsafe { + moq_origin_announced_broadcast( + origin, + path.as_ptr() as *const c_char, + path.len(), + Some(channel_callback), + cb.ptr, + ) + }); + + let first_run = publish_run(origin, "live"); + let first = id(cb.recv()); + let second_run = publish_run(origin, "live"); + let second = id(cb.recv()); + assert_ne!(first, second, "the restart delivered the old handle"); + + assert_eq!(moq_consume_close(first), 0); + assert_eq!(moq_consume_close(second), 0); + assert_eq!(moq_origin_announced_broadcast_cancel(task), 0); + assert_eq!(cb.recv_terminal(), 0, "cancel delivers terminal 0"); + drop((first_run, second_run)); + assert_eq!(moq_origin_close(origin), 0); +} + +/// Another instance taking over while the wait resolves is the one it delivers, so the restart +/// that announced it must not deliver it again. `dynamic_at` is where the route the wait first +/// resolves through covers the path: above it, the takeover's restart lands while resolving; +/// at it, the origin holds that restart for [`moq_net::origin::DEFAULT_UPDATE_HOLD`]. +fn consume_announced_delivers_a_takeover_once(dynamic_at: &str) { + let origin = id(moq_origin_create()); + let dynamic = { + let _runtime = crate::ffi::RUNTIME.enter(); + let origin = crate::Id::try_from(origin).unwrap(); + let route = moq_net::origin::Route::default(); + State::lock() + .origin + .get(origin) + .unwrap() + .dynamic(dynamic_at, route) + .unwrap() + }; + let cb = Callback::new(); + let path = b"live"; + let task = id(unsafe { + moq_origin_announced_broadcast( + origin, + path.as_ptr() as *const c_char, + path.len(), + Some(channel_callback), + cb.ptr, + ) + }); + + // The wait is on the dynamic route's answer when the exact path takes over without an epoch. + let _unanswered = crate::ffi::RUNTIME.block_on(dynamic.requested_broadcast()).unwrap(); + let takeover = publish(origin, "live", moq_net::origin::Route::default()); + let delivered = id(cb.recv()); + assert!( + cb.rx.recv_timeout(moq_net::origin::DEFAULT_UPDATE_HOLD * 2).is_err(), + "the takeover was delivered twice" + ); + + assert_eq!(moq_consume_close(delivered), 0); + assert_eq!(moq_origin_announced_broadcast_cancel(task), 0); + assert_eq!(cb.recv_terminal(), 0, "cancel delivers terminal 0"); + drop((takeover, dynamic)); + assert_eq!(moq_origin_close(origin), 0); +} + +#[test] +fn consume_announced_delivers_a_takeover_once_while_resolving() { + consume_announced_delivers_a_takeover_once(""); +} + +#[test] +fn consume_announced_delivers_a_held_takeover_once() { + consume_announced_delivers_a_takeover_once("live"); +} + +/// Serve `prefix` on `origin` dynamically through `route`. +fn serve_dynamic(origin: u32, prefix: &str, route: moq_net::origin::Route) -> moq_net::origin::Dynamic { + let _runtime = crate::ffi::RUNTIME.enter(); + let origin = crate::Id::try_from(origin).unwrap(); + State::lock() + .origin + .get(origin) + .unwrap() + .dynamic(prefix, route) + .unwrap() +} + +/// Wait for the broadcast at `path` on `origin`, delivering each instance through `cb`. +fn consume_announced(origin: u32, path: &[u8], cb: &Callback) -> u32 { + id(unsafe { + moq_origin_announced_broadcast( + origin, + path.as_ptr() as *const c_char, + path.len(), + Some(channel_callback), + cb.ptr, + ) + }) +} + +/// An end is a gap that ended every request on the old route, even when a route above the path +/// with the same epoch carries the instance on. So the instance serving the path afterwards is +/// delivered again, though the caller still holds its handle. +#[test] +fn consume_announced_delivers_the_instance_again_after_an_end() { + let origin = id(moq_origin_create()); + let cb = Callback::new(); + let task = consume_announced(origin, b"live", &cb); + + let route = moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()); + let run = publish(origin, "live", route.clone()); + let first = id(cb.recv()); + + let dynamic = serve_dynamic(origin, "", route); + { + let _runtime = crate::ffi::RUNTIME.enter(); + run.unannounce(); + } + let request = crate::ffi::RUNTIME.block_on(dynamic.requested_broadcast()).unwrap(); + let served = { + let _runtime = crate::ffi::RUNTIME.enter(); + let served = moq_net::broadcast::Info::new().produce(); + request.accept(served.consume()); + served + }; + let again = id(cb.recv()); + + assert_eq!(moq_consume_close(first), 0); + assert_eq!(moq_consume_close(again), 0); + assert_eq!(moq_origin_announced_broadcast_cancel(task), 0); + assert_eq!(cb.recv_terminal(), 0, "cancel delivers terminal 0"); + drop((run, dynamic, served)); + assert_eq!(moq_origin_close(origin), 0); +} + +/// A handle the caller freed no longer counts as delivered, so the restart that names its +/// instance delivers it again rather than leaving the caller with nothing. +#[test] +fn consume_announced_delivers_a_freed_takeover_again() { + let origin = id(moq_origin_create()); + let dynamic = serve_dynamic(origin, "live", moq_net::origin::Route::default()); + let cb = Callback::new(); + let task = consume_announced(origin, b"live", &cb); + + // Without an epoch, the origin holds the takeover's restart past its resolution. + let _unanswered = crate::ffi::RUNTIME.block_on(dynamic.requested_broadcast()).unwrap(); + let takeover = publish(origin, "live", moq_net::origin::Route::default()); + assert_eq!(moq_consume_close(id(cb.recv())), 0); + let again = id(cb.recv()); + + assert_eq!(moq_consume_close(again), 0); + assert_eq!(moq_origin_announced_broadcast_cancel(task), 0); + assert_eq!(cb.recv_terminal(), 0, "cancel delivers terminal 0"); + drop((takeover, dynamic)); + assert_eq!(moq_origin_close(origin), 0); +} + +/// The route a wait resolves through leaving, and another instance arriving before it resolves, +/// reach the follower as an end and a start after the wait delivers the newcomer. That start +/// must not deliver it again. +#[test] +fn consume_announced_delivers_a_gap_while_resolving_once() { + let origin = id(moq_origin_create()); + let dynamic = serve_dynamic(origin, "", moq_net::origin::Route::default()); + let cb = Callback::new(); + let task = consume_announced(origin, b"live", &cb); + + let _unanswered = crate::ffi::RUNTIME.block_on(dynamic.requested_broadcast()).unwrap(); + { + let _runtime = crate::ffi::RUNTIME.enter(); + drop(dynamic); + } + std::thread::sleep(moq_net::origin::DEFAULT_UPDATE_HOLD * 2); + let arrival = publish(origin, "live", moq_net::origin::Route::default()); + let delivered = id(cb.recv()); + assert!( + cb.rx.recv_timeout(moq_net::origin::DEFAULT_UPDATE_HOLD * 2).is_err(), + "the arrival was delivered twice" + ); + + assert_eq!(moq_consume_close(delivered), 0); + assert_eq!(moq_origin_announced_broadcast_cancel(task), 0); + assert_eq!(cb.recv_terminal(), 0, "cancel delivers terminal 0"); + drop(arrival); + assert_eq!(moq_origin_close(origin), 0); +} + #[test] fn video_publish_consume() { let origin = id(moq_origin_create()); diff --git a/rs/moq-cli/tests/integration/export.rs b/rs/moq-cli/tests/integration/export.rs index 3465e5af161..3d958d68eb1 100644 --- a/rs/moq-cli/tests/integration/export.rs +++ b/rs/moq-cli/tests/integration/export.rs @@ -1,6 +1,7 @@ //! `moq export ts --linger` and `--stitch` over a real relay: the export rides out the same //! publisher instance leaving and coming back, switches to a replacement only when asked, and -//! exits with the verdict of the broadcast's last end. +//! exits with the verdict of the broadcast's last end. The other formats fail once the route +//! they read dies. #![cfg(unix)] use std::io::Cursor; @@ -246,15 +247,19 @@ fn import_as(relay: &Relay, global: &[&str], format: Format) -> (Child, JoinHand .stdin(Stdio::piped()) .spawn() .expect("spawn import"); - let mut stdin = child.stdin.take().expect("stdin"); - let feeding = tokio::spawn(async move { + let feeding = feed(child.stdin.take().expect("stdin"), format); + (child, feeding) +} + +/// Feed `format`'s clip into `stdin` at about real time, handing it back once it is all written. +fn feed(mut stdin: ChildStdin, format: Format) -> JoinHandle { + tokio::spawn(async move { for chunk in format.clip.chunks(format.chunk) { stdin.write_all(chunk).await.expect("write stdin"); tokio::time::sleep(Duration::from_millis(150)).await; } stdin - }); - (child, feeding) + }) } /// End the publisher the way an operator does, which drops the broadcast unfinished. @@ -709,3 +714,58 @@ async fn an_old_publisher_that_stays_up_keeps_the_export() { drop(new_stdin); wait(&mut new).await; } + +/// [`import_as`] that returns once the publisher has sent its announcement to the relay, which +/// it does once its input declares a track. +async fn import_announced(relay: &Relay, format: Format) -> (Child, JoinHandle) { + let mut child = moq(relay, &[]) + .args(["import", "--max-age", "1s", format.name]) + .env("RUST_LOG", "warn,moq_net::lite::publisher=debug") + .stdin(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .expect("spawn import"); + let feeding = feed(child.stdin.take().expect("stdin"), format); + + let mut stderr = BufReader::new(child.stderr.take().expect("stderr")).lines(); + let announced = async { + while let Some(line) = stderr.next_line().await.expect("read import stderr") { + let other = ["unannounce", "reannounce", "no longer", "cancelled", "withholding"]; + if line.contains("announce") && line.contains("demo") && !other.iter().any(|o| line.contains(o)) { + return; + } + } + panic!("the import exited before announcing"); + }; + tokio::time::timeout(TIMEOUT, announced) + .await + .expect("the import never announced"); + // Keep draining, so a full pipe never blocks the import. + tokio::spawn(async move { while let Ok(Some(_)) = stderr.next_line().await {} }); + (child, feeding) +} + +/// A route change alone never ends a non-TS export, which cannot mark a break, but the old +/// route dying does: its tracks end with an error rather than cleanly. The old publisher is +/// interrupted as soon as the replacement has announced, so the relay is usually still +/// holding the replacement's restart. +#[tokio::test] +async fn a_replaced_publisher_dying_fails_an_fmp4_export() { + let relay = relay().await; + let mut export = export_as(&relay, &[], FMP4, &[]); + let (old, feeding) = import_as(&relay, &[], FMP4); + export.output_past(0).await; + let old_stdin = feeding.await.unwrap(); + + let (mut new, replacing) = import_announced(&relay, FMP4).await; + interrupt(old, old_stdin).await; + + let (status, errors) = export.wait().await; + assert!( + !status.success(), + "the dead route ended the export cleanly; exporter stderr: {errors}" + ); + + drop(replacing.await.unwrap()); + wait(&mut new).await; +} diff --git a/rs/moq-ffi/src/origin.rs b/rs/moq-ffi/src/origin.rs index 191813ac8e6..79ca3a595a5 100644 --- a/rs/moq-ffi/src/origin.rs +++ b/rs/moq-ffi/src/origin.rs @@ -1,4 +1,4 @@ -use std::sync::Arc; +use std::sync::{Arc, Weak}; use crate::consumer::MoqBroadcastConsumer; use crate::error::MoqError; @@ -202,27 +202,87 @@ impl Announced { } } -/// Waits for a route covering one exact path, then resolves it. +/// Follows one exact path, resolving each publisher instance that serves it. struct AnnouncedBroadcast { origin: moq_net::origin::Consumer, path: moq_net::PathOwned, + follow: moq_net::announce::Follow, + /// An instance serves the path that was not handed out yet. + unseen: bool, + /// The handle handed out last, until the caller drops it or the path goes offline. Weak, so + /// it never keeps an instance the caller let go of from retiring. + last: Weak, driver: Background, } impl AnnouncedBroadcast { async fn available(&mut self) -> Result, MoqError> { - // `routed_broadcast` rides out the churn between something covering the path - // and the path actually resolving (failover, an advertise-only announce - // racing its handler). - let broadcast = self - .driver - .watch(async { Ok(self.origin.routed_broadcast(&self.path).await?) }) - .await?; - Ok(Arc::new(MoqBroadcastConsumer::routed( - broadcast, - self.origin.clone(), - self.driver.clone(), - ))) + let driver = self.driver.clone(); + driver + .watch(async { + loop { + // Apply every change on hand before resolving, so a burst of them hands out + // only the instance serving the path now. + self.settle()?; + if !self.unseen { + let event = self.follow.next().await; + self.apply(event)?; + continue; + } + + // `routed_broadcast` rides out the churn between something covering the path + // and the path actually resolving (failover, an advertise-only announce + // racing its handler). The instance stays unseen until it resolves: a foreign + // cancel only drops this future, and the caller may call again. + let broadcast = self.origin.routed_broadcast(&self.path).await?; + + // A change that came through while resolving may predate the answer, such as + // an end and a start for the instance it resolved. Resolve afresh once caught up. + if self.settle()? { + continue; + } + self.unseen = false; + + // Resolving can run ahead of the follower: a takeover resolves at once while + // its restart is held, and that restart then names the instance handed out. + if self.last.upgrade().is_some_and(|last| last.inner.is_clone(&broadcast)) { + continue; + } + let broadcast = Arc::new(MoqBroadcastConsumer::routed( + broadcast, + self.origin.clone(), + self.driver.clone(), + )); + self.last = Arc::downgrade(&broadcast); + return Ok(broadcast); + } + }) + .await + } + + /// Apply every change on hand, returning whether any came through. + fn settle(&mut self) -> Result { + let mut changed = false; + while let std::task::Poll::Ready(event) = self.follow.poll_next(&kio::Waiter::noop()) { + changed |= self.apply(event)?; + } + Ok(changed) + } + + /// Apply one change to the path, returning whether it is one. + fn apply(&mut self, event: Option) -> Result { + match event.ok_or(MoqError::Closed)? { + moq_net::announce::Event::Start(_) | moq_net::announce::Event::Restart(_) => self.unseen = true, + // An end is a gap that ended every request on the old route, so whatever serves the + // path next is handed out, even the same instance coming back. + moq_net::announce::Event::End(_) => { + self.unseen = false; + self.last = Weak::new(); + } + // The same instance over another route, which its subscriptions ride out. + moq_net::announce::Event::Update(_) => return Ok(false), + } + Ok(true) } } @@ -287,7 +347,7 @@ impl From for MoqAnnounceEvent { } } -/// Waits for a specific broadcast to be announced. +/// Follows a specific broadcast across the publisher instances that serve it. #[derive(uniffi::Object)] pub struct MoqAnnouncedBroadcast { task: Task, @@ -448,7 +508,7 @@ impl MoqOriginConsumer { })) } - /// Resolve the broadcast at `path`, waiting until something can serve it. + /// Follow the broadcast at `path`, resolving each publisher instance that serves it. /// /// This is how you resolve a path right after connecting: announcements arrive over the /// session after it opens, so `request_broadcast` on its own races them. A @@ -459,19 +519,21 @@ impl MoqOriginConsumer { self.driver.check()?; let path = moq_net::Path::new(&path).to_owned(); - // Probe the permission eagerly so an unreachable path fails here, rather than - // surfacing later as a `Closed` the caller can't tell from the origin ending. - self.inner - .scope(&path, &moq_net::Patterns::from(moq_net::Pattern::all()))?; + // Following checks the permission eagerly, so an unreachable path fails here rather + // than surfacing later as a `Closed` the caller can't tell from the origin ending. + let follow = self.inner.follow(&path)?; Ok(Arc::new(MoqAnnouncedBroadcast { task: Task::new(AnnouncedBroadcast { - // The wait runs on the *unrooted* cursor. `announced_broadcast` narrows with - // `scope`, which leaves the root alone, so the broadcast is handed out named by - // its full path. Rooting the cursor at `path` would name it "" instead, making - // it its own root, and a catalog's `../sibling` reference would read as escaping. + // The wait runs on the *unrooted* cursor, so the broadcast is handed out named + // by its full path. Rooting the cursor at `path` would name it "" instead, + // making it its own root, and a catalog's `../sibling` reference would read as + // escaping. origin: self.inner.clone(), path, + follow, + unseen: false, + last: Weak::new(), driver: self.driver.clone(), }), })) @@ -603,9 +665,14 @@ impl MoqAnnounceConsumer { #[uniffi::export] impl MoqAnnouncedBroadcast { - /// Wait until the broadcast is announced. Returns `Closed` if cancelled or the origin is closed. + /// Wait for the next publisher instance serving the broadcast. Returns `Closed` if cancelled + /// or the origin is closed. /// - /// Its end arrives as a [`MoqAnnounceEvent::End`] on the origin's announcements. + /// The first call resolves the instance serving the path, once one is announced. Each later + /// call waits for another to take over, as a restarted publisher or a replacement session + /// does, or for the broadcast to return after it went offline. Drop everything read from the + /// previous instance then and read the new one: the old one's tracks end when its route + /// dies, with an error unless this side closed that session. pub async fn available(&self) -> Result, MoqError> { self.task.run(|mut state| async move { state.available().await }).await } diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index 410534358af..d54d37b36e0 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -4506,6 +4506,298 @@ async fn server_client_roundtrip() { server.cancel(); } +/// Subscribe to the first audio rendition in `broadcast`'s catalog. +async fn subscribe_audio(broadcast: &MoqBroadcastConsumer) -> Arc { + let catalogs = MoqMediaCatalogConsumer::subscribe(broadcast).await.unwrap(); + let catalog = tokio::time::timeout(TIMEOUT, catalogs.next()) + .await + .expect("timed out waiting for the catalog") + .unwrap() + .expect("expected a catalog"); + let (id, config) = catalog.audio.iter().next().expect("an audio rendition"); + MoqMediaTrackConsumer::subscribe( + broadcast, + MoqMediaTrackConfig { + rendition: MoqRendition::Audio { + id: id.clone(), + config: config.clone(), + }, + subscription: None, + }, + ) + .await + .unwrap() +} + +/// Write `payload` and read through `track` until it arrives, past what a new subscription +/// starts with. +async fn roundtrip(media: &MoqMediaTrackProducer, track: &MoqMediaTrackConsumer, payload: &[u8], timestamp_us: u64) { + media + .write_frame(MoqFrame { + payload: payload.to_vec(), + timestamp_us: Some(timestamp_us), + }) + .unwrap(); + loop { + let frame = tokio::time::timeout(TIMEOUT, track.next()) + .await + .expect("timed out waiting for a frame") + .unwrap() + .expect("the track ended"); + if frame.payload == payload { + return; + } + } +} + +/// A replacement session from the same publisher re-announces every route while the old one +/// is still open, as a GOAWAY drain does. Without an epoch that is another instance: the +/// next `available()` hands it out, and reading it carries on past the old session's close. +#[tokio::test] +async fn announced_broadcast_follows_an_epochless_replacement() { + let server_origin = MoqOriginProducer::new(MoqOriginConfig::default()); + let server = MoqServer::new(MoqServerConfig { + bind: Some("127.0.0.1:0".into()), + tls: localhost_tls(), + publish: Some(server_origin.clone()), + ..Default::default() + }) + .unwrap(); + let addr = tokio::time::timeout(TIMEOUT, server.listen()) + .await + .expect("listen timed out") + .expect("listen failed"); + let url = format!("https://{addr}"); + let accepting = server.clone(); + let accept = tokio::spawn(async move { + let mut sessions = Vec::new(); + while let Ok(Some(request)) = accepting.accept().await { + sessions.push(request.accept(None, None).await.expect("handshake failed")); + } + sessions + }); + + let broadcast = create_announced(&server_origin, "cam"); + let media = MoqMediaTrackProducer::audio( + &broadcast, + MoqMediaTarget::Named { name: None }, + audio_init(MoqAudioFormat::Opus, opus_head()), + ) + .unwrap(); + + let viewer = MoqOriginProducer::new(MoqOriginConfig::default()); + let connect = || async { + let client = MoqClient::new(MoqClientConfig { + tls: insecure_tls(), + bind: Some("127.0.0.1:0".into()), + consume: Some(viewer.clone()), + ..Default::default() + }) + .unwrap(); + tokio::time::timeout(TIMEOUT, client.connect(url.clone())) + .await + .expect("connect timed out") + .expect("connect failed") + }; + + let first = connect().await; + let followed = viewer.consume().announced_broadcast("cam".into()).unwrap(); + let old = tokio::time::timeout(TIMEOUT, followed.available()) + .await + .expect("timed out waiting for the broadcast") + .unwrap(); + let old_track = subscribe_audio(&old).await; + roundtrip(&media, &old_track, b"before", 1_000_000).await; + + let second = connect().await; + let fresh = tokio::time::timeout(TIMEOUT, followed.available()) + .await + .expect("the replacement was never handed out") + .unwrap(); + first.shutdown().await.unwrap(); + + let fresh_track = subscribe_audio(&fresh).await; + roundtrip(&media, &fresh_track, b"after", 2_000_000).await; + + media.finish().unwrap(); + broadcast.close().unwrap(); + second.shutdown().await.unwrap(); + server.cancel(); + drop(accept.await.unwrap()); +} + +/// A foreign cancel only drops the `available()` future, so a caller retrying after a timeout +/// keeps the same handle. Dropped while the instance is covered but not yet served, the retry +/// must still hand that instance out rather than wait for another. +#[tokio::test] +async fn announced_broadcast_retries_a_dropped_wait() { + let origin = MoqOriginProducer::new(MoqOriginConfig::default()); + let dynamic = serve(&origin, ""); + let followed = origin.consume().announced_broadcast("live".into()).unwrap(); + + let waiting = followed.clone(); + let dropped = tokio::spawn(async move { waiting.available().await }); + let request = tokio::time::timeout(TIMEOUT, dynamic.requested_broadcast()) + .await + .expect("the wait never asked for the broadcast") + .unwrap(); + dropped.abort(); + assert!(matches!(dropped.await, Err(err) if err.is_cancelled())); + + let retry = tokio::spawn(async move { followed.available().await }); + let served = MoqBroadcastProducer::new().unwrap(); + request.accept(&served).unwrap(); + tokio::time::timeout(TIMEOUT, retry) + .await + .expect("the retry waited for another instance") + .unwrap() + .unwrap(); + served.close().unwrap(); +} + +/// Another instance taking over while `available()` resolves is the one it hands out, so the +/// restart that announced it must not hand it out again while the caller holds it. `dynamic_at` +/// is where the route the wait first resolves through covers the path: above it, the takeover's +/// restart lands while resolving; at it, the origin holds that restart for +/// [`moq_net::origin::DEFAULT_UPDATE_HOLD`]. +async fn announced_broadcast_hands_out_a_takeover_once(dynamic_at: &str) { + let origin = MoqOriginProducer::new(MoqOriginConfig::default()); + let dynamic = serve(&origin, dynamic_at); + let followed = origin.consume().announced_broadcast("live".into()).unwrap(); + + let waiting = followed.clone(); + let first = tokio::spawn(async move { waiting.available().await }); + // The wait is on the dynamic route's answer when the exact path takes over without an epoch. + let _unanswered = tokio::time::timeout(TIMEOUT, dynamic.requested_broadcast()) + .await + .expect("the wait never asked for the broadcast") + .unwrap(); + let takeover = create_announced(&origin, "live"); + let _held = tokio::time::timeout(TIMEOUT, first) + .await + .expect("the takeover never resolved") + .unwrap() + .unwrap(); + + let next = tokio::spawn(async move { followed.available().await }); + tokio::time::sleep(moq_net::origin::DEFAULT_UPDATE_HOLD * 2).await; + assert!(!next.is_finished(), "the takeover was handed out twice"); + next.abort(); + takeover.close().unwrap(); +} + +#[tokio::test] +async fn announced_broadcast_hands_out_a_takeover_once_while_resolving() { + announced_broadcast_hands_out_a_takeover_once("").await; +} + +#[tokio::test] +async fn announced_broadcast_hands_out_a_held_takeover_once() { + announced_broadcast_hands_out_a_takeover_once("live").await; +} + +/// An end is a gap that ended every request on the old route, even when a route above the path +/// with the same epoch carries the instance on. So the instance serving the path afterwards is +/// handed out again, though it is the one the caller still holds. +#[tokio::test] +async fn announced_broadcast_hands_out_the_instance_again_after_an_end() { + let origin = MoqOriginProducer::new(MoqOriginConfig::default()); + let followed = origin.consume().announced_broadcast("live".into()).unwrap(); + let route = MoqRoute { + epoch: Some(moq_mint_epoch()), + ..Default::default() + }; + let broadcast = origin.create_broadcast("live".into()).unwrap(); + broadcast.announce(route.clone()).unwrap(); + let first = tokio::time::timeout(TIMEOUT, followed.available()) + .await + .expect("timed out waiting for the broadcast") + .unwrap(); + + let dynamic = origin.dynamic("".into(), route).unwrap(); + broadcast.unannounce().unwrap(); + let again = tokio::spawn(async move { followed.available().await }); + let request = tokio::time::timeout(TIMEOUT, dynamic.requested_broadcast()) + .await + .expect("the instance never asked the route above the path") + .unwrap(); + let served = MoqBroadcastProducer::new().unwrap(); + request.accept(&served).unwrap(); + + let again = tokio::time::timeout(TIMEOUT, again) + .await + .expect("the instance was never handed out again") + .unwrap() + .unwrap(); + assert!( + again.inner().is_clone(first.inner()), + "the route above did not carry the instance on" + ); + served.close().unwrap(); + broadcast.close().unwrap(); +} + +/// The route a wait resolves through leaving, and another instance arriving before it resolves, +/// reach the follower as an end and a start after the wait hands the newcomer out. That start +/// must not hand it out again. +#[tokio::test] +async fn announced_broadcast_hands_out_a_gap_while_resolving_once() { + let origin = MoqOriginProducer::new(MoqOriginConfig::default()); + let dynamic = serve(&origin, ""); + let followed = origin.consume().announced_broadcast("live".into()).unwrap(); + + let waiting = followed.clone(); + let first = tokio::spawn(async move { waiting.available().await }); + let _unanswered = tokio::time::timeout(TIMEOUT, dynamic.requested_broadcast()) + .await + .expect("the wait never asked for the broadcast") + .unwrap(); + dynamic.cancel(); + tokio::time::sleep(moq_net::origin::DEFAULT_UPDATE_HOLD * 2).await; + let arrival = create_announced(&origin, "live"); + let _held = tokio::time::timeout(TIMEOUT, first) + .await + .expect("the arrival never resolved") + .unwrap() + .unwrap(); + + let next = tokio::spawn(async move { followed.available().await }); + tokio::time::sleep(moq_net::origin::DEFAULT_UPDATE_HOLD * 2).await; + assert!(!next.is_finished(), "the arrival was handed out twice"); + next.abort(); + arrival.close().unwrap(); +} + +/// A handle the caller dropped no longer counts as handed out, so the restart that names it +/// hands it out again rather than leaving the caller with nothing. +#[tokio::test] +async fn announced_broadcast_hands_out_a_dropped_takeover_again() { + let origin = MoqOriginProducer::new(MoqOriginConfig::default()); + let dynamic = serve(&origin, "live"); + let followed = origin.consume().announced_broadcast("live".into()).unwrap(); + + let waiting = followed.clone(); + let first = tokio::spawn(async move { waiting.available().await }); + let _unanswered = tokio::time::timeout(TIMEOUT, dynamic.requested_broadcast()) + .await + .expect("the wait never asked for the broadcast") + .unwrap(); + let takeover = create_announced(&origin, "live"); + drop( + tokio::time::timeout(TIMEOUT, first) + .await + .expect("the takeover never resolved") + .unwrap() + .unwrap(), + ); + + tokio::time::timeout(TIMEOUT, followed.available()) + .await + .expect("the dropped takeover was never handed out again") + .unwrap(); + takeover.close().unwrap(); +} + #[tokio::test] async fn server_client_roundtrip_auto_origin() { // Same shape as `server_client_roundtrip` but the client config omits diff --git a/rs/moq-net/tests/integration/replacement.rs b/rs/moq-net/tests/integration/replacement.rs index fa906a190cd..147c663aea3 100644 --- a/rs/moq-net/tests/integration/replacement.rs +++ b/rs/moq-net/tests/integration/replacement.rs @@ -170,3 +170,81 @@ async fn epoch_lite07() { .await .expect("timed out"); } + +/// A route change alone is never fatal, but a dead route is: once an epochless replacement +/// is announced, a subscription on the old instance keeps going while the old publisher +/// serves it, and its session closing ends that subscription with an error, never cleanly. +/// The viewer reads either straight from the publisher or through a relay, and the old +/// session closes after `wait`: inside the update hold, before the `Restart` is delivered, +/// or past it. +async fn route_death(version: &str, relayed: bool, wait: Duration) { + let version: Version = version.parse().unwrap(); + let publisher = produce_origin(1); + let relay = produce_origin(2); + let viewer = produce_origin(3); + + let cam = publisher.publish("cam", origin::Route::default()).unwrap(); + let video = cam.create_track("video", timed()).unwrap(); + + let upstream = if relayed { &relay } else { &viewer }; + let old_session = link(version, &publisher, upstream).await; + let _downstream = match relayed { + true => Some(link(version, &relay, &viewer).await), + false => None, + }; + let mut announced = viewer.consume().announced(); + next_event(&mut announced).await; + + let old = viewer.consume().request_broadcast("cam", None).await.unwrap(); + let mut sticky = subscribe(&old, "video").await.unwrap(); + write(&video, "video:0"); + assert_eq!(read(&mut sticky).await, "video:0", "{version}"); + + let _replacement = link(version, &publisher, upstream).await; + moq_clock::sleep(wait).await; + write(&video, "video:1"); + assert_eq!( + read(&mut sticky).await, + "video:1", + "{version}: the replacement cut the old instance off while it served" + ); + + let _ = old_session.server.close().await; + let end = moq_clock::timeout(Duration::from_secs(10), async { + loop { + match sticky.recv_group().await { + Ok(Some(_)) => continue, + Ok(None) => return None, + Err(err) => return Some(err), + } + } + }) + .await + .expect("the dead route's subscription hung"); + assert!( + end.is_some(), + "{version} (relayed: {relayed}): the dead route's subscription ended cleanly" + ); +} + +#[moq_clock::test] +async fn route_death_within_hold_errors() { + for version in ["moq-lite-06", "moq-lite-07-wip"] { + for relayed in [false, true] { + moq_clock::timeout(TEST_TIMEOUT, route_death(version, relayed, WITHIN_HOLD)) + .await + .expect("timed out"); + } + } +} + +#[moq_clock::test] +async fn route_death_past_hold_errors() { + for version in ["moq-lite-06", "moq-lite-07-wip"] { + for relayed in [false, true] { + moq_clock::timeout(TEST_TIMEOUT, route_death(version, relayed, PAST_HOLD)) + .await + .expect("timed out"); + } + } +} diff --git a/swift/Sources/Moq/Origin.swift b/swift/Sources/Moq/Origin.swift index 9bca84e20d1..081e2406f67 100644 --- a/swift/Sources/Moq/Origin.swift +++ b/swift/Sources/Moq/Origin.swift @@ -116,7 +116,7 @@ public final class OriginConsumer: Sendable { AnnounceConsumer(try ffi.announced(config: MoqAnnounceConfig(prefix: prefix, filter: filter, hidden: hidden))) } - /// Wait for a route covering an exact path, then resolve the broadcast there. + /// Follow an exact path, resolving each publisher instance that serves it. public func announcedBroadcast(path: String) throws -> AnnouncedBroadcast { AnnouncedBroadcast(try ffi.announcedBroadcast(path: path)) } @@ -171,8 +171,12 @@ public final class AnnouncedBroadcast: Sendable { self.ffi = ffi } - /// Suspend until a route covers the path, then resolve the broadcast there. - /// Throws `Closed` if cancelled or the origin closes first. + /// Suspend until the next publisher instance serves the path, then resolve it. + /// The first call resolves the instance serving the path once one is announced. + /// Each later call waits for another to take over (a restarted publisher, or a + /// replacement session) or for the broadcast to return after it went offline. + /// Drop what was read from the previous instance then: its tracks end when its + /// session does. Throws `Closed` if cancelled or the origin closes first. public func available() async throws -> BroadcastConsumer { BroadcastConsumer(try await ffi.available()) } diff --git a/test/interop/clients/c/subscribe.c b/test/interop/clients/c/subscribe.c index d960dc01acf..44fd96570a5 100644 --- a/test/interop/clients/c/subscribe.c +++ b/test/interop/clients/c/subscribe.c @@ -74,8 +74,9 @@ static void on_frame(void *ud, int32_t frame) { moq_consume_frame_free((uint32_t)frame); } -// Delivers the broadcast handle once it's announced, then once more with a -// terminal code (<= 0). Store the first positive handle and wake main. +// Delivers a broadcast handle for each publisher instance serving the path, then +// a terminal code (<= 0). Store the first positive handle and wake main; one frame +// is all this client reads, so a later instance's handle is freed unread. static void on_broadcast(void *ud, int32_t broadcast) { ctx_t *c = (ctx_t *)ud; if (broadcast <= 0) { // 0 = ended, negative = error @@ -83,11 +84,13 @@ static void on_broadcast(void *ud, int32_t broadcast) { return; } pthread_mutex_lock(&c->mu); - if (c->broadcast <= 0) { + int first = c->broadcast <= 0; + if (first) { c->broadcast = broadcast; pthread_cond_broadcast(&c->cv); } pthread_mutex_unlock(&c->mu); + if (!first) moq_consume_close((uint32_t)broadcast); } static void on_catalog(void *ud, int32_t catalog) { @@ -118,19 +121,35 @@ static void on_catalog(void *ud, int32_t catalog) { moq_consume_catalog_free((uint32_t)catalog); } +// The context whose session a failure closes, once connected. +static ctx_t *failing; + // Report a failure and exit 1, for the paths where moq-c still holds &ctx. // // _exit rather than a return: it leaves main's frame (and so ctx) intact for the -// callbacks still pointing at it, which is what the failure paths want. Draining -// first would work too, but a client that has already failed should print its -// reason and go, not block for up to ten seconds on a teardown that may be exactly -// what is broken. +// callbacks still pointing at it, which is what the failure paths want. Only the +// session is closed first, so the relay sees a close rather than timing the +// connection out; a client that has already failed should not wait on the rest of +// a teardown that may be exactly what is broken. static _Noreturn void fail(const char *fmt, ...) { va_list ap; va_start(ap, fmt); vfprintf(stderr, fmt, ap); va_end(ap); fflush(stderr); + + ctx_t *c = failing; + failing = NULL; // a failure while closing exits at once + if (c && moq_session_close((uint32_t)c->session) == 0) { + struct timespec deadline; + clock_gettime(CLOCK_REALTIME, &deadline); + deadline.tv_sec += 5; + pthread_mutex_lock(&c->mu); + while (!c->done_status) { + if (pthread_cond_timedwait(&c->cv, &c->mu, &deadline) != 0) break; // timed out + } + pthread_mutex_unlock(&c->mu); + } _exit(1); } @@ -206,6 +225,7 @@ int main(int argc, char **argv) { fprintf(stderr, "error: moq_session_connect failed: %d\n", c.session); return 1; } + failing = &c; struct timespec deadline; clock_gettime(CLOCK_REALTIME, &deadline); diff --git a/test/interop/clients/cpp/main.cpp b/test/interop/clients/cpp/main.cpp index 28cc6d4729d..29083c62132 100644 --- a/test/interop/clients/cpp/main.cpp +++ b/test/interop/clients/cpp/main.cpp @@ -22,6 +22,7 @@ #include #include #include +#include #include #ifdef _WIN32 @@ -53,14 +54,26 @@ constexpr uint32_t AUDIO_FRAME_DURATION_US = 2'500; constexpr uint64_t AUDIO_BATCH_US = 20'000; constexpr size_t AUDIO_BATCH_SAMPLES = AUDIO_RATE * AUDIO_BATCH_US / 1'000'000; +// The open session, which a failure closes so the relay sees a close rather than timing the +// connection out. +std::shared_ptr open_session; + +[[noreturn]] void exit_failed() { + // Taken first, so a failure while closing cannot close it again. + if (auto session = std::exchange(open_session, nullptr)) { + session->shutdown().wait_for(std::chrono::seconds(5)); + } + std::exit(1); +} + [[noreturn]] void fail(const char *what, const moq::Error &error) { std::fprintf(stderr, "error: %s: %s\n", what, error.to_string().c_str()); - std::exit(1); + exit_failed(); } [[noreturn]] void fail(const char *what) { std::fprintf(stderr, "error: %s\n", what); - std::exit(1); + exit_failed(); } template @@ -91,12 +104,14 @@ std::shared_ptr connect(const std::string &url, Clock::time_point moq::ClientConfig config; config.tls.insecure = true; auto client = ok(moq::Client::init(config), "client init"); - return until(client->connect(url), deadline, "connect"); + open_session = until(client->connect(url), deadline, "connect"); + return open_session; } // Closes the session and waits for the close to reach the relay. `cancel` would return at // once, and stopping the runtime right after it can leave the relay to time the connection out. void close_session(const std::shared_ptr &session) { + open_session = nullptr; until(session->shutdown(), Clock::now() + std::chrono::seconds(5), "session shutdown"); }