Clarify group delivery semantics with recv_group and next_group_ordered (#1324)

Co-authored-by: Claude <noreply@anthropic.com>

Author: kixelated

doc/rs/index.md

@@ -253,7 +253,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let mut track = broadcast.subscribe("chat").await?;
 
     // Read groups and frames
-    while let Some(group) = track.next_group().await? {
+    while let Some(group) = track.recv_group().await? {
         while let Some(frame) = group.read().await? {
             println!("Received: {:?}", frame);
         }

js/clock/src/main.ts

@@ -147,7 +147,7 @@ async function subscribe(config: Config) {
 
 	// Handle groups and frames like the Rust implementation
 	for (;;) {
-		const group = await track.nextGroup();
+		const group = await track.recvGroup();
 		if (!group) {
 			console.log("❌ Connection ended");
 			break;

js/hang/src/container/legacy.ts

@@ -109,7 +109,7 @@ export class Consumer {
 	async #run() {
 		// Start fetching groups in the background
 		for (;;) {
-			const consumer = await this.#track.nextGroup();
+			const consumer = await this.#track.recvGroup();
 			if (!consumer) break;
 
 			// To improve TTV, we always start with the first group.

js/lite/examples/subscribe.ts

@@ -12,7 +12,7 @@ async function main() {
 
 	// Read data as it arrives
 	for (;;) {
-		const group = await track.nextGroup();
+		const group = await track.recvGroup();
 		if (!group) break;
 
 		for (;;) {

js/lite/src/ietf/publisher.ts

@@ -161,7 +161,7 @@ export class Publisher {
 			// Serve track groups, racing with stream close (= Unsubscribe)
 			const serving = (async () => {
 				for (;;) {
-					const group = await track.nextGroup();
+					const group = await track.recvGroup();
 					if (!group) return;
 					void this.#runGroup(msg.requestId, group);
 				}

js/lite/src/lite/publisher.ts

@@ -201,7 +201,7 @@ export class Publisher {
 	async #runTrack(sub: bigint, broadcast: Path.Valid, track: Track, stream: Writer) {
 		try {
 			for (;;) {
-				const next = track.nextGroup();
+				const next = track.recvGroup();
 				const group = await Promise.race([next, stream.closed]);
 				if (!group) {
 					next.then((group) => group?.close()).catch(() => {});

js/lite/src/track.test.ts

@@ -0,0 +1,51 @@
+import { expect, test } from "bun:test";
+import { Group } from "./group.ts";
+import { Track } from "./track.ts";
+
+test("nextGroupOrdered skips late arrivals", async () => {
+	const track = new Track("test");
+
+	track.writeGroup(new Group(5));
+
+	const first = await track.nextGroupOrdered();
+	expect(first?.sequence).toBe(5);
+
+	// Late arrivals with sequence <= last returned are skipped.
+	track.writeGroup(new Group(3));
+	track.writeGroup(new Group(4));
+	track.writeGroup(new Group(7));
+
+	const next = await track.nextGroupOrdered();
+	expect(next?.sequence).toBe(7);
+});
+
+test("nextGroupOrdered returns buffered groups in sequence", async () => {
+	const track = new Track("test");
+
+	track.writeGroup(new Group(3));
+	track.writeGroup(new Group(5));
+
+	expect((await track.nextGroupOrdered())?.sequence).toBe(3);
+	expect((await track.nextGroupOrdered())?.sequence).toBe(5);
+});
+
+test("recvGroup after nextGroupOrdered still returns late arrivals", async () => {
+	const track = new Track("test");
+
+	track.writeGroup(new Group(5));
+
+	// Ordered returns seq 5, advancing its cursor.
+	const ordered = await track.nextGroupOrdered();
+	expect(ordered?.sequence).toBe(5);
+
+	// recvGroup is independent of the ordered cursor: a late seq 3 still surfaces.
+	track.writeGroup(new Group(3));
+	const recv = await track.recvGroup();
+	expect(recv?.sequence).toBe(3);
+});
+
+test("nextGroupOrdered returns undefined when track closes", async () => {
+	const track = new Track("test");
+	track.close();
+	expect(await track.nextGroupOrdered()).toBeUndefined();
+});

js/lite/src/track.ts

@@ -11,6 +11,7 @@ export class Track {
 
 	state = new TrackState();
 	#next?: number;
+	#nextSequence = 0;
 
 	readonly closed: Promise<Error | undefined>;
 
@@ -91,7 +92,14 @@ export class Track {
 		group.close();
 	}
 
-	async nextGroup(): Promise<Group | undefined> {
+	/**
+	 * Receive the next group available on this track, in arrival order.
+	 *
+	 * Groups may arrive out of order or with gaps due to network conditions.
+	 * Use {@link nextGroupOrdered} if you need groups in sequence order,
+	 * skipping those that arrive too late.
+	 */
+	async recvGroup(): Promise<Group | undefined> {
 		for (;;) {
 			const groups = this.state.groups.peek();
 			if (groups.length > 0) {
@@ -106,6 +114,33 @@ export class Track {
 		}
 	}
 
+	/**
+	 * @deprecated Use {@link recvGroup} for arrival order, or {@link nextGroupOrdered} for sequence order.
+	 */
+	async nextGroup(): Promise<Group | undefined> {
+		return this.recvGroup();
+	}
+
+	/**
+	 * Return the next group with a strictly-greater sequence number than the last returned.
+	 *
+	 * Late arrivals (with a sequence number at or below the last one returned) are silently skipped.
+	 *
+	 * NOTE: This will be renamed to `nextGroup` in the next major version.
+	 */
+	async nextGroupOrdered(): Promise<Group | undefined> {
+		for (;;) {
+			const group = await this.recvGroup();
+			if (!group) return undefined;
+			if (group.sequence < this.#nextSequence) {
+				group.close();
+				continue;
+			}
+			this.#nextSequence = group.sequence + 1;
+			return group;
+		}
+	}
+
 	async readFrame(): Promise<Uint8Array | undefined> {
 		return (await this.readFrameSequence())?.data;
 	}

js/watch/src/audio/decoder.ts

@@ -292,7 +292,7 @@ export class Decoder {
 			// Process data segments
 			// TODO: Use a consumer wrapper for CMAF to support latency control
 			for (;;) {
-				const group = await sub.nextGroup();
+				const group = await sub.recvGroup();
 				if (!group) break;
 
 				effect.spawn(async () => {

js/watch/src/video/decoder.ts

@@ -380,7 +380,7 @@ class DecoderTrack {
 			// Process data segments
 			// TODO: Use a consumer wrapper for CMAF to support latency control
 			for (;;) {
-				const group = await Promise.race([sub.nextGroup(), effect.cancel]);
+				const group = await Promise.race([sub.recvGroup(), effect.cancel]);
 				if (!group) break;
 
 				effect.spawn(async () => {