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.