From 0d17c0dee531e5183b52aafc8a6042632f2909a9 Mon Sep 17 00:00:00 2001 From: Vsevolod Strukchinsky Date: Sat, 26 Sep 2026 11:23:09 +0500 Subject: [PATCH 1/3] fix(relay): forward a later upstream's Object Properties on a merged Subgroup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit With several upstreams feeding one Subgroup (§9.3), each subscriber's downstream stream copies the first contributor's SUBGROUP_HEADER. When that header had PROPERTIES clear (§11.4.2), a later contributor's Object Properties were silently left out on write, though a relay MUST forward them (§2.5). The writer now tracks whether its streams set PROPERTIES. An Object with Properties on a stream without the bit turns it on: a stream not yet opened just opens with it; an open one is reset and reopened with it (reopen on demand, the design chosen with the user). Later streams keep the bit. New metrics cause relay.ResetCauseProperties ("properties"), appended to the enum. TestFanout_MultiPublisher_ForwardsEveryContributorsProperties was verified red on the unpatched relay ("first without, second with"); the other two cases guard the paths that already worked. Co-Authored-By: Claude Opus 5.5 (1M context) --- STATUS.md | 2 +- pkg/relay/handler_fanout.go | 40 +++++++- pkg/relay/handler_fanout_multipub_test.go | 112 ++++++++++++++++++++++ pkg/relay/metrics.go | 8 ++ pkg/relay/metrics_test.go | 2 + 5 files changed, 159 insertions(+), 5 deletions(-) diff --git a/STATUS.md b/STATUS.md index ff64fde5..0e41f448 100644 --- a/STATUS.md +++ b/STATUS.md @@ -144,7 +144,7 @@ By package, bottom-up along the dependency stack: |-------|--------------------------------------|--------|-------| | 9.1 | Caching relays | DONE | LRU+TTL object cache (`cache/cache.go`); updates limited to non-existence/properties. | | 9.2 | Forward handling | DONE | FORWARD flag honoured; Forward=0 pauses delivery. Upstream Forward is set to 1 only when a downstream subscriber forwards, else the relay pauses it (Forward=0) and resumes on the first forwarding subscriber. | -| 9.3 | Multiple publishers | DONE | Per-track upstreams; dedup by `{GroupID, ObjectID}`. | +| 9.3 | Multiple publishers | DONE | Per-track upstreams; dedup by `{GroupID, ObjectID}`. Upstreams of one Subgroup share one downstream stream per subscriber, with the first one's SUBGROUP_HEADER; a later one's Object Properties reopen it with PROPERTIES set, so none are dropped (§2.5). | | 9.4 | Subscriber interactions | DONE | Upstream subscription established before SUBSCRIBE_OK; aggregation. | | 9.4.1 | Graceful subscriber switchover | DONE | GOAWAY grace period (`GoawayTimeout`). | | 9.5 | Publisher interactions | DONE | PUBLISH_NAMESPACE / PUBLISH with prefix matching (`namespace_registry.go`). | diff --git a/pkg/relay/handler_fanout.go b/pkg/relay/handler_fanout.go index 65648356..33a86902 100644 --- a/pkg/relay/handler_fanout.go +++ b/pkg/relay/handler_fanout.go @@ -513,6 +513,9 @@ type subgroupWriter struct { // past the filtered Objects read after it; zero once the relay drops // one. Only touched by admit and publish, under sg.Mu. lastPos inboundPos + // withProps: the streams run opens set PROPERTIES (§11.4.2). Only + // touched by run. + withProps bool } // admit decides whether w takes the Object at objectID of the subgroup hdr @@ -682,9 +685,11 @@ func (w *subgroupWriter) run() { // next is not known to follow it. dropped bool ) + w.withProps = w.hdr.Properties // reopen resets the current outbound stream (if any) and opens a fresh - // one, for the lazy first open and after a §11.4.3 gap. first sets the + // one, for the lazy first open, after a §11.4.3 gap, and to carry Object + // Properties the old header could not. first sets the // §11.4.2 FIRST_OBJECT bit; otherwise the stream is a replay. All its // blocking I/O is bounded by w.ctx. reopen := func(first bool) bool { @@ -694,6 +699,7 @@ func (w *subgroupWriter) run() { } w.closeOut(false, w.resetCode()) hdr := w.hdr + hdr.Properties = w.withProps hdr.ReplayingSubgroup = !first if !first && hdr.SubgroupIDMode == message.SubgroupIDImplicitFirstObject { // A replay stream's first object would imply the wrong ID. @@ -751,6 +757,8 @@ func (w *subgroupWriter) run() { continue } + cause, stale := w.reopenCause(fwd, prevID, hasWritten, dropped) + // Lazy first open, off sg.Mu (see openWriterForSub). if w.out == nil { if !reopen(fwd.first) { @@ -759,9 +767,8 @@ func (w *subgroupWriter) run() { } } - // §11.4.3: only "the next Object" may go on an existing stream. - if hasWritten && !isNextObject(fwd, prevID, dropped) { - w.metrics.SubgroupStreamReset(w.ref, w.hdr.SubgroupID, ResetCauseGap) + if stale { + w.metrics.SubgroupStreamReset(w.ref, w.hdr.SubgroupID, cause) if !reopen(fwd.first) { failWrites() continue @@ -870,6 +877,31 @@ func (w *subgroupWriter) run() { w.closeOut(true, 0) } +// reopenCause reports whether fwd needs a fresh outbound stream after one +// whose last Object is prevID, and why. §11.4.3: only "the next Object" may go +// on an existing stream. And the header is the first contributor's (§9.3), so +// a later one's Object Properties, which MUST be forwarded (§2.5), turn +// PROPERTIES on for this and every later stream. +func (w *subgroupWriter) reopenCause( + fwd fwdObject, + prevID uint64, + hasWritten, dropped bool, +) (ResetCause, bool) { + needProps := !w.withProps && len(fwd.obj.Properties) > 0 + if needProps { + w.withProps = true + } + switch { + case !hasWritten: + return 0, false + case needProps: + return ResetCauseProperties, true + case !isNextObject(fwd, prevID, dropped): + return ResetCauseGap, true + } + return 0, false +} + // isNextObject reports whether fwd is "the next Object" (§11.4.3) on a stream // whose last Object is prevID. Of the draft's ways to tell, the relay uses: // the Object ID is one greater; fwd follows the last Object on its inbound diff --git a/pkg/relay/handler_fanout_multipub_test.go b/pkg/relay/handler_fanout_multipub_test.go index 326a6586..c41b76ce 100644 --- a/pkg/relay/handler_fanout_multipub_test.go +++ b/pkg/relay/handler_fanout_multipub_test.go @@ -1,6 +1,7 @@ package relay_test import ( + "bytes" "context" "errors" "io" @@ -12,6 +13,7 @@ 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/wire" "github.com/floatdrop/moq-go/pkg/relay" ) @@ -426,3 +428,113 @@ func awaitStreamEnd(t *testing.T, events <-chan objEvent) objEvent { return objEvent{} } } + +// TestFanout_MultiPublisher_ForwardsEveryContributorsProperties: Object +// Properties MUST be forwarded (§2.5), whichever contributor's SUBGROUP_HEADER +// set the merged stream's PROPERTIES bit (§11.4.2). +func TestFanout_MultiPublisher_ForwardsEveryContributorsProperties(t *testing.T) { + t.Parallel() + props := message.AppendTrackProperties([]wire.KVPair{{Type: 0x40, IntVal: 7}}) + for _, tc := range []struct { + name string + // first and second: whether each contributor's header has PROPERTIES; + // the second writes Object 1 with props. + first, second bool + reopens int // ResetCauseProperties reopens + }{ + {"first without, second with", false, true, 1}, + {"first with, second with", true, true, 0}, + {"first with, second without", true, false, 0}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + rec := &recordingMetrics{} + pubA, teardown := connectRelay(t, relay.Config{Metrics: rec}) + defer teardown() + pubB := dialAnotherClient(t, pubA) + subSess := dialAnotherClient(t, pubA) + aPub := publishVideoTrack(t, pubA, "cam1", 1) + bPub := publishVideoTrack(t, pubB, "cam1", 2) + subscribeCam1(t, subSess) + + type received struct { + id uint64 + props []byte + } + got := make(chan received, 8) + go func() { + for { + ds, err := subSess.AcceptDataStream(t.Context()) + if err != nil { + return + } + sg, ok := ds.(*session.IncomingSubgroupStream) + if !ok { + return + } + go func() { + for { + o, err := sg.ReadDecoded() + if err != nil { + return + } + got <- received{o.ObjectID, o.Properties} + } + }() + } + }() + await := func(id uint64) received { + t.Helper() + select { + case r := <-got: + if r.id != id { + t.Fatalf("received Object %d, want %d", r.id, id) + } + return r + case <-time.After(2 * time.Second): + t.Fatalf("Object %d not forwarded", id) + return received{} + } + } + + hdr := message.SubgroupHeader{SubgroupIDMode: message.SubgroupIDExplicit} + aHdr, bHdr := hdr, hdr + aHdr.Properties, bHdr.Properties = tc.first, tc.second + a, err := aPub.OpenSubgroup(aHdr) + if err != nil { + t.Fatalf("A OpenSubgroup: %v", err) + } + if err := a.WriteObjectAt(0, &message.SubgroupObject{Payload: []byte("a")}); err != nil { + t.Fatalf("A WriteObjectAt 0: %v", err) + } + await(0) // A's header is now the merged stream's + b, err := bPub.OpenSubgroup(bHdr) + if err != nil { + t.Fatalf("B OpenSubgroup: %v", err) + } + var bProps []byte + if tc.second { + bProps = props + } + if err := b.WriteObjectAt( + 1, + &message.SubgroupObject{Properties: bProps, Payload: []byte("b")}, + ); err != nil { + t.Fatalf("B WriteObjectAt 1: %v", err) + } + if r := await(1); !bytes.Equal(r.props, bProps) { + t.Fatalf("Object 1 Properties = %x, want %x", r.props, bProps) + } + // A contributor without the bit still reaches the subscriber. + if err := a.WriteObjectAt(2, &message.SubgroupObject{Payload: []byte("a")}); err != nil { + t.Fatalf("A WriteObjectAt 2: %v", err) + } + if r := await(2); len(r.props) != 0 { + t.Fatalf("Object 2 Properties = %x, want none", r.props) + } + if got := rec.resetCount(relay.ResetCauseProperties); got != tc.reopens { + t.Fatalf("properties reopens = %d, want %d", got, tc.reopens) + } + }) + } +} diff --git a/pkg/relay/metrics.go b/pkg/relay/metrics.go index d6f45ac3..ef739640 100644 --- a/pkg/relay/metrics.go +++ b/pkg/relay/metrics.go @@ -105,6 +105,12 @@ const ( // stream — the subscriber's session is in trouble, not the relay's // scheduling. ResetCauseWriteError + + // ResetCauseProperties is a reopen to carry Object Properties (§2.5): + // the outbound stream's SUBGROUP_HEADER had PROPERTIES clear (§11.4.2), + // taken from another upstream of the same Subgroup (§9.3), so a fresh + // stream with the bit set was opened. No Object is lost. + ResetCauseProperties ) // String returns a stable, lowercase name suitable for use as a metric label @@ -123,6 +129,8 @@ func (c ResetCause) String() string { return "inbound_reset" case ResetCauseWriteError: return "write_error" + case ResetCauseProperties: + return "properties" default: return "unknown" } diff --git a/pkg/relay/metrics_test.go b/pkg/relay/metrics_test.go index a6c2908f..cd067452 100644 --- a/pkg/relay/metrics_test.go +++ b/pkg/relay/metrics_test.go @@ -231,6 +231,7 @@ func TestMetricsHooks(t *testing.T) { relay.ResetCauseDeliveryTimeout, relay.ResetCauseWriteError, relay.ResetCauseInboundReset, + relay.ResetCauseProperties, } { if got := rec.resetCount(c); got != 0 { t.Errorf("SubgroupStreamReset(%s) = %d on a clean subgroup, want 0", c, got) @@ -321,6 +322,7 @@ func TestResetCauseString(t *testing.T) { relay.ResetCauseExcessiveLoad: "excessive_load", relay.ResetCauseInboundReset: "inbound_reset", relay.ResetCauseWriteError: "write_error", + relay.ResetCauseProperties: "properties", relay.ResetCause(99): "unknown", } { if got := cause.String(); got != want { From 492608383129c1d0fc961c1d8a802d4c13b7892b Mon Sep 17 00:00:00 2001 From: Vsevolod Strukchinsky Date: Sat, 26 Sep 2026 11:27:37 +0500 Subject: [PATCH 2/3] fix(relay): no FIRST_OBJECT on a reopened stream after an Object was sent MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review follow-up. A properties reopen (or a gap reopen) for an Object that was its contributor's first could open the new stream with FIRST_OBJECT set (§11.4.2), though another contributor's Object had gone out before it. reopen now clears it once the writer sent any Object. Also: the ResetCauseProperties doc no longer claims no Object is lost; written Objects survive the reset only where RESET_STREAM_AT is in use, noted in STATUS.md with the always-set-PROPERTIES alternative. The FIRST_OBJECT assertion added to TestFanout_MultiPublisher_ForwardsEveryContributorsProperties was verified red before the fix. Co-Authored-By: Claude Opus 5.5 (1M context) --- STATUS.md | 2 +- pkg/relay/handler_fanout.go | 13 ++++++++++--- pkg/relay/handler_fanout_multipub_test.go | 20 ++++++++++++++------ pkg/relay/metrics.go | 5 ++++- 4 files changed, 29 insertions(+), 11 deletions(-) diff --git a/STATUS.md b/STATUS.md index 0e41f448..71d81bf1 100644 --- a/STATUS.md +++ b/STATUS.md @@ -144,7 +144,7 @@ By package, bottom-up along the dependency stack: |-------|--------------------------------------|--------|-------| | 9.1 | Caching relays | DONE | LRU+TTL object cache (`cache/cache.go`); updates limited to non-existence/properties. | | 9.2 | Forward handling | DONE | FORWARD flag honoured; Forward=0 pauses delivery. Upstream Forward is set to 1 only when a downstream subscriber forwards, else the relay pauses it (Forward=0) and resumes on the first forwarding subscriber. | -| 9.3 | Multiple publishers | DONE | Per-track upstreams; dedup by `{GroupID, ObjectID}`. Upstreams of one Subgroup share one downstream stream per subscriber, with the first one's SUBGROUP_HEADER; a later one's Object Properties reopen it with PROPERTIES set, so none are dropped (§2.5). | +| 9.3 | Multiple publishers | DONE | Per-track upstreams; dedup by `{GroupID, ObjectID}`. Upstreams of one Subgroup share one downstream stream per subscriber, with the first one's SUBGROUP_HEADER; a later one's Object Properties reopen it with PROPERTIES set, so none are dropped (§2.5). Like a §11.4.3 gap reopen, the reset keeps already-written Objects only where RESET_STREAM_AT is in use; always setting PROPERTIES would avoid it at a byte per Object. | | 9.4 | Subscriber interactions | DONE | Upstream subscription established before SUBSCRIBE_OK; aggregation. | | 9.4.1 | Graceful subscriber switchover | DONE | GOAWAY grace period (`GoawayTimeout`). | | 9.5 | Publisher interactions | DONE | PUBLISH_NAMESPACE / PUBLISH with prefix matching (`namespace_registry.go`). | diff --git a/pkg/relay/handler_fanout.go b/pkg/relay/handler_fanout.go index 33a86902..45d36859 100644 --- a/pkg/relay/handler_fanout.go +++ b/pkg/relay/handler_fanout.go @@ -684,15 +684,21 @@ func (w *subgroupWriter) run() { // dropped: an Object was dropped since the last one written, so the // next is not known to follow it. dropped bool + // sentAny: an Object was written on some stream of this writer, so no + // later stream begins with the Subgroup's first Object. + sentAny bool ) w.withProps = w.hdr.Properties // reopen resets the current outbound stream (if any) and opens a fresh // one, for the lazy first open, after a §11.4.3 gap, and to carry Object - // Properties the old header could not. first sets the - // §11.4.2 FIRST_OBJECT bit; otherwise the stream is a replay. All its - // blocking I/O is bounded by w.ctx. + // Properties the old header could not. first sets the §11.4.2 + // FIRST_OBJECT bit, unless an Object was already sent; otherwise the + // stream is a replay. All its blocking I/O is bounded by w.ctx. reopen := func(first bool) bool { + // A contributor's first Object is not the Subgroup's once another + // contributor's went out (§9.3). + first = first && !sentAny if w.unbridge != nil { w.unbridge() w.unbridge = nil @@ -814,6 +820,7 @@ func (w *subgroupWriter) run() { } prevID = fwd.absID hasWritten = true + sentAny = true dropped = false // §11.4.3: a later reset still delivers what was written. w.out.MarkReliable() diff --git a/pkg/relay/handler_fanout_multipub_test.go b/pkg/relay/handler_fanout_multipub_test.go index c41b76ce..b776bf61 100644 --- a/pkg/relay/handler_fanout_multipub_test.go +++ b/pkg/relay/handler_fanout_multipub_test.go @@ -458,12 +458,14 @@ func TestFanout_MultiPublisher_ForwardsEveryContributorsProperties(t *testing.T) subscribeCam1(t, subSess) type received struct { - id uint64 - props []byte + id uint64 + props []byte + stream int // 1-based index of the stream it came on + replay bool // its stream's header had FIRST_OBJECT clear } got := make(chan received, 8) go func() { - for { + for stream := 1; ; stream++ { ds, err := subSess.AcceptDataStream(t.Context()) if err != nil { return @@ -478,7 +480,7 @@ func TestFanout_MultiPublisher_ForwardsEveryContributorsProperties(t *testing.T) if err != nil { return } - got <- received{o.ObjectID, o.Properties} + got <- received{o.ObjectID, o.Properties, stream, sg.Header.ReplayingSubgroup} } }() } @@ -507,7 +509,7 @@ func TestFanout_MultiPublisher_ForwardsEveryContributorsProperties(t *testing.T) if err := a.WriteObjectAt(0, &message.SubgroupObject{Payload: []byte("a")}); err != nil { t.Fatalf("A WriteObjectAt 0: %v", err) } - await(0) // A's header is now the merged stream's + r0 := await(0) // A's header is now the merged stream's b, err := bPub.OpenSubgroup(bHdr) if err != nil { t.Fatalf("B OpenSubgroup: %v", err) @@ -522,9 +524,15 @@ func TestFanout_MultiPublisher_ForwardsEveryContributorsProperties(t *testing.T) ); err != nil { t.Fatalf("B WriteObjectAt 1: %v", err) } - if r := await(1); !bytes.Equal(r.props, bProps) { + r := await(1) + if !bytes.Equal(r.props, bProps) { t.Fatalf("Object 1 Properties = %x, want %x", r.props, bProps) } + // B's header claims FIRST_OBJECT, but a stream beginning with Object 1 + // is mid-Subgroup (§11.4.2): Object 0 went out first. + if r.stream != r0.stream && !r.replay { + t.Fatal("Object 1's new stream sets FIRST_OBJECT, though Object 0 was sent before it") + } // A contributor without the bit still reaches the subscriber. if err := a.WriteObjectAt(2, &message.SubgroupObject{Payload: []byte("a")}); err != nil { t.Fatalf("A WriteObjectAt 2: %v", err) diff --git a/pkg/relay/metrics.go b/pkg/relay/metrics.go index ef739640..dee63a58 100644 --- a/pkg/relay/metrics.go +++ b/pkg/relay/metrics.go @@ -109,7 +109,10 @@ const ( // ResetCauseProperties is a reopen to carry Object Properties (§2.5): // the outbound stream's SUBGROUP_HEADER had PROPERTIES clear (§11.4.2), // taken from another upstream of the same Subgroup (§9.3), so a fresh - // stream with the bit set was opened. No Object is lost. + // stream with the bit set was opened. Objects already written survive + // the reset only where RESET_STREAM_AT is in use (see + // [session.OutgoingSubgroupStream.MarkReliable]); elsewhere the unacked + // ones are lost, as with [ResetCauseGap]. ResetCauseProperties ) From c85a067b863d71625e439f5a712b39f2155e7dd6 Mon Sep 17 00:00:00 2001 From: Vsevolod Strukchinsky Date: Sat, 26 Sep 2026 11:36:04 +0500 Subject: [PATCH 3/3] fix(relay): a FIRST_OBJECT claim holds unless a lower Object ID was forwarded MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review follow-up. The writer-local sentAny counted Objects sent to one subscriber, but FIRST_OBJECT (§11.4.2) is about the first Object ever published in the Subgroup (§2.2): a subscriber whose filter hid Object 0 still got a FIRST_OBJECT stream beginning at another contributor's Object 1. And any earlier Object is not a contradiction, only a lower one: Objects are published in ascending order (§2.2), so a replay starting at Object 5 must not suppress the true Object 0's bit. The Subgroup's writer set now records the lowest Object ID forwarded, under sg.Mu (claimFirst), and fwdObject.first honours a contributor's claim only below it; sentAny is gone. Limitation, documented on the field: the state lives as long as the set, so after every contributor left a new one's claim is not checked against earlier Objects. TestFanout_MultiPublisher_FirstObjectOnlyForSubgroupsFirst was verified red for each case against the code before its fix; removing the check fails it and the properties test's FIRST_OBJECT assertion. Co-Authored-By: Claude Opus 5.5 (1M context) --- pkg/relay/handler_fanout.go | 34 +++++-- pkg/relay/handler_fanout_multipub_test.go | 113 ++++++++++++++++++++++ 2 files changed, 137 insertions(+), 10 deletions(-) diff --git a/pkg/relay/handler_fanout.go b/pkg/relay/handler_fanout.go index 45d36859..35d23114 100644 --- a/pkg/relay/handler_fanout.go +++ b/pkg/relay/handler_fanout.go @@ -63,6 +63,14 @@ type subgroupWriterSet struct { // skipped while it is unchanged. gen uint64 + // lowest is the lowest Object ID forwarded, if forwarded. Objects are + // published in ascending order (§2.2), so a FIRST_OBJECT claim (§11.4.2) + // for a higher ID is wrong, whether or not a given subscriber got the + // lower one. It lives only as long as the set: after every contributor + // left, a new one's claim is not checked against earlier Objects. + lowest uint64 + forwarded bool + // sawClean records that some contributor ended cleanly, so the merged // stream FINs even if a peer reset; resetCode is used only when every // contributor reset. @@ -70,6 +78,17 @@ type subgroupWriterSet struct { resetCode moqt.StreamResetCode } +// claimFirst records that the Object at objectID is forwarded and reports +// whether it starts the Subgroup: its contributor claims so (claimed) and no +// lower ID was forwarded (see subgroupWriterSet.lowest). Callers hold sg.Mu. +func (s *subgroupWriterSet) claimFirst(objectID uint64, claimed bool) bool { + lowest := !s.forwarded || objectID < s.lowest + if lowest { + s.lowest, s.forwarded = objectID, true + } + return claimed && lowest +} + // resolveImplicitSubgroupID handles §11.4.2 SUBGROUP_ID_MODE 0b01 (Subgroup ID // = first Object ID): it reads the first object and rewrites hdr to the // explicit form. The returned pending object must be processed as the @@ -380,6 +399,8 @@ func (h *sessionHandler) runFanout(ctx context.Context, stream *session.Incoming h.openWriterForSub(ctx, set.hdr, sub, set.writers, entry.DeliveryTimeouts(), ref) } + first := set.claimFirst(objectID, isTrueFirst) + // §5.1.2 filters run before enqueue, so a miss takes no queue slot. for _, w := range set.writers { if w == nil { @@ -389,7 +410,7 @@ func (h *sessionHandler) runFanout(ctx context.Context, stream *session.Incoming w.publish(fwdObject{ obj: obj, absID: objectID, - first: isTrueFirst, + first: first, maxCacheAge: liveMaxAge, follows: follows, }) @@ -684,21 +705,15 @@ func (w *subgroupWriter) run() { // dropped: an Object was dropped since the last one written, so the // next is not known to follow it. dropped bool - // sentAny: an Object was written on some stream of this writer, so no - // later stream begins with the Subgroup's first Object. - sentAny bool ) w.withProps = w.hdr.Properties // reopen resets the current outbound stream (if any) and opens a fresh // one, for the lazy first open, after a §11.4.3 gap, and to carry Object // Properties the old header could not. first sets the §11.4.2 - // FIRST_OBJECT bit, unless an Object was already sent; otherwise the - // stream is a replay. All its blocking I/O is bounded by w.ctx. + // FIRST_OBJECT bit; otherwise the stream is a replay. All its blocking + // I/O is bounded by w.ctx. reopen := func(first bool) bool { - // A contributor's first Object is not the Subgroup's once another - // contributor's went out (§9.3). - first = first && !sentAny if w.unbridge != nil { w.unbridge() w.unbridge = nil @@ -820,7 +835,6 @@ func (w *subgroupWriter) run() { } prevID = fwd.absID hasWritten = true - sentAny = true dropped = false // §11.4.3: a later reset still delivers what was written. w.out.MarkReliable() diff --git a/pkg/relay/handler_fanout_multipub_test.go b/pkg/relay/handler_fanout_multipub_test.go index b776bf61..499c6e1a 100644 --- a/pkg/relay/handler_fanout_multipub_test.go +++ b/pkg/relay/handler_fanout_multipub_test.go @@ -546,3 +546,116 @@ func TestFanout_MultiPublisher_ForwardsEveryContributorsProperties(t *testing.T) }) } } + +// streamHeaders emits the header of each subgroup stream sess accepts, and +// drains the stream so the relay can open the next. +func streamHeaders(t *testing.T, sess *session.Session) <-chan message.SubgroupHeader { + ch := make(chan message.SubgroupHeader, 4) + go func() { + for { + ds, err := sess.AcceptDataStream(t.Context()) + if err != nil { + return + } + sg, ok := ds.(*session.IncomingSubgroupStream) + if !ok { + return + } + select { + case ch <- sg.Header: + case <-t.Context().Done(): + return + } + go func() { + for { + if _, err := sg.ReadObject(); err != nil { + return + } + } + }() + } + }() + return ch +} + +// awaitHeader waits for the next header from [streamHeaders]. +func awaitHeader(t *testing.T, ch <-chan message.SubgroupHeader) message.SubgroupHeader { + t.Helper() + select { + case h := <-ch: + return h + case <-time.After(2 * time.Second): + t.Fatal("no subgroup stream forwarded") + return message.SubgroupHeader{} + } +} + +// TestFanout_MultiPublisher_FirstObjectOnlyForSubgroupsFirst: Objects are +// published in ascending ID order (§2.2), so a contributor's FIRST_OBJECT +// claim holds unless an Object with a lower ID was forwarded (§11.4.2, §2.2), +// whether or not a given subscriber got it. +func TestFanout_MultiPublisher_FirstObjectOnlyForSubgroupsFirst(t *testing.T) { + t.Parallel() + for _, tc := range []struct { + name string + // A writes aID first, then B, whose header claims FIRST_OBJECT, + // writes bID; the subscriber's stream beginning with bID must have + // FIRST_OBJECT clear iff replay. + aID, bID uint64 + aReplay bool + filter *message.RangeFilter // the subscriber's, hiding A's Object + replay bool + }{ + { + name: "lower Object forwarded first, filtered out", + aID: 0, bID: 1, + filter: &message.RangeFilter{ + Type: message.ParamObjectIDFilter, Ranges: []message.Range{{Start: 1, End: 1}}, + }, + replay: true, + }, + {name: "higher Object forwarded first", aID: 5, bID: 0, aReplay: true, replay: false}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + pubA, teardown := connectRelay(t, relay.Config{}) + defer teardown() + pubB := dialAnotherClient(t, pubA) + aPub := publishVideoTrack(t, pubA, "cam1", 1) + bPub := publishVideoTrack(t, pubB, "cam1", 2) + witness := newCam1Subscriber(t, pubA) // sees A's Object reach the relay + var params []message.Parameter + if tc.filter != nil { + params = append(params, message.RangeFilterParam(tc.filter)) + } + sub := newCam1Subscriber(t, pubA, params...) + witnessed, got := streamHeaders(t, witness), streamHeaders(t, sub) + + hdr := message.SubgroupHeader{SubgroupIDMode: message.SubgroupIDExplicit} + aHdr := hdr + aHdr.ReplayingSubgroup = tc.aReplay + a, err := aPub.OpenSubgroup(aHdr) + if err != nil { + t.Fatalf("A OpenSubgroup: %v", err) + } + if err := a.WriteObjectAt(tc.aID, &message.SubgroupObject{Payload: []byte("a")}); err != nil { + t.Fatalf("A WriteObjectAt %d: %v", tc.aID, err) + } + awaitHeader(t, witnessed) + if tc.filter == nil { + awaitHeader(t, got) // A's stream; B's comes next + } + b, err := bPub.OpenSubgroup(hdr) // FIRST_OBJECT set + if err != nil { + t.Fatalf("B OpenSubgroup: %v", err) + } + if err := b.WriteObjectAt(tc.bID, &message.SubgroupObject{Payload: []byte("b")}); err != nil { + t.Fatalf("B WriteObjectAt %d: %v", tc.bID, err) + } + if h := awaitHeader(t, got); h.ReplayingSubgroup != tc.replay { + t.Fatalf("stream beginning with Object %d: FIRST_OBJECT clear = %v, want %v", + tc.bID, h.ReplayingSubgroup, tc.replay) + } + }) + } +}