From 205d4df7a0710ad4e4e880af501be698bf5bbbdb Mon Sep 17 00:00:00 2001 From: Vsevolod Strukchinsky Date: Tue, 20 Oct 2026 19:59:27 +0500 Subject: [PATCH] relay: a resume outlives the downstream that asked for it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit propagateForwardUpstream ran the §7.2 REQUEST_UPDATE FORWARD=1 under the triggering downstream's context. A subscriber that left while the relay awaited REQUEST_UPDATE_OK cancelled the wait, so SetForwardState(1) was skipped although the publisher had already resumed. The relay then recorded the upstream as paused, and the pause the leaving subscriber scheduled (Config.PauseUnwatchedPublishes, 85049b6) found nothing to do: the publisher kept forwarding a track nobody watched. Run the resume detached from ctx, as the pause already is; each round trip stays bounded by UpstreamSub.Update's own timeout. The explicit WithoutCancel at the late-publisher call site is now redundant. This is what made TestRelay_UnwatchedPublishPaused fail ~15% of runs (36/200 on draft-22). TestRelay_UnwatchedPublishPausedAfterAnInterruptedResume holds the publisher's REQUEST_UPDATE_OK until the subscriber has left; verified red before the fix (20/20), green after (100/100). Co-Authored-By: Claude Opus 5.5 --- pkg/relay/forward_state_test.go | 39 +++++++++++++++++++++++++++++++++ pkg/relay/handler_namespace.go | 5 ++--- pkg/relay/handler_subscribe.go | 8 +++++-- 3 files changed, 47 insertions(+), 5 deletions(-) diff --git a/pkg/relay/forward_state_test.go b/pkg/relay/forward_state_test.go index 7aef851c..53d79a38 100644 --- a/pkg/relay/forward_state_test.go +++ b/pkg/relay/forward_state_test.go @@ -104,6 +104,45 @@ func TestRelay_UnwatchedPublishPaused(t *testing.T) { requireForward(t, forwards, false, "once the subscriber left") } +// TestRelay_UnwatchedPublishPausedAfterAnInterruptedResume: a subscriber that +// leaves while the resume it caused awaits the publisher's REQUEST_UPDATE_OK +// still leaves the track paused after the delay. The resume reached the +// publisher, so cutting its round trip short must not leave the relay +// believing the upstream is still paused. +func TestRelay_UnwatchedPublishPausedAfterAnInterruptedResume(t *testing.T) { + t.Parallel() + anchor, teardown := connectRelay(t, relay.Config{PauseUnwatchedPublishes: 100 * time.Millisecond}) + defer teardown() + pubSess := dialAnotherClient(t, anchor) + pub, err := pubSess.Publish(t.Context(), &message.Publish{Namespace: ns("video"), Name: []byte("cam1")}) + if err != nil { + t.Fatalf("Publish: %v", err) + } + t.Cleanup(func() { _ = pub.Close() }) + forwards := make(chan bool, 4) + release := make(chan struct{}) + b := pub.Broker() + // Holds a resume's REQUEST_UPDATE_OK until release. + b.HandleUpdates(func(upd *message.RequestUpdate) (*message.RequestOK, error) { + if f, found := upd.Parameters.Find(message.ParamForward); found { + forwards <- f.Byte == 1 + if f.Byte == 1 { + <-release + } + } + return pub.ApplyUpdate(upd) + }) + go func() { _ = b.Serve(t.Context(), nil) }() + + requireForward(t, forwards, false, "for a PUBLISH nobody subscribed to") + sub := newCam1Subscriber(t, anchor) + requireForward(t, forwards, true, "when a subscriber came") + _ = sub.Close(0, "done") + time.Sleep(50 * time.Millisecond) // the relay sees the subscriber gone + close(release) + requireForward(t, forwards, false, "once the subscriber left mid-resume") +} + // TestRelay_UnwatchedPublishNotPausedByDefault: without // PauseUnwatchedPublishes, a PUBLISH nobody subscribes to flows on. func TestRelay_UnwatchedPublishNotPausedByDefault(t *testing.T) { diff --git a/pkg/relay/handler_namespace.go b/pkg/relay/handler_namespace.go index fe0c9487..1565c27e 100644 --- a/pkg/relay/handler_namespace.go +++ b/pkg/relay/handler_namespace.go @@ -168,10 +168,9 @@ func (h *sessionHandler) subscribeLatePublisher( return } // Or switched to Forward=1 before this upstream was registered for §7.2 - // propagation. Not bound to ctx: a withdrawal (§7.6) stops only new - // subscriptions. + // propagation. if up.ForwardState() == 0 && anyDownstreamForwards(entry) { - h.propagateForwardUpstream(context.WithoutCancel(ctx), fullName) + h.propagateForwardUpstream(ctx, fullName) } } diff --git a/pkg/relay/handler_subscribe.go b/pkg/relay/handler_subscribe.go index 3cf12b37..c8639515 100644 --- a/pkg/relay/handler_subscribe.go +++ b/pkg/relay/handler_subscribe.go @@ -330,9 +330,13 @@ func (h *sessionHandler) propagateNewGroupUpstream( } // propagateForwardUpstream sends REQUEST_UPDATE Forward=1 to each paused -// upstream of fullName (§7.2), saving any LARGEST_OBJECT in the reply. +// upstream of fullName (§7.2), saving any LARGEST_OBJECT in the reply. Not +// bound to ctx (each round trip is bounded by [registry.UpstreamSub.Update]): +// a downstream leaving mid round trip would leave the upstream recorded +// paused after the publisher resumed, and never paused again +// ([Config.PauseUnwatchedPublishes]). func (h *sessionHandler) propagateForwardUpstream(ctx context.Context, fullName track.FullTrackName) { - h.reconcileForwardUpstream(ctx, fullName, false) + h.reconcileForwardUpstream(context.WithoutCancel(ctx), fullName, false) } // pauseUnwatched schedules a pause of fullName's PUBLISH-fed upstreams after