Follow epochless restarts in exports and bindings

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);
 

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() !=

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.

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
 

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)
 }

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 {

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<void> {
+	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<typeof consume.announced> | undefined;
+	let replacement: Awaited<ReturnType<typeof connect>> | undefined;
+	let replacementServer: Awaited<ReturnType<typeof accept>> | 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);
+	});
+}

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)
 

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`.
         """

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.

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.

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

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.