Relay: forward a later upstream's Object Properties on a merged Subgroup

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). 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`). |

pkg/relay/handler_fanout.go

@@ -63,13 +63,32 @@ 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.
 	sawClean  bool
 	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,
 				})
@@ -513,6 +534,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,18 +706,21 @@ 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
-	// §11.4.2 FIRST_OBJECT bit; otherwise the stream is a replay. All its
-	// blocking I/O is bounded by w.ctx.
+	// 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 {
 		if w.unbridge != nil {
 			w.unbridge()
 			w.unbridge = nil
 		}
 		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 +778,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 +788,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 +898,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

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,234 @@ 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
+				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 stream := 1; ; stream++ {
+					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, stream, sg.Header.ReplayingSubgroup}
+						}
+					}()
+				}
+			}()
+			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)
+			}
+			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)
+			}
+			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)
+			}
+			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)
+			}
+			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)
+			}
+		})
+	}
+}
+
+// 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)
+			}
+		})
+	}
+}