relay: a resume outlives the downstream that asked for it

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) {

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)
 	}
 }
 

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