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