From b3414039bdc07fe27f33b09ca3c09e2ab06cf418 Mon Sep 17 00:00:00 2001
From: Eli Mallon <eli@iame.li>
Date: Sat, 17 Oct 2026 18:12:02 -0700
Subject: [PATCH] Announce a pushed stream afresh after its broadcast ends

A pushed broadcast ends on its own when a track's codec config changes,
but its live entry stayed, so later segments only refreshed it and the
stream stayed retracted on the peer. The worker now forgets the entry,
and the next segment announces a fresh broadcast.
---
 pkg/replication/websocketrep/moq.go      | 15 ++++++++--
 pkg/replication/websocketrep/moq_test.go | 38 ++++++++++++++++++++++++
 2 files changed, 50 insertions(+), 3 deletions(-)

diff --git a/pkg/replication/websocketrep/moq.go b/pkg/replication/websocketrep/moq.go
index 07a88582a..05eb3d4f5 100644
--- a/pkg/replication/websocketrep/moq.go
+++ b/pkg/replication/websocketrep/moq.go
@@ -655,7 +655,9 @@ type liveStream struct {
 
 // touchLive notes a local segment of streamer's. On the first, the stream
 // is announced on pushOrigin, its tracks served from the bus like the
-// listener's; it is retracted once expireLive finds it quiet.
+// listener's; it is retracted once expireLive finds it quiet, or when its
+// broadcast ends on its own (a track's codec config changed), after which
+// the next segment announces it afresh.
 func (r *WebsocketReplicator) touchLive(ctx context.Context, streamer string) {
 	r.liveMu.Lock()
 	defer r.liveMu.Unlock()
@@ -669,10 +671,17 @@ func (r *WebsocketReplicator) touchLive(ctx context.Context, streamer string) {
 		return
 	}
 	ctx, stop := context.WithCancel(ctx)
-	r.live[streamer] = &liveStream{last: time.Now(), stop: stop}
+	s := &liveStream{last: time.Now(), stop: stop}
+	r.live[streamer] = s
 	r.group.Go(func() error {
-		defer broadcast.Close()
 		r.serveTracks(ctx, streamer, broadcast, func() error { return broadcast.Announce(official.Route{}) })
+		_ = broadcast.Close()
+		r.liveMu.Lock()
+		if r.live[streamer] == s {
+			delete(r.live, streamer)
+		}
+		r.liveMu.Unlock()
+		stop()
 		return nil
 	})
 }
diff --git a/pkg/replication/websocketrep/moq_test.go b/pkg/replication/websocketrep/moq_test.go
index 7d7d14460..88274f131 100644
--- a/pkg/replication/websocketrep/moq_test.go
+++ b/pkg/replication/websocketrep/moq_test.go
@@ -243,6 +243,44 @@ func TestLiveStreamsAreAnnounced(t *testing.T) {
 	require.False(t, a.Active(), "retracted, after one announcement")
 }
 
+// A pushed broadcast that ends on its own, as one does when a track's codec
+// config changes, is announced afresh from the stream's next segment
+// instead of staying retracted while the stream goes on.
+func TestEndedPushIsAnnouncedAgain(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	b := bus.NewBus()
+	r := NewWebsocketReplicator(&config.CLI{StreamSessionTimeout: time.Minute}, b, nil, nil, nil, nil)
+	r.group, ctx = errgroup.WithContext(ctx)
+	announced, err := r.pushOrigin.Consume().Announced(official.AnnounceOptions{})
+	require.NoError(t, err)
+	defer announced.Cancel()
+	next := func() bool {
+		a, err := announced.Next(ctx)
+		require.NoError(t, err)
+		return a.Active()
+	}
+	segments := signedSegments(t, ctx)
+	publish := func(segment []byte) {
+		b.PublishSegment(ctx, "did:plc:a", moqSourceChannel, &bus.Seg{Muxl: segment, Published: true})
+	}
+
+	publish(segments[0])
+	r.touchLive(ctx, "did:plc:a")
+	require.True(t, next())
+	publish([]byte("not a segment the broadcast can carry"))
+	require.False(t, next(), "the broadcast ended")
+	require.Eventually(t, func() bool {
+		r.liveMu.Lock()
+		defer r.liveMu.Unlock()
+		return r.live["did:plc:a"] == nil
+	}, 5*time.Second, 10*time.Millisecond, "and was forgotten")
+	publish(segments[0])
+	publish(segments[1])
+	r.touchLive(ctx, "did:plc:a")
+	require.True(t, next(), "announced again")
+}
+
 // A peer this node pushes to gets each live stream announced on the session
 // and its segments served for as long as the stream is live, not just the
 // first one.
-- 
2.54.0