From 4725affabd54b4c33517b119d2f6cc4da0a71164 Mon Sep 17 00:00:00 2001
From: Vsevolod Strukchinsky <vsevolod@strukchinsky.ru>
Date: Mon, 28 Sep 2026 10:31:03 +0500
Subject: [PATCH 1/4] =?UTF-8?q?fix(relay):=20answer=20a=20request=20not=20?=
 =?UTF-8?q?opened=20for=20want=20of=20stream=20credit=20with=20EXCESSIVE?=
 =?UTF-8?q?=5FLOAD=20(=C2=A710.6.2)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A SUBSCRIBE, or forwarded TRACK_STATUS, that the relay could not open to a
live publisher for want of bidi-stream credit was answered DOES_NOT_EXIST
with Retry Interval 0: "not available at the publisher", and "SHOULD NOT be
retried". But the publisher is there, and the relay only "cannot process
the request at this time".

upstreamRejection now answers it EXCESSIVE_LOAD with the relay's jittered
~1s retry, the code it already uses for its own caps. The maintainer chose
this. It ranks as "any other refusal", so it outranks another candidate's
GOING_AWAY, TIMEOUT or DOES_NOT_EXIST. candidateRetry reports its Retry
Interval so ties compare what is sent.

TestRendezvous_NoStreamCreditEndsHold now expects EXCESSIVE_LOAD with a
retry. TestSubscribe_NoStreamCreditOutranksNoPublisherYet covers both orders
against a GOING_AWAY, and forwarded TRACK_STATUS. All failed before this
change.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 STATUS.md                          |  5 ---
 pkg/relay/handler_subscribe.go     | 17 +++++++--
 pkg/relay/session_handler.go       |  2 ++
 pkg/relay/session_upstream_test.go | 56 ++++++++++++++++++++++++++++--
 4 files changed, 71 insertions(+), 9 deletions(-)

diff --git a/STATUS.md b/STATUS.md
index 1758082f..19c12d8d 100644
--- a/STATUS.md
+++ b/STATUS.md
@@ -540,11 +540,6 @@ Relay:
 - Filters are not aggregated upstream (§6.3.1 SHOULD).
 - Concurrent TRACK_STATUS requests for one track with no Established
   subscription are each forwarded upstream, not coalesced.
-- A SUBSCRIBE the relay cannot open to a candidate for want of bidi-stream
-  credit is answered DOES_NOT_EXIST with no Retry Interval ("SHOULD NOT be
-  retried"), and outranks another candidate's GOING_AWAY or TIMEOUT, though
-  the publisher is there; EXCESSIVE_LOAD with a retry would fit better
-  (§10.6.2).
 
 Open questions for interop: whether an End of Range marker carries an Object
 Payload Length (Figure 28 vs §11.4.4.2), and whether EXPIRES may appear in
diff --git a/pkg/relay/handler_subscribe.go b/pkg/relay/handler_subscribe.go
index a22c3452..ef6abf3f 100644
--- a/pkg/relay/handler_subscribe.go
+++ b/pkg/relay/handler_subscribe.go
@@ -569,12 +569,16 @@ func preferCandidateErr(last, err error) error {
 }
 
 // candidateRetry is the Retry Interval err is answered with: for a
-// GOING_AWAY without one, the most the relay's own jittered one can be (see
-// [upstreamRejection]), else the upstream's.
+// GOING_AWAY without one, or a request not opened for want of stream credit,
+// the most the relay's own jittered one can be (see [upstreamRejection]), else
+// the upstream's.
 func candidateRetry(err error) uint64 {
 	if errors.Is(err, errGoingAway) {
 		return goingAwayRetry
 	}
+	if errors.Is(err, session.ErrNoStreamCredit) {
+		return uint64(excessiveLoadRetry/time.Millisecond) * 3 / 2
+	}
 	rej, ok := errors.AsType[*session.RequestRejectedError](err)
 	if !ok {
 		return 0
@@ -979,6 +983,15 @@ func upstreamRejection(err error) *session.RequestRejectedError {
 			RetryInterval: retryIntervalAfter(goingAwayRetryAfter),
 		}
 	}
+	if errors.Is(err, session.ErrNoStreamCredit) {
+		// A live publisher the relay could not open a request to for want
+		// of stream credit: it "cannot process the request at this time"
+		// (§10.6.2), and may shortly.
+		return &session.RequestRejectedError{
+			Code:          moqt.RequestExcessiveLoad,
+			RetryInterval: retryIntervalAfter(excessiveLoadRetry),
+		}
+	}
 	up, ok := errors.AsType[*session.RequestRejectedError](err)
 	if !ok {
 		return &session.RequestRejectedError{Code: moqt.RequestDoesNotExist}
diff --git a/pkg/relay/session_handler.go b/pkg/relay/session_handler.go
index 406ccb82..77fc781c 100644
--- a/pkg/relay/session_handler.go
+++ b/pkg/relay/session_handler.go
@@ -425,6 +425,8 @@ const excessiveLoadRetry = time.Second
 
 // retryIntervalAfter is a Retry Interval (§10.6.2) inviting a retry after d
 // plus up to 50% jitter, encoded as milliseconds plus one.
+//
+//nolint:unparam // callers pass separate policy constants, equal only today.
 func retryIntervalAfter(d time.Duration) uint64 {
 	ms := uint64(d / time.Millisecond) //nolint:gosec // G115: callers pass a positive constant duration.
 	return ms + rand.Uint64N(ms/2) + 1 //nolint:gosec // G404: retry jitter, not a secret.
diff --git a/pkg/relay/session_upstream_test.go b/pkg/relay/session_upstream_test.go
index 7db7423f..6f801724 100644
--- a/pkg/relay/session_upstream_test.go
+++ b/pkg/relay/session_upstream_test.go
@@ -539,10 +539,25 @@ func TestSubscribe_UnsupportedMandatoryPropertyOutranksMalformed(t *testing.T) {
 	}
 }
 
+// requireRetryableExcessiveLoad fails unless err is EXCESSIVE_LOAD inviting a
+// retry after about a second (§10.6.2).
+func requireRetryableExcessiveLoad(t *testing.T, err error) {
+	t.Helper()
+	requireRejectedWithCode(t, err, moqt.RequestExcessiveLoad)
+	if rej, _ := errors.AsType[*session.RequestRejectedError](
+		err,
+	); rej.RetryInterval < 1001 ||
+		rej.RetryInterval > 1500 {
+		t.Fatalf("Retry Interval = %d, want a retry after about 1s", rej.RetryInterval)
+	}
+}
+
 // TestRendezvous_NoStreamCreditEndsHold: a SUBSCRIBE the relay cannot even
 // open to a live publisher, for want of bidi-stream credit, is not a sign the
 // track has no publisher, so a RENDEZVOUS_TIMEOUT hold does not wait out its
-// budget on it: the subscriber is answered at once, as without a hold.
+// budget on it: the subscriber is answered at once, with EXCESSIVE_LOAD and a
+// retry, since the relay "cannot process the request at this time"
+// (§10.6.2).
 func TestRendezvous_NoStreamCreditEndsHold(t *testing.T) {
 	t.Parallel()
 	subSess, teardown := connectRelay(t, relay.Config{})
@@ -552,12 +567,49 @@ func TestRendezvous_NoStreamCreditEndsHold(t *testing.T) {
 	done := subscribeRendezvous(t.Context(), subSess, 5*time.Second)
 	select {
 	case err := <-done:
-		requireRejectedWithCode(t, err, moqt.RequestDoesNotExist)
+		requireRetryableExcessiveLoad(t, err)
 	case <-time.After(time.Second):
 		t.Fatal("SUBSCRIBE held against a live publisher the relay had no stream credit for")
 	}
 }
 
+// TestSubscribe_NoStreamCreditOutranksNoPublisherYet: a candidate the relay
+// has no stream credit for is a live publisher, so its EXCESSIVE_LOAD wins
+// over another candidate's GOING_AWAY, whatever the order; and a forwarded
+// TRACK_STATUS gets the same answer.
+func TestSubscribe_NoStreamCreditOutranksNoPublisherYet(t *testing.T) {
+	t.Parallel()
+	for _, order := range []string{"credit-less first", "credit-less last"} {
+		t.Run(order, func(t *testing.T) {
+			t.Parallel()
+			subSess, teardown := connectRelay(t, relay.Config{})
+			t.Cleanup(teardown)
+			creditless := func() {
+				pub := dialAnotherClientWithLimits(t, subSess, -1, 0)
+				publishNS(t, pub, "video")
+			}
+			if order == "credit-less first" {
+				creditless()
+			}
+			refusingPublishers(t, subSess, refuse(moqt.RequestGoingAway, 2001))
+			if order == "credit-less last" {
+				creditless()
+			}
+			_, err := subSess.Subscribe(t.Context(), &message.Subscribe{Namespace: ns("video"), Name: []byte("cam1")})
+			requireRetryableExcessiveLoad(t, err)
+		})
+	}
+	t.Run("TRACK_STATUS", func(t *testing.T) {
+		t.Parallel()
+		subSess, teardown := connectRelay(t, relay.Config{})
+		t.Cleanup(teardown)
+		pub := dialAnotherClientWithLimits(t, subSess, -1, 0)
+		publishNS(t, pub, "video")
+		_, err := trackStatusCam1(t, subSess)
+		requireRetryableExcessiveLoad(t, err)
+	})
+}
+
 // TestSubscribe_OtherRefusalTieIsOrderFree: two refusals of the "any other"
 // kind with the same Retry Interval, one answered as INTERNAL_ERROR (an
 // UNAUTHORIZED about the relay's hop) and one passed on (EXCESSIVE_LOAD),

From d1e81726d0856203d19531067092da5473c3ab70 Mon Sep 17 00:00:00 2001
From: Vsevolod Strukchinsky <vsevolod@strukchinsky.ru>
Date: Mon, 28 Sep 2026 10:40:57 +0500
Subject: [PATCH 2/4] feat(relay): share one forwarded TRACK_STATUS round among
 concurrent requests for a track
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Each TRACK_STATUS for a track with no Established subscription forwarded
its own round upstream, so N concurrent requests made N upstream ones to
every candidate (§13.1 amplification). The maintainer chose to share rounds.

forwardTrackStatus now runs trackStatusUpstream through a relay-wide
singleflight.Group, keyed by track:
- requests for a track that arrive while a round is in flight wait for it;
- each requester is answered with its own INCLUDE_PROPERTIES and counts
  against its own session's cap;
- the round is bounded by its 5s timeout, not by any one requester;
- a requester's STOP_SENDING stops only its own wait, freeing its slot.

A request from a session the relay is asking about the track right now may
be the relay's own TRACK_STATUS routed back (§6.2). It does not join a round
that waits on it: it runs its own, where the loop guard skips that session.

TestTrackStatus_ConcurrentRequestsShareOneRound failed before this change
(the publisher was asked twice). TestTrackStatus_CancelReleasesRequester
replaces TestTrackStatus_CancelEndsUpstream, since a requester's cancel no
longer ends a round others may share. TestRelay_TrackStatusLoopStopsAtSecondHop
now bounds the answer time, and fails, deadlocked for the full bound, when
a routed-back request joins the round.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 STATUS.md                              |  4 +-
 pkg/relay/handler_track_status.go      | 42 +++++++++++--
 pkg/relay/relay.go                     |  8 ++-
 pkg/relay/session_handler.go           |  7 +++
 pkg/relay/track_status_forward_test.go | 82 ++++++++++++++++++++++----
 5 files changed, 121 insertions(+), 22 deletions(-)

diff --git a/STATUS.md b/STATUS.md
index 19c12d8d..51690fc2 100644
--- a/STATUS.md
+++ b/STATUS.md
@@ -199,7 +199,7 @@ By package, bottom-up along the dependency stack:
 | 10.12   | PUBLISH_DONE                  | 0x0B   | DONE   | Sent once every stream of the subscription has closed and no datagram send is in progress, with the exact Stream Count; written on its own goroutine, so subscribers do not wait on each other. When a track's last upstream ends, its PUBLISH_DONE code reaches subscribers if it is about the track (TRACK_ENDED, MALFORMED_TRACK); codes about the relay's own upstream subscription become INTERNAL_ERROR. Session `Publication.Done` resets the subgroups still open with CANCELLED, refuses later opens and writes (`ErrPublicationEnded`), and counts every subgroup opened, however the opens race it. |
 | 10.13   | FETCH                         | 0x16   | DONE   | Standalone, the only kind in draft-20. From the cache, a Location is non-existent only on a signal: a Prior Group or Object ID Gap, a Group's or the Track's end, or an upstream's FETCH. Other uncached Locations are FETCHed from a fetch-capable upstream in one span, within FILL_TIMEOUT, or else marked End of Unknown (or Timed-Out) Range. |
 | 10.14   | FETCH_OK                      | 0x18   | DONE   | The relay sets End Of Track when the End Location is the Object an END_OF_TRACK status made the Track's final one; it does not learn a Track's end from an upstream FETCH_OK's End Of Track. An End Location before the FETCH's Start closes the session. A Start relative to the Largest Object is compared through End ≤ Largest; an End of {0,0} is let through, as it cannot be told apart from "no content yet". |
-| 10.15   | TRACK_STATUS                  | 0x0D   | DONE   | Reply via REQUEST_OK, then FIN; any follow-up from the requester closes the session. The relay answers from an Established subscription, else forwards TRACK_STATUS to every candidate SUBSCRIBE would try, concurrently, and combines the answers as a SUBSCRIBE_OK would (the entry's Track Properties, else the first answer's; the largest LARGEST_OBJECT), or gives the refusal SUBSCRIBE would. A forwarding TRACK_STATUS counts against `Config.MaxSubscriptionsPerSession` (§13.1), and each upstream round trip is bounded at 5s, counting as TIMEOUT (§13.6). Relay policy: not to the requester's own session while a TRACK_STATUS for the track to it is in flight (a loop, §6.2). |
+| 10.15   | TRACK_STATUS                  | 0x0D   | DONE   | Reply via REQUEST_OK, then FIN; any follow-up from the requester closes the session. The relay answers from an Established subscription, else forwards TRACK_STATUS to every candidate SUBSCRIBE would try, concurrently, and combines the answers as a SUBSCRIBE_OK would (the entry's Track Properties, else the first answer's; the largest LARGEST_OBJECT), or gives the refusal SUBSCRIBE would. A forwarding TRACK_STATUS counts against `Config.MaxSubscriptionsPerSession` (§13.1), and the round is bounded at 5s, a silent candidate counting as TIMEOUT (§13.6). Concurrent requests for a track share one round, each answered with its own INCLUDE_PROPERTIES; a requester's STOP_SENDING stops only its wait. Relay policy: not to the requester's own session while a TRACK_STATUS for the track to it is in flight (a loop, §6.2). |
 | 10.16   | PUBLISH_NAMESPACE             | 0x06   | DONE   | |
 | 10.17   | NAMESPACE                     | 0x08   | DONE   | Per namespace, counted over local and remote sources. |
 | 10.18   | NAMESPACE_DONE                | 0x0E   | DONE   | Never before its NAMESPACE. |
@@ -538,8 +538,6 @@ Relay:
   by a relay peer (§6.2 has no loop protection), so it declines the second hop
   rather than loop. Self-subscriptions are otherwise "identical" (§5.1).
 - Filters are not aggregated upstream (§6.3.1 SHOULD).
-- Concurrent TRACK_STATUS requests for one track with no Established
-  subscription are each forwarded upstream, not coalesced.
 
 Open questions for interop: whether an End of Range marker carries an Object
 Payload Length (Figure 28 vs §11.4.4.2), and whether EXPIRES may appear in
diff --git a/pkg/relay/handler_track_status.go b/pkg/relay/handler_track_status.go
index 80886c0c..e7fd199a 100644
--- a/pkg/relay/handler_track_status.go
+++ b/pkg/relay/handler_track_status.go
@@ -63,7 +63,7 @@ func (h *sessionHandler) handleTrackStatus(ctx context.Context, req *session.Req
 			h.rejectExcessiveLoad(ctx, req, "subscription")
 			return
 		}
-		oks, err := h.trackStatusUpstream(ctx, req, fullName)
+		oks, err := h.forwardTrackStatus(ctx, req, fullName)
 		h.limiter.releaseSub()
 		if len(oks) == 0 {
 			h.rejectTrackStatus(ctx, req, err)
@@ -125,6 +125,39 @@ func (h *sessionHandler) rejectTrackStatus(ctx context.Context, req *session.Req
 	}
 }
 
+// forwardTrackStatus answers TRACK_STATUS for fullName from a round of
+// [sessionHandler.trackStatusUpstream], shared with the other requests for the
+// track that arrive while it is in flight (relay policy, so N requests do not
+// make N upstream ones). The round is bounded by its timeout, not by any one
+// requester, who stops waiting on its own STOP_SENDING.
+//
+// A request from a session the relay is asking about the track right now may
+// be the relay's own TRACK_STATUS routed back to it (§6.2), which the round
+// waits on: it runs a round of its own instead, where the loop guard skips
+// that session.
+func (h *sessionHandler) forwardTrackStatus(
+	ctx context.Context,
+	req *session.Request,
+	fullName track.FullTrackName,
+) ([]*message.TrackStatusOK, error) {
+	waitCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	defer context.AfterFunc(req.Stream.Context(), cancel)()
+	if h.tracks.RequestPending(message.TypeTrackStatus, h.sess, fullName.Key()) {
+		return h.trackStatusUpstream(waitCtx, fullName)
+	}
+	round := h.statusRounds.DoChan(string(fullName.Key().Bytes()), func() (any, error) {
+		return h.trackStatusUpstream(context.WithoutCancel(ctx), fullName)
+	})
+	select {
+	case r := <-round:
+		oks, _ := r.Val.([]*message.TrackStatusOK)
+		return oks, r.Err
+	case <-waitCtx.Done():
+		return nil, waitCtx.Err()
+	}
+}
+
 // trackStatusUpstream forwards TRACK_STATUS for fullName, concurrently, to
 // every candidate SUBSCRIBE would try (see [sessionHandler.subscribeUpstream]):
 // each local publisher of a covering namespace and each relay Discovery
@@ -132,16 +165,14 @@ func (h *sessionHandler) rejectTrackStatus(ctx context.Context, req *session.Req
 // are none, the refusal SUBSCRIBE would give ([preferCandidateErr]), nil when there
 // was no candidate.
 //
-// trackStatusUpstreamTimeout bounds the whole forwarding, resolving the
-// Discovery candidates included, and it all ends when the requester cancels
-// (STOP_SENDING). A draining candidate is sent nothing
+// trackStatusUpstreamTimeout bounds the whole round, resolving the Discovery
+// candidates included. A draining candidate is sent nothing
 // (§10.4). Relay policy, as for SUBSCRIBE and FETCH: the requester's own
 // session is skipped while a TRACK_STATUS for the track to it is in flight,
 // since this request may be that one routed back, and a second would loop
 // (§6.2).
 func (h *sessionHandler) trackStatusUpstream(
 	ctx context.Context,
-	req *session.Request,
 	fullName track.FullTrackName,
 ) ([]*message.TrackStatusOK, error) {
 	timeout := trackStatusUpstreamTimeout
@@ -150,7 +181,6 @@ func (h *sessionHandler) trackStatusUpstream(
 	}
 	upCtx, cancel := context.WithTimeout(ctx, timeout)
 	defer cancel()
-	defer context.AfterFunc(req.Stream.Context(), cancel)()
 
 	key := fullName.Key()
 	var lastErr error
diff --git a/pkg/relay/relay.go b/pkg/relay/relay.go
index db62783f..680543b7 100644
--- a/pkg/relay/relay.go
+++ b/pkg/relay/relay.go
@@ -14,6 +14,8 @@ import (
 	"sync"
 	"time"
 
+	"golang.org/x/sync/singleflight"
+
 	"github.com/floatdrop/moq-go/pkg/moqt"
 	"github.com/floatdrop/moq-go/pkg/moqt/message"
 	"github.com/floatdrop/moq-go/pkg/moqt/session"
@@ -299,6 +301,10 @@ type Relay struct {
 	// the FETCH. Shared across every session handler.
 	fetch *registry.FetchRouter
 
+	// statusRounds shares forwarded TRACK_STATUS rounds across every
+	// session handler; see [sessionHandler.forwardTrackStatus].
+	statusRounds singleflight.Group
+
 	// upstreams dials and pools relay-to-relay sessions for Discovery-driven
 	// cross-relay upstream SUBSCRIBE. nil when Config.Dialer is unset (the
 	// single-instance case); session handlers treat a nil pool as "no
@@ -642,7 +648,7 @@ func (r *Relay) serveSession(ctx context.Context, sess *session.Session, leg Leg
 
 	handler := newSessionHandler(
 		sess, r.log, r.tracks, r.names,
-		r.cfg.Authorizer, r.cfg.Metrics, leg, r.fetch, r.upstreams,
+		r.cfg.Authorizer, r.cfg.Metrics, leg, r.fetch, &r.statusRounds, r.upstreams,
 		r.cfg.Discovery, r.cfg.RelayAddr,
 		r.cfg.SendQueueSize, r.cfg.MaxDropsBeforeReset, r.cfg.MaxFanoutLag,
 		r.cfg.MaxSubscriptionsPerSession, r.cfg.MaxNamespaceRequestsPerSession,
diff --git a/pkg/relay/session_handler.go b/pkg/relay/session_handler.go
index 77fc781c..655f5489 100644
--- a/pkg/relay/session_handler.go
+++ b/pkg/relay/session_handler.go
@@ -12,6 +12,8 @@ import (
 	"sync/atomic"
 	"time"
 
+	"golang.org/x/sync/singleflight"
+
 	"github.com/floatdrop/moq-go/pkg/moqt"
 	"github.com/floatdrop/moq-go/pkg/moqt/message"
 	"github.com/floatdrop/moq-go/pkg/moqt/session"
@@ -41,6 +43,9 @@ type sessionHandler struct {
 	auth    Authorizer
 	metrics Metrics
 	fetch   *registry.FetchRouter
+	// statusRounds shares forwarded TRACK_STATUS rounds across the relay's
+	// sessions; see [sessionHandler.forwardTrackStatus].
+	statusRounds *singleflight.Group
 	// leg records whether this session was dialled by the relay
 	// (LegUpstream) or by the peer (LegLocal). Every [Metrics] call this
 	// handler makes carries it, so an operator can separate what the
@@ -90,6 +95,7 @@ func newSessionHandler(
 	metrics Metrics,
 	leg Leg,
 	fetch *registry.FetchRouter,
+	statusRounds *singleflight.Group,
 	upstreams *upstreamPool,
 	discovery discovery.DiscoveryStore,
 	relayAddr string,
@@ -110,6 +116,7 @@ func newSessionHandler(
 		metrics:             metrics,
 		leg:                 leg,
 		fetch:               fetch,
+		statusRounds:        statusRounds,
 		upstreams:           upstreams,
 		discovery:           discovery,
 		relayAddr:           relayAddr,
diff --git a/pkg/relay/track_status_forward_test.go b/pkg/relay/track_status_forward_test.go
index 37ab838a..066a1726 100644
--- a/pkg/relay/track_status_forward_test.go
+++ b/pkg/relay/track_status_forward_test.go
@@ -197,10 +197,15 @@ func TestRelay_TrackStatusLoopStopsAtSecondHop(t *testing.T) {
 			_ = r.AcceptTrackStatus(nil)
 		}()
 	})
+	start := time.Now()
 	_, _ = trackStatusCam1(t, dialAnotherClient(t, peer))
 	if n := asked.Load(); n != 1 {
 		t.Fatalf("the relay forwarded TRACK_STATUS to the peer %d times, want 1", n)
 	}
+	// Not by waiting out a round the routed-back request waits on itself.
+	if d := time.Since(start); d > 2*time.Second {
+		t.Fatalf("answered after %v: the routed-back request deadlocked with the round", d)
+	}
 }
 
 // namespacePeer is a fresh session on the relay sess is on that
@@ -304,29 +309,82 @@ func TestTrackStatus_SilentCandidateTimesOut(t *testing.T) {
 	requireRejectedWithCode(t, err, moqt.RequestTimeout)
 }
 
-// TestTrackStatus_CancelEndsUpstream: the requester cancelling its
-// TRACK_STATUS (§3.3.3) cancels the relay's upstream one.
-func TestTrackStatus_CancelEndsUpstream(t *testing.T) {
+// TestTrackStatus_CancelReleasesRequester: the requester cancelling its
+// TRACK_STATUS (§3.3.3) stops the relay waiting on the upstream round for it,
+// freeing its place under MaxSubscriptionsPerSession (§13.1) before the round
+// ends; the round itself, which other requesters may share, runs to its bound.
+func TestTrackStatus_CancelReleasesRequester(t *testing.T) {
 	t.Parallel()
-	client, teardown := connectRelay(t, relay.Config{})
+	client, teardown := connectRelay(t, relay.Config{MaxSubscriptionsPerSession: 1})
 	t.Cleanup(teardown)
-	upstream := make(chan session.Stream, 1)
-	namespacePeer(t, client, func(r *session.Request) { upstream <- r.Stream })
+	asked := make(chan struct{}, 2)
+	namespacePeer(t, client, func(*session.Request) { asked <- struct{}{} }) // never answers
 	ctx, cancel := context.WithCancel(t.Context())
 	go func() {
 		_, _ = client.TrackStatus(ctx, &message.TrackStatus{Namespace: ns("video"), Name: []byte("cam1")})
 	}()
-	var s session.Stream
 	select {
-	case s = <-upstream:
+	case <-asked:
 	case <-time.After(2 * time.Second):
 		t.Fatal("the relay never forwarded the TRACK_STATUS")
 	}
 	cancel()
-	select {
-	case <-s.Context().Done():
-	case <-time.After(2 * time.Second):
-		t.Fatal("the relay's upstream TRACK_STATUS outlived the requester's cancel")
+	waitFor(t, 2*time.Second, func() bool {
+		// No publisher: DOES_NOT_EXIST at once, once the slot is free.
+		_, err := client.Subscribe(t.Context(), &message.Subscribe{Namespace: ns("audio"), Name: []byte("mic")})
+		rej, ok := errors.AsType[*session.RequestRejectedError](err)
+		return !ok || rej.Code != moqt.RequestExcessiveLoad
+	}, "the cancelled TRACK_STATUS still held the session's only subscription slot")
+}
+
+// TestTrackStatus_ConcurrentRequestsShareOneRound: TRACK_STATUS requests for
+// one track that arrive while a forwarded round is in flight wait for it
+// rather than each going upstream, and each is answered with its own
+// INCLUDE_PROPERTIES (§10.2.21).
+func TestTrackStatus_ConcurrentRequestsShareOneRound(t *testing.T) {
+	t.Parallel()
+	first, teardown := connectRelay(t, relay.Config{})
+	t.Cleanup(teardown)
+	second := dialAnotherClient(t, first)
+	release := make(chan struct{})
+	var asked atomic.Int32
+	namespacePeer(t, first, func(r *session.Request) {
+		asked.Add(1)
+		go func() {
+			<-release
+			_ = r.AcceptTrackStatus(&message.TrackStatusOK{TrackProperties: opaqueProps("shared")})
+		}()
+	})
+	type answer struct {
+		ok  *message.TrackStatusOK
+		err error
+	}
+	withProps, withoutProps := make(chan answer, 1), make(chan answer, 1)
+	go func() {
+		ok, err := trackStatusCam1(t, first)
+		withProps <- answer{ok, err}
+	}()
+	waitFor(t, 2*time.Second, func() bool { return asked.Load() == 1 }, "the first TRACK_STATUS was never forwarded")
+	go func() {
+		ok, err := trackStatusCam1(t, second, message.IncludePropertiesParam(false))
+		withoutProps <- answer{ok, err}
+	}()
+	time.Sleep(100 * time.Millisecond) // the second reaches the relay while the round is in flight
+	close(release)
+
+	a, b := <-withProps, <-withoutProps
+	if a.err != nil || b.err != nil {
+		t.Fatalf("TrackStatus: %v, %v", a.err, b.err)
+	}
+	if n := asked.Load(); n != 1 {
+		t.Fatalf("the publisher was asked %d times, want 1", n)
+	}
+	if !bytes.Equal(a.ok.TrackProperties, opaqueProps("shared")) || len(b.ok.TrackProperties) != 0 {
+		t.Fatalf(
+			"TrackProperties = %x and %x, want the publisher's and empty",
+			a.ok.TrackProperties,
+			b.ok.TrackProperties,
+		)
 	}
 }
 

From 5209e746da0d4ba08984db40c14ea6d42d820f65 Mon Sep 17 00:00:00 2001
From: Vsevolod Strukchinsky <vsevolod@strukchinsky.ru>
Date: Mon, 28 Sep 2026 10:59:44 +0500
Subject: [PATCH 3/4] fix(relay): run a shared TRACK_STATUS round as relay work
 Stop cuts short and joins

d1e8172 ran the shared round on singleflight's own goroutine, on a context
detached from any session. Relay.Stop neither ended nor joined it, so it
could return while the round was still in Discovery. Calling relayGo from
that goroutine instead could race Stop's handlers.Wait.

forwardTrackStatus now keeps its own map of rounds in flight (golang.org/x/sync
is no longer imported by the relay):
- a round starts through relayGo from the requesting handler's own
  goroutine, so Stop joins it;
- it runs on a relay-scoped context that Stop cancels first, so a round
  waiting on Discovery or an upstream is cut short;
- requesters wait for its result or their own STOP_SENDING, as before.
A test hook marks each request that has joined a round.

Tests, each red under a mutant removing the behaviour:
- TestTrackStatus_StopEndsRound: a round blocked in Discovery ends before
  Stop returns, within 2s;
- TestTrackStatus_LeaderCancelKeepsSharedRound: the requester that started
  a round cancelling does not end it for another;
- TestTrackStatus_ConcurrentRequestsShareOneRound: now waits on the hook
  rather than sleeping.

Also: the forwardTrackStatus doc no longer implies it catches a loop through
other sessions. STATUS.md records that, and that a request joining a round
late gets the candidates from its start. excessiveLoadRetryMax is named.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 STATUS.md                              |   5 +
 pkg/relay/export_test.go               |   7 ++
 pkg/relay/handler_subscribe.go         |   6 +-
 pkg/relay/handler_track_status.go      |  71 +++++++++--
 pkg/relay/relay.go                     |  24 ++--
 pkg/relay/session_handler.go           |   6 +-
 pkg/relay/track_status_forward_test.go | 165 ++++++++++++++++++++-----
 7 files changed, 225 insertions(+), 59 deletions(-)

diff --git a/STATUS.md b/STATUS.md
index 51690fc2..cf81288f 100644
--- a/STATUS.md
+++ b/STATUS.md
@@ -538,6 +538,11 @@ Relay:
   by a relay peer (§6.2 has no loop protection), so it declines the second hop
   rather than loop. Self-subscriptions are otherwise "identical" (§5.1).
 - Filters are not aggregated upstream (§6.3.1 SHOULD).
+- A shared TRACK_STATUS round asks the candidates known when it started: a
+  request joining it later is not answered by a publisher that arrived in
+  between. A TRACK_STATUS loop through different sessions (the relay's
+  request reaching it back over another one) is not detected, and its round
+  waits out its 5s bound.
 
 Open questions for interop: whether an End of Range marker carries an Object
 Payload Length (Figure 28 vs §11.4.4.2), and whether EXPIRES may appear in
diff --git a/pkg/relay/export_test.go b/pkg/relay/export_test.go
index 478a5561..564d262e 100644
--- a/pkg/relay/export_test.go
+++ b/pkg/relay/export_test.go
@@ -59,3 +59,10 @@ func SetTrackStatusTimeout(d time.Duration) (restore func()) {
 	prev := trackStatusTimeout.Swap(int64(d))
 	return func() { trackStatusTimeout.Store(prev) }
 }
+
+// SetTestHookTrackStatusJoined installs hook as testHookTrackStatusJoined and
+// returns the restore.
+func SetTestHookTrackStatusJoined(hook func(track.FullTrackName)) (restore func()) {
+	prev := testHookTrackStatusJoined.Swap(&hook)
+	return func() { testHookTrackStatusJoined.Store(prev) }
+}
diff --git a/pkg/relay/handler_subscribe.go b/pkg/relay/handler_subscribe.go
index ef6abf3f..3b9a0c06 100644
--- a/pkg/relay/handler_subscribe.go
+++ b/pkg/relay/handler_subscribe.go
@@ -577,7 +577,7 @@ func candidateRetry(err error) uint64 {
 		return goingAwayRetry
 	}
 	if errors.Is(err, session.ErrNoStreamCredit) {
-		return uint64(excessiveLoadRetry/time.Millisecond) * 3 / 2
+		return excessiveLoadRetryMax
 	}
 	rej, ok := errors.AsType[*session.RequestRejectedError](err)
 	if !ok {
@@ -1027,6 +1027,10 @@ const (
 	goingAwayRetry      = uint64(goingAwayRetryAfter/time.Millisecond) * 3 / 2
 )
 
+// excessiveLoadRetryMax is the largest Retry Interval the jitter can make of
+// excessiveLoadRetry.
+const excessiveLoadRetryMax = uint64(excessiveLoadRetry/time.Millisecond) * 3 / 2
+
 // isTrackPropertiesErr reports whether err is a Track Properties validation
 // failure from [session.Session.Subscribe] or [session.Session.Fetch]: an
 // unknown Mandatory Track Property, or Track Properties that do not parse.
diff --git a/pkg/relay/handler_track_status.go b/pkg/relay/handler_track_status.go
index e7fd199a..98146a8f 100644
--- a/pkg/relay/handler_track_status.go
+++ b/pkg/relay/handler_track_status.go
@@ -125,16 +125,46 @@ func (h *sessionHandler) rejectTrackStatus(ctx context.Context, req *session.Req
 	}
 }
 
+// testHookTrackStatusJoined, when set by a test, runs once a request has
+// joined, or started, the forwarded TRACK_STATUS round for its track.
+var testHookTrackStatusJoined atomic.Pointer[func(track.FullTrackName)]
+
+// trackStatusRounds is a relay's forwarded TRACK_STATUS rounds in flight, by
+// track; see [sessionHandler.forwardTrackStatus]. Its ctx ends when the relay
+// stops.
+type trackStatusRounds struct {
+	ctx context.Context
+	end context.CancelFunc
+
+	mu     sync.Mutex
+	rounds map[track.Key]*trackStatusRound
+}
+
+// trackStatusRound is one round's result, set before done closes.
+type trackStatusRound struct {
+	done chan struct{}
+	oks  []*message.TrackStatusOK
+	err  error
+}
+
+func newTrackStatusRounds() *trackStatusRounds {
+	ctx, end := context.WithCancel(context.Background())
+	return &trackStatusRounds{ctx: ctx, end: end, rounds: make(map[track.Key]*trackStatusRound)}
+}
+
 // forwardTrackStatus answers TRACK_STATUS for fullName from a round of
 // [sessionHandler.trackStatusUpstream], shared with the other requests for the
 // track that arrive while it is in flight (relay policy, so N requests do not
-// make N upstream ones). The round is bounded by its timeout, not by any one
-// requester, who stops waiting on its own STOP_SENDING.
+// make N upstream ones). The round runs on a relay-scoped goroutine, bounded
+// by its timeout and by the relay stopping, not by any one requester, who
+// stops waiting on its own STOP_SENDING.
 //
 // A request from a session the relay is asking about the track right now may
-// be the relay's own TRACK_STATUS routed back to it (§6.2), which the round
-// waits on: it runs a round of its own instead, where the loop guard skips
-// that session.
+// be the relay's own TRACK_STATUS routed back to it on that session (§6.2),
+// which the round waits on: it runs a round of its own instead, where the loop
+// guard skips that session. A loop through other sessions (relay A's request
+// reaching it back over a different session) is not detected; such a round
+// waits out its timeout.
 func (h *sessionHandler) forwardTrackStatus(
 	ctx context.Context,
 	req *session.Request,
@@ -143,16 +173,33 @@ func (h *sessionHandler) forwardTrackStatus(
 	waitCtx, cancel := context.WithCancel(ctx)
 	defer cancel()
 	defer context.AfterFunc(req.Stream.Context(), cancel)()
-	if h.tracks.RequestPending(message.TypeTrackStatus, h.sess, fullName.Key()) {
+	key := fullName.Key()
+	if h.tracks.RequestPending(message.TypeTrackStatus, h.sess, key) {
 		return h.trackStatusUpstream(waitCtx, fullName)
 	}
-	round := h.statusRounds.DoChan(string(fullName.Key().Bytes()), func() (any, error) {
-		return h.trackStatusUpstream(context.WithoutCancel(ctx), fullName)
-	})
+
+	rs := h.statusRounds
+	rs.mu.Lock()
+	round := rs.rounds[key]
+	if round == nil {
+		round = &trackStatusRound{done: make(chan struct{})}
+		rs.rounds[key] = round
+		// Started from this tracked handler, so Stop joins it.
+		h.relayGo(func() {
+			round.oks, round.err = h.trackStatusUpstream(rs.ctx, fullName)
+			rs.mu.Lock()
+			delete(rs.rounds, key)
+			rs.mu.Unlock()
+			close(round.done)
+		})
+	}
+	rs.mu.Unlock()
+	if hook := testHookTrackStatusJoined.Load(); hook != nil {
+		(*hook)(fullName)
+	}
 	select {
-	case r := <-round:
-		oks, _ := r.Val.([]*message.TrackStatusOK)
-		return oks, r.Err
+	case <-round.done:
+		return round.oks, round.err
 	case <-waitCtx.Done():
 		return nil, waitCtx.Err()
 	}
diff --git a/pkg/relay/relay.go b/pkg/relay/relay.go
index 680543b7..a28f5abf 100644
--- a/pkg/relay/relay.go
+++ b/pkg/relay/relay.go
@@ -14,8 +14,6 @@ import (
 	"sync"
 	"time"
 
-	"golang.org/x/sync/singleflight"
-
 	"github.com/floatdrop/moq-go/pkg/moqt"
 	"github.com/floatdrop/moq-go/pkg/moqt/message"
 	"github.com/floatdrop/moq-go/pkg/moqt/session"
@@ -303,7 +301,7 @@ type Relay struct {
 
 	// statusRounds shares forwarded TRACK_STATUS rounds across every
 	// session handler; see [sessionHandler.forwardTrackStatus].
-	statusRounds singleflight.Group
+	statusRounds *trackStatusRounds
 
 	// upstreams dials and pools relay-to-relay sessions for Discovery-driven
 	// cross-relay upstream SUBSCRIBE. nil when Config.Dialer is unset (the
@@ -400,14 +398,15 @@ func New(listener Listener, cfg Config) *Relay {
 	}
 
 	r := &Relay{
-		listener: listener,
-		cfg:      cfg,
-		log:      log.With("component", "relay"),
-		tracks:   registry.NewTrackRegistry(trackOpts...),
-		names:    registry.NewNamespaceRegistry(nameOpts...),
-		fetch:    registry.NewFetchRouter(),
-		sessions: make(map[*session.Session]struct{}),
-		stopCh:   make(chan struct{}),
+		listener:     listener,
+		cfg:          cfg,
+		log:          log.With("component", "relay"),
+		tracks:       registry.NewTrackRegistry(trackOpts...),
+		names:        registry.NewNamespaceRegistry(nameOpts...),
+		fetch:        registry.NewFetchRouter(),
+		statusRounds: newTrackStatusRounds(),
+		sessions:     make(map[*session.Session]struct{}),
+		stopCh:       make(chan struct{}),
 	}
 
 	// When a Dialer is configured, the relay can follow Discovery
@@ -648,7 +647,7 @@ func (r *Relay) serveSession(ctx context.Context, sess *session.Session, leg Leg
 
 	handler := newSessionHandler(
 		sess, r.log, r.tracks, r.names,
-		r.cfg.Authorizer, r.cfg.Metrics, leg, r.fetch, &r.statusRounds, r.upstreams,
+		r.cfg.Authorizer, r.cfg.Metrics, leg, r.fetch, r.statusRounds, r.upstreams,
 		r.cfg.Discovery, r.cfg.RelayAddr,
 		r.cfg.SendQueueSize, r.cfg.MaxDropsBeforeReset, r.cfg.MaxFanoutLag,
 		r.cfg.MaxSubscriptionsPerSession, r.cfg.MaxNamespaceRequestsPerSession,
@@ -670,6 +669,7 @@ func (r *Relay) Stop(ctx context.Context) error {
 	r.stopOnce.Do(func() {
 		r.log.LogAttrs(ctx, slog.LevelInfo, "relay stopping")
 		close(r.stopCh)
+		r.statusRounds.end() // rounds are cut short; handlers.Wait joins them
 
 		// 1. Withdraw from Discovery first, before anything else: a peer that
 		//    resolves this relay via FindTrack / FindNamespace after step 2 has
diff --git a/pkg/relay/session_handler.go b/pkg/relay/session_handler.go
index 655f5489..728e4749 100644
--- a/pkg/relay/session_handler.go
+++ b/pkg/relay/session_handler.go
@@ -12,8 +12,6 @@ import (
 	"sync/atomic"
 	"time"
 
-	"golang.org/x/sync/singleflight"
-
 	"github.com/floatdrop/moq-go/pkg/moqt"
 	"github.com/floatdrop/moq-go/pkg/moqt/message"
 	"github.com/floatdrop/moq-go/pkg/moqt/session"
@@ -45,7 +43,7 @@ type sessionHandler struct {
 	fetch   *registry.FetchRouter
 	// statusRounds shares forwarded TRACK_STATUS rounds across the relay's
 	// sessions; see [sessionHandler.forwardTrackStatus].
-	statusRounds *singleflight.Group
+	statusRounds *trackStatusRounds
 	// leg records whether this session was dialled by the relay
 	// (LegUpstream) or by the peer (LegLocal). Every [Metrics] call this
 	// handler makes carries it, so an operator can separate what the
@@ -95,7 +93,7 @@ func newSessionHandler(
 	metrics Metrics,
 	leg Leg,
 	fetch *registry.FetchRouter,
-	statusRounds *singleflight.Group,
+	statusRounds *trackStatusRounds,
 	upstreams *upstreamPool,
 	discovery discovery.DiscoveryStore,
 	relayAddr string,
diff --git a/pkg/relay/track_status_forward_test.go b/pkg/relay/track_status_forward_test.go
index 066a1726..8f8f27d1 100644
--- a/pkg/relay/track_status_forward_test.go
+++ b/pkg/relay/track_status_forward_test.go
@@ -12,6 +12,8 @@ import (
 	"github.com/floatdrop/moq-go/pkg/moqt"
 	"github.com/floatdrop/moq-go/pkg/moqt/message"
 	"github.com/floatdrop/moq-go/pkg/moqt/session"
+	"github.com/floatdrop/moq-go/pkg/moqt/track"
+	"github.com/floatdrop/moq-go/pkg/moqt/wire"
 	"github.com/floatdrop/moq-go/pkg/relay"
 	"github.com/floatdrop/moq-go/pkg/relay/discovery"
 )
@@ -337,17 +339,21 @@ func TestTrackStatus_CancelReleasesRequester(t *testing.T) {
 	}, "the cancelled TRACK_STATUS still held the session's only subscription slot")
 }
 
-// TestTrackStatus_ConcurrentRequestsShareOneRound: TRACK_STATUS requests for
-// one track that arrive while a forwarded round is in flight wait for it
-// rather than each going upstream, and each is answered with its own
-// INCLUDE_PROPERTIES (§10.2.21).
-func TestTrackStatus_ConcurrentRequestsShareOneRound(t *testing.T) {
-	t.Parallel()
+// sharedRound sets up a relay whose only candidate for video/cam1 holds its
+// answer until release is closed, and returns two requesters on it and the
+// number of times the candidate was asked. joined reports each request that
+// joins or starts the forwarded round (not parallel-safe: a package hook).
+func sharedRound(
+	t *testing.T,
+) (first, second *session.Session, release chan struct{}, asked *atomic.Int32, joined <-chan struct{}) {
+	t.Helper()
+	j := make(chan struct{}, 8)
+	t.Cleanup(relay.SetTestHookTrackStatusJoined(func(track.FullTrackName) { j <- struct{}{} }))
 	first, teardown := connectRelay(t, relay.Config{})
 	t.Cleanup(teardown)
-	second := dialAnotherClient(t, first)
-	release := make(chan struct{})
-	var asked atomic.Int32
+	second = dialAnotherClient(t, first)
+	release = make(chan struct{})
+	asked = &atomic.Int32{}
 	namespacePeer(t, first, func(r *session.Request) {
 		asked.Add(1)
 		go func() {
@@ -355,36 +361,135 @@ func TestTrackStatus_ConcurrentRequestsShareOneRound(t *testing.T) {
 			_ = r.AcceptTrackStatus(&message.TrackStatusOK{TrackProperties: opaqueProps("shared")})
 		}()
 	})
-	type answer struct {
-		ok  *message.TrackStatusOK
-		err error
+	return first, second, release, asked, j
+}
+
+func awaitJoin(t *testing.T, joined <-chan struct{}) {
+	t.Helper()
+	select {
+	case <-joined:
+	case <-time.After(2 * time.Second):
+		t.Fatal("a TRACK_STATUS never joined the forwarded round")
 	}
-	withProps, withoutProps := make(chan answer, 1), make(chan answer, 1)
-	go func() {
-		ok, err := trackStatusCam1(t, first)
-		withProps <- answer{ok, err}
-	}()
-	waitFor(t, 2*time.Second, func() bool { return asked.Load() == 1 }, "the first TRACK_STATUS was never forwarded")
+}
+
+type trackStatusAnswer struct {
+	ok  *message.TrackStatusOK
+	err error
+}
+
+func trackStatusAsync(
+	ctx context.Context,
+	t *testing.T,
+	sess *session.Session,
+	params ...message.Parameter,
+) <-chan trackStatusAnswer {
+	t.Helper()
+	out := make(chan trackStatusAnswer, 1)
 	go func() {
-		ok, err := trackStatusCam1(t, second, message.IncludePropertiesParam(false))
-		withoutProps <- answer{ok, err}
+		ts, err := sess.TrackStatus(
+			ctx,
+			&message.TrackStatus{Namespace: ns("video"), Name: []byte("cam1"), Parameters: params},
+		)
+		if err != nil {
+			out <- trackStatusAnswer{err: err}
+			return
+		}
+		_ = ts.Close()
+		out <- trackStatusAnswer{ok: ts.OK}
 	}()
-	time.Sleep(100 * time.Millisecond) // the second reaches the relay while the round is in flight
+	return out
+}
+
+// TestTrackStatus_ConcurrentRequestsShareOneRound: TRACK_STATUS requests for
+// one track that arrive while a forwarded round is in flight wait for it
+// rather than each going upstream, and each is answered with its own
+// INCLUDE_PROPERTIES (§10.2.21).
+func TestTrackStatus_ConcurrentRequestsShareOneRound(t *testing.T) {
+	first, second, release, asked, joined := sharedRound(t)
+	a := trackStatusAsync(t.Context(), t, first)
+	awaitJoin(t, joined)
+	b := trackStatusAsync(t.Context(), t, second, message.IncludePropertiesParam(false))
+	awaitJoin(t, joined)
 	close(release)
 
-	a, b := <-withProps, <-withoutProps
-	if a.err != nil || b.err != nil {
-		t.Fatalf("TrackStatus: %v, %v", a.err, b.err)
+	withProps, withoutProps := <-a, <-b
+	if withProps.err != nil || withoutProps.err != nil {
+		t.Fatalf("TrackStatus: %v, %v", withProps.err, withoutProps.err)
 	}
 	if n := asked.Load(); n != 1 {
 		t.Fatalf("the publisher was asked %d times, want 1", n)
 	}
-	if !bytes.Equal(a.ok.TrackProperties, opaqueProps("shared")) || len(b.ok.TrackProperties) != 0 {
-		t.Fatalf(
-			"TrackProperties = %x and %x, want the publisher's and empty",
-			a.ok.TrackProperties,
-			b.ok.TrackProperties,
-		)
+	if !bytes.Equal(withProps.ok.TrackProperties, opaqueProps("shared")) || len(withoutProps.ok.TrackProperties) != 0 {
+		t.Fatalf("TrackProperties = %x and %x, want the publisher's and empty",
+			withProps.ok.TrackProperties, withoutProps.ok.TrackProperties)
+	}
+}
+
+// TestTrackStatus_LeaderCancelKeepsSharedRound: the requester whose request
+// started a round cancelling it (§3.3.3) does not end the round for another
+// requester sharing it.
+func TestTrackStatus_LeaderCancelKeepsSharedRound(t *testing.T) {
+	first, second, release, _, joined := sharedRound(t)
+	ctx, cancel := context.WithCancel(t.Context())
+	leader := trackStatusAsync(ctx, t, first)
+	awaitJoin(t, joined)
+	follower := trackStatusAsync(t.Context(), t, second)
+	awaitJoin(t, joined)
+	cancel()
+	<-leader
+	close(release)
+	if got := <-follower; got.err != nil {
+		t.Fatalf("the follower's TrackStatus failed after the leader cancelled: %v", got.err)
+	}
+}
+
+// blockingDiscovery is a Discovery store whose FindNamespace blocks until its
+// ctx ends, recording that it returned.
+type blockingDiscovery struct {
+	discovery.DiscoveryStore
+
+	entered  chan struct{}
+	returned atomic.Bool
+}
+
+func (b *blockingDiscovery) FindNamespace(
+	ctx context.Context,
+	_ wire.TrackNamespace,
+) ([]discovery.NamespaceInfo, error) {
+	select {
+	case b.entered <- struct{}{}:
+	default:
+	}
+	<-ctx.Done()
+	b.returned.Store(true)
+	return nil, ctx.Err()
+}
+
+// TestTrackStatus_StopEndsRound: a forwarded round is relay work, so Stop cuts
+// it short and joins it, rather than returning while it still reaches into
+// Discovery.
+func TestTrackStatus_StopEndsRound(t *testing.T) {
+	t.Parallel()
+	mem := discovery.NewMemoryStore()
+	defer mem.Close()
+	store := &blockingDiscovery{DiscoveryStore: mem, entered: make(chan struct{}, 1)}
+	tr := startTestRelay(t.Context(), relay.Config{
+		Discovery: store, RelayAddr: "relay-A", Dialer: dialerTo(nil),
+	})
+	go func() { _, _ = trackStatusCam1(t, dialClient(t, tr)) }()
+	select {
+	case <-store.entered:
+	case <-time.After(2 * time.Second):
+		t.Fatal("the round never resolved Discovery candidates")
+	}
+	start := time.Now()
+	tr.stop(t)
+	if !store.returned.Load() {
+		t.Fatal("Stop returned while the forwarded round was still in Discovery")
+	}
+	if d := time.Since(start); d > 2*time.Second {
+		t.Fatalf("Stop took %v: it waited the round out instead of cutting it short", d)
 	}
 }
 

From c7b434f5f8c4a87b5c964b0d02d8d4e9f88d6993 Mon Sep 17 00:00:00 2001
From: Vsevolod Strukchinsky <vsevolod@strukchinsky.ru>
Date: Mon, 28 Sep 2026 11:05:37 +0500
Subject: [PATCH 4/4] =?UTF-8?q?fix(relay):=20end=20forwarded=20TRACK=5FSTA?=
 =?UTF-8?q?TUS=20rounds=20with=20the=20upstream=20sessions,=20after=20the?=
 =?UTF-8?q?=20grace=20period=20(=C2=A73.6)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

5209e74 cancelled the relay's TRACK_STATUS rounds as soon as Stop began,
before the GOAWAY and its grace period. A request in flight then got
DOES_NOT_EXIST with no retry, though its publisher answered within the
grace. Stop drops upstream work only at step 6, after the downstream GOAWAY
and grace ("prior to unsubscribing from upstream publishers", §3.6). The
rounds now end there too, next to the upstream pool, and handlers.Wait still
joins them.

TestTrackStatus_StopKeepsRoundThroughGrace (1s grace, publisher answering
after 200ms, Stop begun mid-round) failed before this change.
TestTrackStatus_StopEndsRound still holds.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 pkg/relay/relay.go                     |  4 +++-
 pkg/relay/track_status_forward_test.go | 33 ++++++++++++++++++++++++++
 2 files changed, 36 insertions(+), 1 deletion(-)

diff --git a/pkg/relay/relay.go b/pkg/relay/relay.go
index a28f5abf..36987d2c 100644
--- a/pkg/relay/relay.go
+++ b/pkg/relay/relay.go
@@ -669,7 +669,6 @@ func (r *Relay) Stop(ctx context.Context) error {
 	r.stopOnce.Do(func() {
 		r.log.LogAttrs(ctx, slog.LevelInfo, "relay stopping")
 		close(r.stopCh)
-		r.statusRounds.end() // rounds are cut short; handlers.Wait joins them
 
 		// 1. Withdraw from Discovery first, before anything else: a peer that
 		//    resolves this relay via FindTrack / FindNamespace after step 2 has
@@ -764,6 +763,9 @@ func (r *Relay) Stop(ctx context.Context) error {
 		if r.upstreams != nil {
 			r.upstreams.close()
 		}
+		// Forwarded TRACK_STATUS rounds are upstream work too: cut short
+		// here, joined by handlers.Wait below.
+		r.statusRounds.end()
 
 		// 7. Force-close anything still standing, with GOAWAY_TIMEOUT
 		//    only where it is true: "the peer took too long to close the
diff --git a/pkg/relay/track_status_forward_test.go b/pkg/relay/track_status_forward_test.go
index 8f8f27d1..817eff1b 100644
--- a/pkg/relay/track_status_forward_test.go
+++ b/pkg/relay/track_status_forward_test.go
@@ -511,3 +511,36 @@ func TestTrackStatus_ForwardingCountsAgainstSubscriptionCap(t *testing.T) {
 	_, err := trackStatusCam1(t, client)
 	requireRejectedWithCode(t, err, moqt.RequestExcessiveLoad)
 }
+
+// TestTrackStatus_StopKeepsRoundThroughGrace: Stop drops upstream work only
+// after the downstream GOAWAY's grace period (§3.6), so a forwarded round in
+// flight when Stop begins still delivers a publisher's answer given within it.
+func TestTrackStatus_StopKeepsRoundThroughGrace(t *testing.T) {
+	t.Parallel()
+	tr := startTestRelay(t.Context(), relay.Config{GoawayTimeout: time.Second})
+	pub := dialClient(t, tr)
+	asked := make(chan struct{}, 1)
+	answerTrackStatus(t, pub, func(r *session.Request) {
+		asked <- struct{}{}
+		go func() {
+			time.Sleep(200 * time.Millisecond)
+			_ = r.AcceptTrackStatus(&message.TrackStatusOK{TrackProperties: opaqueProps("late")})
+		}()
+	})
+	publishNS(t, pub, "video")
+	answer := trackStatusAsync(t.Context(), t, dialClient(t, tr))
+	select {
+	case <-asked:
+	case <-time.After(2 * time.Second):
+		t.Fatal("the relay never forwarded the TRACK_STATUS")
+	}
+	stopped := make(chan struct{})
+	go func() {
+		defer close(stopped)
+		_ = tr.r.Stop(context.Background())
+	}()
+	if got := <-answer; got.err != nil {
+		t.Fatalf("TrackStatus during Stop's grace period: %v", got.err)
+	}
+	<-stopped
+}