diff --git a/STATUS.md b/STATUS.md index ff64fde5..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}`. | +| 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 65648356..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, }) @@ -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,11 +706,13 @@ 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() @@ -694,6 +720,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 +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 diff --git a/pkg/relay/handler_fanout_multipub_test.go b/pkg/relay/handler_fanout_multipub_test.go index 326a6586..499c6e1a 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,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) + } + }) + } +} diff --git a/pkg/relay/metrics.go b/pkg/relay/metrics.go index d6f45ac3..dee63a58 100644 --- a/pkg/relay/metrics.go +++ b/pkg/relay/metrics.go @@ -105,6 +105,15 @@ 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. 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 ) // String returns a stable, lowercase name suitable for use as a metric label @@ -123,6 +132,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 {