Relay: detect the rest of §2.4.2's malformed-track conditions

STATUS.md

@@ -428,22 +428,43 @@ Known protocol gaps, roughly ordered by how load-bearing they are:
   subscription"), nor migrates to the New Session URI, nor closes the session
   once no subscriptions remain (§3.6 RECOMMENDED). It waits for the sender to
   close.
-- **Malformed tracks: per-Object conditions (§2.4.2, §12.8, §12.9)** — the
-  session reports Object Properties that make a track malformed
-  (`session.ErrMalformedTrack`), and the relay then ends the track: PUBLISH_DONE
-  MALFORMED_TRACK to every downstream subscriber, its subscription to that
-  publisher cancelled, the Object not cached. The relay also ends it for two
-  Prior Group ID Gap values in one Group, and for a duplicate that differs from
-  the cached first copy (§9.1; Payloads and Immutable Properties compared only
-  when both copies are Normal, since Normal may become End of Group). A
-  duplicate is not compared once the first copy left the cache (evicted or
-  expired), nor while a concurrent contributor has yet to cache it. Not
-  detected: the rest of §2.4.2's list other than an Object after END_OF_GROUP
-  on the same stream. A downstream FETCH already being served from the cache
-  when the track is found malformed is not reset: the relay does not track
-  fetch streams per track. One interpretation: an Object with two Immutable
-  Properties is treated as malformed, although §12.7 states "MUST NOT contain
-  more than one instance" outside its list of malformed conditions.
+- **Malformed tracks (§2.4.2, §9.1, §12.8, §12.9)** — the session reports
+  Object Properties that make a track malformed (`session.ErrMalformedTrack`),
+  and the relay then ends the track: PUBLISH_DONE MALFORMED_TRACK to every
+  downstream subscriber, its subscription to that publisher cancelled, the
+  Objects triggering it not cached (removed, if earlier ones were). The relay
+  also detects, on live subgroup and datagram Objects of any upstream, against
+  the last 32 Groups (`registry.TrackEntry.ClaimDelivered`, `SubgroupEnded`,
+  `RecordDuplicate`): §2.4.2's list — a Subgroup's Publisher Priority
+  changing; an Object past a Subgroup's, Group's or Track's end, or two
+  different ends; a duplicate that differs from the cached first copy (§9.1;
+  Payloads and Immutable Properties compared only when both copies are
+  Normal, since Normal may become End of Group) — and two Prior Group ID Gap
+  values in one Group. An end is the first missing ID: an END_OF_GROUP or
+  END_OF_TRACK status at M ends the Group at M (END_OF_TRACK the Track too),
+  and a FIN (§11.4.3; the Group's too with END_OF_GROUP set, §11.4.2) or a
+  datagram's END_OF_GROUP bit after Object N at N+1. Each end is also checked
+  against the Objects already received, duplicates included. Interpretations,
+  the first three chosen with the maintainer: §2.4.2 calls both the status
+  Object at M and the Object N the "final Object", which would make the
+  §9.1-equivalent M = N+1 two different finals; a Normal Object at an end is
+  past it only if a FIN or bit set it — with status Objects alone it is the
+  §9.1 existing-to-not-existing change or the §2.1 late Object; for the same
+  reason a status end at M and a FIN or bit end at M+1 agree, ending at M; a
+  datagram's END_OF_GROUP bit counts as a Group's end, which §2.4.2's
+  non-exhaustive list does not name; and §2.4.2 item 7 (another Forwarding
+  Preference) is read per Object, since it may vary within a Track (§11.2.1),
+  so it is the §9.1 duplicate check. Not detected: in upstream FETCH
+  responses (whose End of Track is not used either), in a session that is not
+  a relay's, past the window, and a duplicate once the first copy left the
+  cache (evicted or expired) or before a concurrent contributor cached it. An
+  Object another upstream had claimed but not yet cached when a later end put
+  it past that end stays cached. A downstream FETCH already being served from
+  the cache when the track is found malformed is not reset: the relay does not
+  track fetch streams per track. One interpretation: an Object with two
+  Immutable Properties is treated as malformed, although §12.7 states "MUST
+  NOT contain more than one instance" outside its list of malformed
+  conditions.
 - **Objects inside an announced gap are dropped, not malformed (§2.1, §9.1,
   §12.8, §12.9)** — an interpretation. §12.8 and §12.9 list "an Object with an
   ID within a previously communicated gap" and "a gap covering an Object it

pkg/relay/duplicate_consistency_test.go

@@ -175,3 +175,41 @@ func TestRelay_DuplicateConsistency(t *testing.T) {
 		})
 	}
 }
+
+// TestRelay_DuplicateEndOfGroupEndsGroup: an END_OF_GROUP arriving as a
+// duplicate of a Normal Object (§9.1: existing to not existing) still ends the
+// Group, so an Object past it makes the track malformed (§2.4.2).
+func TestRelay_DuplicateEndOfGroupEndsGroup(t *testing.T) {
+	t.Parallel()
+	pubA, teardown := connectRelay(t, relay.Config{})
+	defer teardown()
+	pubB := dialAnotherClient(t, pubA)
+	publishVideoTrack(t, pubA, "cam1", 1)
+	publishVideoTrack(t, pubB, "cam1", 2)
+	subSess := dialAnotherClient(t, pubA)
+	subReq := subscribeCam1(t, subSess)
+	received := make(chan got, 8)
+	go receiveDatagrams(t.Context(), subSess, received)
+
+	send := func(sess *session.Session, alias uint64, d *message.ObjectDatagram) {
+		t.Helper()
+		d.TrackAlias, d.GroupID = alias, 1
+		d.Type |= message.DatagramDefaultPriorityBit
+		if err := sess.SendDatagram(d); err != nil {
+			t.Fatalf("SendDatagram: %v", err)
+		}
+	}
+	send(pubA, 1, &message.ObjectDatagram{ObjectID: 2, ObjectPayload: []byte("x")})
+	select {
+	case <-received: // forwarded, so cached
+	case <-time.After(2 * time.Second):
+		t.Fatal("Object 2 not forwarded")
+	}
+	send(pubB, 2, &message.ObjectDatagram{
+		Type: message.DatagramStatusBit, ObjectID: 2, ObjectStatus: message.ObjectStatusEndOfGroup,
+	})
+	send(pubB, 2, &message.ObjectDatagram{ObjectID: 3, ObjectPayload: []byte("x")})
+	if pd := awaitPublishDone(t, subReq); pd.StatusCode != moqt.PublishDoneMalformedTrack {
+		t.Fatalf("PUBLISH_DONE %#x, want MALFORMED_TRACK", uint64(pd.StatusCode))
+	}
+}

pkg/relay/handler_datagram.go

@@ -63,7 +63,16 @@ func (h *sessionHandler) handleDatagram(ctx context.Context, d *message.ObjectDa
 
 	// §9.3: the first copy of {GroupID, ObjectID} wins, unless an announced
 	// gap says it does not exist (§2.1, §9.1).
-	fresh, err := entry.ClaimDelivered(d.GroupID, d.ObjectID, message.ObjectPriorGaps(d.Properties))
+	info := registry.ObjectInfo{
+		Group:      d.GroupID,
+		Object:     d.ObjectID,
+		Datagram:   true,
+		Priority:   d.PublisherPriority,
+		Status:     d.ObjectStatus,
+		EndOfGroup: d.HasEndOfGroup(),
+		Gaps:       message.ObjectPriorGaps(d.Properties),
+	}
+	fresh, err := entry.ClaimDelivered(info)
 	if err != nil {
 		h.endMalformedTrack(ctx, entry, h.sess, err)
 		return
@@ -79,6 +88,10 @@ func (h *sessionHandler) handleDatagram(ctx context.Context, d *message.ObjectDa
 			Payload:           d.ObjectPayload,
 		}); err != nil {
 			h.endMalformedTrack(ctx, entry, h.sess, err)
+			return
+		}
+		if err := entry.RecordDuplicate(info); err != nil {
+			h.endMalformedTrack(ctx, entry, h.sess, err)
 		}
 		return
 	}

pkg/relay/handler_duplicate.go

@@ -23,6 +23,9 @@ import (
 // A copy an announced gap says does not exist is compared too, when the
 // first copy is cached: §2.1 excuses its arrival, not a different content.
 //
+// On a difference the first copy is removed from the cache: it triggered the
+// Malformed Track status too, and such Objects MUST NOT be cached (§2.4.2).
+//
 // Limitation: nothing is compared when the first copy is not in the cache:
 // evicted, expired (§12.3), or not yet put there by a concurrent contributor.
 func checkDuplicate(c *cache.ObjectCache, dup *cache.CachedObject) error {
@@ -47,6 +50,7 @@ func checkDuplicate(c *cache.ObjectCache, dup *cache.CachedObject) error {
 	default:
 		return nil
 	}
+	c.Delete(dup.GroupID, dup.ObjectID)
 	return fmt.Errorf("%w: a duplicate of Object %d in Group %d has a different %s (§9.1, §2.4.2)",
 		session.ErrMalformedTrack, dup.ObjectID, dup.GroupID, field)
 }

pkg/relay/handler_duplicate_test.go

@@ -0,0 +1,25 @@
+package relay
+
+import (
+	"testing"
+
+	"github.com/floatdrop/moq-go/pkg/relay/cache"
+)
+
+// TestCheckDuplicatePurgesFirstCopy: a first copy that a differing duplicate
+// makes malformed is removed from the cache: Object(s) triggering Malformed
+// Track status MUST NOT be cached (§2.4.2).
+func TestCheckDuplicatePurgesFirstCopy(t *testing.T) {
+	t.Parallel()
+	c := cache.NewObjectCache(8, 0)
+	c.Put(&cache.CachedObject{GroupID: 1, ObjectID: 0, Payload: []byte("a")})
+	if err := checkDuplicate(c, &cache.CachedObject{GroupID: 1, ObjectID: 0, Payload: []byte("a")}); err != nil {
+		t.Fatalf("an identical copy: %v", err)
+	}
+	if err := checkDuplicate(c, &cache.CachedObject{GroupID: 1, ObjectID: 0, Payload: []byte("b")}); err == nil {
+		t.Fatal("a copy with another Payload is not malformed")
+	}
+	if _, ok := c.Get(1, 0); ok {
+		t.Error("the first copy is still cached")
+	}
+}

pkg/relay/handler_fanout.go

@@ -297,6 +297,8 @@ func (h *sessionHandler) runFanout(ctx context.Context, stream *session.Incoming
 
 	var (
 		firstObj = true
+		// last is the last Object read, which a FIN ends the Subgroup after.
+		last *message.SubgroupObject
 		// pos counts every Object read, dedup losers included: each is an
 		// Object between its neighbours (§11.4.3).
 		pos = inboundPos{src: stream}
@@ -320,25 +322,10 @@ func (h *sessionHandler) runFanout(ctx context.Context, stream *session.Incoming
 			obj, err = stream.ReadObject()
 		}
 		if err != nil {
-			if errors.Is(err, io.EOF) {
-				return // clean end of stream — last contributor will FIN.
-			}
-			if errors.Is(err, context.Canceled) {
-				// The session is going away: reset, never FIN.
-				inboundReset = true
-				return
-			}
-			if errors.Is(err, session.ErrMalformedTrack) {
-				malformed(err)
-				return
-			}
-			h.log.LogAttrs(ctx, slog.LevelDebug, "fanout: inbound ReadObject failed",
-				slog.String("err", err.Error()))
-			// An unparseable object leaves the publisher writing; stop it.
-			stream.Cancel(moqt.StreamResetInternalError)
-			inboundReset = true
+			inboundReset, inboundResetCode = h.inboundEnded(ctx, entry, stream, hdr, last, err)
 			return
 		}
+		last = obj
 
 		if terminalSeen {
 			h.log.LogAttrs(ctx, slog.LevelDebug,
@@ -360,7 +347,15 @@ func (h *sessionHandler) runFanout(ctx context.Context, stream *session.Incoming
 		// §9.3: the first upstream to deliver {GroupID, ObjectID} forwards it,
 		// unless an announced gap says it does not exist (§2.1, §9.1). Outside
 		// sg.Mu, so dedup losers never touch the writer set.
-		fresh, err := entry.ClaimDelivered(hdr.GroupID, objectID, message.ObjectPriorGaps(obj.Properties))
+		info := registry.ObjectInfo{
+			Group:    hdr.GroupID,
+			Object:   objectID,
+			Subgroup: hdr.SubgroupID,
+			Priority: hdr.PublisherPriority,
+			Status:   obj.ObjectStatus,
+			Gaps:     message.ObjectPriorGaps(obj.Properties),
+		}
+		fresh, err := entry.ClaimDelivered(info)
 		if err != nil {
 			malformed(err)
 			return
@@ -379,6 +374,10 @@ func (h *sessionHandler) runFanout(ctx context.Context, stream *session.Incoming
 				malformed(err)
 				return
 			}
+			if err := entry.RecordDuplicate(info); err != nil {
+				malformed(err)
+				return
+			}
 			continue // redundant copy already forwarded by a peer upstream.
 		}
 
@@ -437,6 +436,50 @@ func (h *sessionHandler) runFanout(ctx context.Context, stream *session.Incoming
 	}
 }
 
+// inboundEnded handles err, which ended the reads of stream (whose header is
+// hdr, and whose last Object read was last, nil if none), and reports whether
+// the stream counts as reset, and with which code, for the Subgroup's FIN
+// (§9.3).
+func (h *sessionHandler) inboundEnded(
+	ctx context.Context,
+	entry *registry.TrackEntry,
+	stream *session.IncomingSubgroupStream,
+	hdr message.SubgroupHeader,
+	last *message.SubgroupObject,
+	err error,
+) (reset bool, code moqt.StreamResetCode) {
+	switch {
+	case errors.Is(err, io.EOF):
+		// Clean end of stream; the last contributor will FIN. It ends the
+		// Subgroup after the last Object read (§2.4.2).
+		if last == nil {
+			return false, 0
+		}
+		err = entry.SubgroupEnded(registry.ObjectInfo{
+			Group:    hdr.GroupID,
+			Object:   stream.ObjectID(),
+			Subgroup: hdr.SubgroupID,
+			Priority: hdr.PublisherPriority,
+			Status:   last.ObjectStatus,
+		}, hdr.EndOfGroup)
+		if err == nil {
+			return false, 0
+		}
+	case errors.Is(err, context.Canceled):
+		// The session is going away: reset, never FIN.
+		return true, moqt.StreamResetCancelled
+	case !errors.Is(err, session.ErrMalformedTrack):
+		h.log.LogAttrs(ctx, slog.LevelDebug, "fanout: inbound ReadObject failed",
+			slog.String("err", err.Error()))
+		// An unparseable object leaves the publisher writing; stop it.
+		stream.Cancel(moqt.StreamResetInternalError)
+		return true, moqt.StreamResetCancelled
+	}
+	stream.Cancel(moqt.StreamResetMalformedTrack)
+	h.endMalformedTrack(ctx, entry, h.sess, err)
+	return true, moqt.StreamResetMalformedTrack
+}
+
 // openWriterForSub starts a subgroupWriter for sub and records it in writers
 // (nil when sub is not Established, so it is not retried).
 //

pkg/relay/handler_fanout_multipub_test.go

@@ -364,15 +364,8 @@ func TestFanout_MultiPublisher_MergesDisjointObjects(t *testing.T) {
 			t.Fatalf("B WriteObjectAt %d: %v", id, err)
 		}
 	}
-	if err := aSg.Close(); err != nil {
-		t.Fatalf("A Close: %v", err)
-	}
-	if err := bSg.Close(); err != nil {
-		t.Fatalf("B Close: %v", err)
-	}
-
-	// Collect every delivered object until the merged stream(s) finish. Six
-	// distinct objects must arrive, each exactly once.
+	// Collect every delivered object. Six distinct objects must arrive, each
+	// exactly once.
 	seen := map[uint64]int{}
 	deadline := time.After(3 * time.Second)
 	for len(seen) < 6 {
@@ -396,6 +389,14 @@ func TestFanout_MultiPublisher_MergesDisjointObjects(t *testing.T) {
 			t.Fatalf("object %d missing from delivered set %v", id, slices.Sorted(maps.Keys(seen)))
 		}
 	}
+
+	// A ends with a reset: a FIN would make its last Object, 4, the
+	// Subgroup's final one, and B's 5 past it would make the track malformed
+	// (§2.4.2).
+	aSg.Cancel(moqt.StreamResetCancelled)
+	if err := bSg.Close(); err != nil {
+		t.Fatalf("B Close: %v", err)
+	}
 }
 
 // awaitObject waits for the next delivered object event (failing on a stream end