fix(moq): mint epochs for published broadcast lifetimes

changelog/5986.fixed.md

@@ -1 +1 @@
-- Fixed `MOQTransport` ending the call when a relay cancelled the peer's audio subscription, which a relay does when the session publishing the peer's broadcast ends: the peer's relay went away, or a reconnecting peer replaced its broadcast with a fresh one. The cancel is now treated as the peer's tracks ending, so the transport subscribes again or redials and follows the peer to its new broadcast.
+- Fixed `MOQTransport` ending the call when a relay cancelled the peer's audio subscription, which a relay does when the session publishing the peer's broadcast ends: the peer's relay went away, or a reconnecting peer replaced its broadcast with a fresh one. The cancel is now treated as the peer's tracks ending, so the transport subscribes again or redials and follows the peer to its new broadcast. Each rebuilt publish broadcast carries a fresh route epoch; re-announcements of that broadcast retain it.

src/pipecat/transports/moq/transport.py

@@ -649,6 +649,7 @@ def __init__(
         self._publish_broadcast: moq.BroadcastProducer = self._publish_origin.create_broadcast(
             self._broadcast_path
         )
+        self._publish_route = moq.Route(epoch=moq.mint_epoch())
         # Lossless, ordered JSON append-log for RTVI (compression on). The
         # stream helper delivers every appended record in order — the
         # snapshot helper (publish_json_snapshot) would collapse to the
@@ -661,7 +662,7 @@ def __init__(
         # transcript is that track, announce now. Audio on:
         # ``open_audio_track`` announces after adding the audio track.
         if not params.audio_out_enabled:
-            self._publish_broadcast.announce()
+            self._publish_broadcast.announce(self._publish_route)
         # Audio track is opened lazily once the pipeline's output sample
         # rate is known (in :meth:`open_audio_track`). We stash the rate
         # because ``publish_audio`` uses it to convert byte length to
@@ -794,7 +795,7 @@ def open_audio_track(self, sample_rate: int):
         # Announce broadcast to peers now that subscriber-needed tracks exist
         # (catalog via audio rendition above, transcript stream). moq-rs 0.5
         # requires explicit announce; until announced, broadcast is unroutable.
-        self._publish_broadcast.announce()
+        self._publish_broadcast.announce(self._publish_route)
 
     @property
     def _audio_track_name(self) -> str:
@@ -830,11 +831,10 @@ def _publish_audio_track(self, sample_rate: int) -> "moq.AudioProducer":
     async def _rebuild_publish_side(self):
         """Replace the publish origin, broadcast and tracks before a redial.
 
-        The origin carries the publisher identity the relay sees. A redial
-        that re-announces under the identity of a publisher the relay still
-        holds a dead route for leaves the path stuck on that route, while a
-        new identity is parked until the dead route is reaped and then takes
-        the path over. The transcript log is replayed into the new stream
+        A rebuilt broadcast gets a fresh route epoch, so peers replace the
+        previous instance instead of resuming its cached groups. Re-announcing
+        the same broadcast keeps its epoch. The transcript log is replayed into
+        the new stream
         with its original ``seq`` and ``epoch``, so subscribers drop what
         they already have, and the audio track reopens at the same rate.
 
@@ -853,6 +853,7 @@ async def _rebuild_publish_side(self):
                 pass
         self._publish_origin = moq.OriginProducer()
         self._publish_broadcast = self._publish_origin.create_broadcast(self._broadcast_path)
+        self._publish_route = moq.Route(epoch=moq.mint_epoch())
         self._transcript_out = self._publish_broadcast.publish_json_stream(
             self._params.transcript_track, compression=True
         )
@@ -862,7 +863,7 @@ async def _rebuild_publish_side(self):
             self._audio_out = self._publish_audio_track(self._audio_out_sample_rate)
         # Re-announce rebuilt broadcast (moq-rs 0.5 requires explicit
         # announce; see open_audio_track).
-        self._publish_broadcast.announce()
+        self._publish_broadcast.announce(self._publish_route)
         # By index against the live log: a record published while the
         # replay yields is only logged (see publish_transcript), and is
         # written from here, after the records before it.

tests/test_moq_transport.py

@@ -2459,5 +2459,32 @@ async def test_catalog_advertises_the_replacement_track(self):
         self.assertEqual(set(updated.audio), {"bot-audio-1"})
 
 
+class TestBroadcastEpochs(unittest.IsolatedAsyncioTestCase):
+    """The broadcast instance, independently of its tracks, owns route identity."""
+
+    async def test_reannouncements_keep_identity_and_rebuild_renews_it(self):
+        client, _stream = _client_with_fake_moq()
+        client.open_audio_track(24000)
+        old = client._publish_broadcast
+        self.assertEqual(len(old.announce.call_args.args), 1)
+        first = old.announce.call_args.args[0]
+        self.assertIsNotNone(first.epoch)
+        client.open_audio_track(24000)
+        self.assertEqual(old.announce.call_args.args[0].epoch, first.epoch)
+        with patch.object(moq_transport.moq, "OriginProducer", side_effect=_fake_origin):
+            await client._rebuild_publish_side()
+        second = client._publish_broadcast.announce.call_args.args[0]
+        self.assertNotEqual(second.epoch, first.epoch)
+
+    async def test_transcript_only_instances_are_distinct(self):
+        a, _ = _client_with_fake_moq(MOQParams(audio_out_enabled=False))
+        b, _ = _client_with_fake_moq(MOQParams(audio_out_enabled=False))
+        calls = [client._publish_broadcast.announce.call_args for client in (a, b)]
+        for call in calls:
+            self.assertEqual(len(call.args), 1)
+            self.assertIsNotNone(call.args[0].epoch)
+        self.assertNotEqual(calls[0].args[0].epoch, calls[1].args[0].epoch)
+
+
 if __name__ == "__main__":
     unittest.main()