feat(net)!: lite-07 SUBSCRIBE_FINISH keeps the Subscribe Stream open (lite-07 ranges, part 1)

drafts/draft-lcurley-moq-lite.md

@@ -499,25 +499,28 @@ An advertiser signals capacity through its route alone, withdrawing or re-pricin
 A subscriber opens Subscribe Streams to request a Track.
 
 The subscriber MUST start a Subscribe Stream with a SUBSCRIBE message followed by any number of SUBSCRIBE_UPDATE messages.
-The publisher replies with a SUBSCRIBE_OK message once each Range's start is resolved, followed by a SUBSCRIBE_END message once the subscription ends.
-For a live track the publisher MAY withhold SUBSCRIBE_OK until the first matching group resolves the start; if the track has already ended with no matching groups, it sends SUBSCRIBE_END with no preceding SUBSCRIBE_OK.
+The publisher replies with a SUBSCRIBE_OK message once each Range's start is resolved, and a SUBSCRIBE_FINISH message once the Track's final group is known.
+For a live track the publisher MAY withhold SUBSCRIBE_OK until the first matching group resolves the start, and withholds the SUBSCRIBE_DROP messages for groups it gives up meanwhile until it sends SUBSCRIBE_OK.
+It MUST NOT withhold it past the point where no group can resolve it: once every Group in a subscription whose every Range has an `End` other than None has been given up, or once the Track has finished, the publisher sends SUBSCRIBE_OK and then the SUBSCRIBE_DROP messages it withheld.
+A Track that finished before producing any Group is the one case with nothing to resolve: SUBSCRIBE_FINISH with a Group of 0 needs no SUBSCRIBE_OK.
 A requested start past the publisher's largest position is the exception: the publisher MUST send SUBSCRIBE_OK without waiting, so a subscriber resuming just past what it holds learns where the live feed is from the answer alone (a quiet track may not reach the start for a long time).
+A relay knows that position only from its upstream's answer, never from its cache: a group it cached earlier says nothing about where the live feed is now.
 A rejection is a stream reset: a publisher that cannot serve the subscription (no such track, an ended broadcast, or any other refusal) MUST promptly reset the stream rather than leave it pending, so a subscriber distinguishes "pending" from "refused" by the reset, not by a timeout.
 A route claims capability rather than inventory, so a subscription for a covered path that names nothing is refused this way too.
 
 The track's immutable publisher properties are not carried here; they are fetched once via a [Track Stream](#track-stream).
 The subscriber needs the track's TRACK_INFO (notably its timescale) to interpret FRAME messages, and MAY open the Track and Subscribe streams concurrently, buffering frames until it arrives.
 
-The publisher sends SUBSCRIBE_OK once the start is resolved, and SUBSCRIBE_END as soon as the exclusive final group is known (see [SUBSCRIBE_OK](#subscribe-ok) and [SUBSCRIBE_END](#subscribe-end)).
-The publisher closes its side of the stream (FIN) after SUBSCRIBE_END, once every Group Stream has finished or been reset.
-A subscription whose every Range has an `End` other than None is covered once every Group in its Ranges has been sent as a Group Stream or a datagram, or named by SUBSCRIBE_DROP.
-The publisher MAY then close its side (FIN) without SUBSCRIBE_END, once every Group Stream has finished or been reset, rather than wait for a SUBSCRIBE_UPDATE to widen it.
-That FIN ends the subscription, not the Track: the subscriber MUST NOT treat it as the Track's end, and opens a new Subscribe Stream to receive more.
-It can arrive after the subscriber widened the Ranges in a SUBSCRIBE_UPDATE that the publisher never read.
-After reading the publisher's FIN, the subscriber MUST close its side (FIN) once its tail accounting has settled: every received Group Stream has been read to its FIN, reset, or dropped, and every sequence in range has arrived, been named by SUBSCRIBE_DROP, or exhausted the subscriber's grace period.
+The publisher sends SUBSCRIBE_OK once the start is resolved, and SUBSCRIBE_FINISH as soon as the exclusive final group is known (see [SUBSCRIBE_OK](#subscribe-ok) and [SUBSCRIBE_FINISH](#subscribe-finish)).
+Neither closes the stream: the publisher keeps reading SUBSCRIBE_UPDATE messages, so an update can still widen the Ranges, including into an ended Track's history, and the publisher serves or drops what it newly covers.
+The subscriber judges a Range complete from what it observed: the Groups it received, the SUBSCRIBE_DROP messages, the Range Starts in SUBSCRIBE_OK, and SUBSCRIBE_FINISH.
+Drops and Groups name absolute sequences, so the subscriber judges them against its latest Ranges, and an update still in flight changes nothing it has counted.
+The publisher never closes its side first: a subscription whose every Range has an `End` other than None and whose every Group has been delivered or dropped stays open for an update to widen it.
+
+The subscriber closes its side (FIN) once it wants nothing more from the subscription, and the publisher then closes its side (FIN), cancelling any Group Stream still open.
+After SUBSCRIBE_FINISH, the subscriber SHOULD close its side once its tail accounting has settled: every received Group Stream has been read to its FIN, reset, or dropped, and every sequence in range has arrived, been named by SUBSCRIBE_DROP, or exhausted the subscriber's grace period.
 A publisher closing its session gracefully MUST wait for the subscriber's FIN or reset on every Subscribe Stream before closing the connection; a transport acknowledgement alone does not confirm the subscriber read the tail.
-A subscriber MAY cancel by closing its side (FIN); the publisher then closes its side (FIN) without owing SUBSCRIBE_END.
-Any other subscription stays open until SUBSCRIBE_END, and either endpoint MAY reset the stream at any time.
+Either endpoint MAY reset the stream at any time.
 
 ### Fetch
 A subscriber opens a Fetch Stream (0x3) to request a single Group from a Track.
@@ -1117,7 +1120,7 @@ Each sum MUST saturate at 2^62-1.
 A publisher sends an ANNOUNCE_END message to retract a previously started route, referencing its Announce ID.
 The id is retired and MUST NOT be referenced again.
 Retraction claims nothing about the content: a broadcast whose publisher stops advertising may remain readable by exact path, serving its stored groups over FETCH.
-Retraction does not disturb subscriptions already in flight, which conclude normally with SUBSCRIBE_END.
+Retraction does not disturb subscriptions already in flight, which conclude normally with SUBSCRIBE_FINISH.
 How a subscriber learns the path of stored content is out of band, e.g. an application catalog.
 
 ~~~
@@ -1495,7 +1498,7 @@ The name a TRACKS_ADD on the stream listed.
 
 ## SUBSCRIBE_OK {#subscribe-ok}
 A SUBSCRIBE_OK message confirms a subscription and resolves where each Range starts.
-It is the first message the publisher sends on the Subscribe Stream, once those starts are known.
+It is the first message the publisher sends on the Subscribe Stream, once those starts are known, except that SUBSCRIBE_FINISH MAY precede it.
 
 This is the trimmed-down counterpart of MoqTransport's SUBSCRIBE_OK: it retains the name and the role of the publisher's positive response, but carries only the publisher's largest position and where each Range starts (all other per-track properties live in [TRACK_INFO](#track-info)).
 
@@ -1525,11 +1528,12 @@ The Range's Groups from this one are owed as described in [SUBSCRIBE_DROP](#subs
 A Range whose `Range Start` is its requested start Group begins at the requested frame, and otherwise at frame 0 (see [Positions](#positions)).
 The Ranges are those of the SUBSCRIBE unless the publisher applied a SUBSCRIBE_UPDATE before answering; a subscriber that sent none MUST treat a count other than its number of Ranges as a PROTOCOL_VIOLATION.
 
-## SUBSCRIBE_END {#subscribe-end}
-A SUBSCRIBE_END message is sent by the publisher to signal that no group at or after a given sequence will be produced.
+## SUBSCRIBE_FINISH {#subscribe-finish}
+A SUBSCRIBE_FINISH message is sent by the publisher to signal that no group at or after a given sequence will be produced.
+It ends the Track, not the subscription: the Subscribe Stream stays open (see [Subscribe](#subscribe)).
 
 ~~~
-SUBSCRIBE_END Message {
+SUBSCRIBE_FINISH Message {
   Type (i) = 0x1
   Message Length (i)
   Group (i)
@@ -1538,7 +1542,7 @@ SUBSCRIBE_END Message {
 ~~~
 
 **Type**:
-Set to 0x1 to indicate a SUBSCRIBE_END message.
+Set to 0x1 to indicate a SUBSCRIBE_FINISH message.
 
 **Group**:
 The exclusive end of the range: the absolute sequence number of the first group that will never be delivered.
@@ -1549,7 +1553,7 @@ The subscriber MUST NOT wait for any group at or after this sequence.
 0 if the track ended cleanly, otherwise the stream error code it ended with plus 1.
 Groups below the end stay deliverable either way; a subscriber reaching the end surfaces the error instead of a clean end, and a relay forwards it with the end.
 
-The publisher SHOULD send SUBSCRIBE_END as soon as the exclusive final group is known, even while lower groups are still in flight.
+The publisher SHOULD send SUBSCRIBE_FINISH as soon as the exclusive final group is known, even while lower groups are still in flight.
 
 ## SUBSCRIBE_DROP {#subscribe-drop}
 
@@ -1573,7 +1577,7 @@ A subscriber MUST end the readers of a dropped group it holds with that error, i
 For every sequence in the subscription's range, the publisher MUST send a Group Stream, send a datagram, or name it in SUBSCRIBE_DROP.
 This includes sequences it skipped and streams reset before their headers were delivered.
 A reliably delivered reset retaining the Group Stream header accounts for that group instead.
-The publisher SHOULD send SUBSCRIBE_DROP as soon as it gives a sequence up, without waiting for SUBSCRIBE_END.
+The publisher SHOULD send SUBSCRIBE_DROP as soon as it gives a sequence up, without waiting for SUBSCRIBE_FINISH.
 A datagram counts as sent even if it is lost; a subscriber SHOULD bound its wait for unaccounted sequences, for example by Subscriber Max Age.
 
 ## FETCH
@@ -1895,14 +1899,15 @@ The `Message Length` describes the payload size on the wire.
 - A SUBSCRIBE_UPDATE that lowers a `Start` covers the groups below the old start that the publisher still holds, not only those that arrive later. A group already sent is not sent again.
 - Replaced SUBSCRIBE_OK `Group` with a `Range Start` per Range: the Group where that Range's delivery starts, including at a Relative or Time Anchor, from which its Groups are owed.
 - Added `Largest Group` and `Largest Frame` to SUBSCRIBE_OK: the publisher's largest position when it answers, which a subscriber takes as where the live feed is. A publisher MUST answer at once when the requested start is past it.
-- Added `Error Code` to SUBSCRIBE_END: a track can end with an error while keeping the groups below its end.
-- Restored SUBSCRIBE_DROP (type 0x2), replacing Stream Count on SUBSCRIBE_END. Every sequence is accounted for by a Group Stream, a datagram, or a drop; a publisher SHOULD send a drop as soon as it gives a sequence up.
+- Renamed SUBSCRIBE_END to SUBSCRIBE_FINISH, matching TRACKS_FINISH: it ends the Track, not the subscription.
+- Added `Error Code` to SUBSCRIBE_FINISH: a track can end with an error while keeping the groups below its end.
+- Restored SUBSCRIBE_DROP (type 0x2), replacing Stream Count on SUBSCRIBE_FINISH. Every sequence is accounted for by a Group Stream, a datagram, or a drop; a publisher SHOULD send a drop as soon as it gives a sequence up.
 - SUBSCRIBE_DROP's Error Code is a stream error code, which ends the dropped groups' readers, including one still arriving.
 - Refuse a SUBSCRIBE_DROP whose Group Start is greater than Group End plus one on every version that carries it.
-- SUBSCRIBE_END declares the final boundary while lower groups can still be in flight; the Subscribe Stream FIN follows once every Group Stream has finished or been reset.
-- A publisher MAY FIN a subscription whose every Range has an End, once each Group in them is sent or dropped, without SUBSCRIBE_END. The FIN ends the subscription, not the Track.
-- The subscriber FINs its Subscribe Stream after settling its tail; graceful session close waits for that FIN or reset.
-- A subscriber MAY cancel by FIN on its Subscribe Stream, and the publisher then FINs without SUBSCRIBE_END.
+- SUBSCRIBE_FINISH declares the final boundary while lower groups can still be in flight, and no longer closes the Subscribe Stream: the publisher keeps reading SUBSCRIBE_UPDATE, so an update can reach an ended Track's history.
+- A covered subscription, whose every Range has an End and whose every Group was sent or dropped, also stays open for an update; the publisher never FINs first.
+- A publisher that withheld SUBSCRIBE_OK sends it, then the drops it withheld, once no group can resolve the start: every Group of a bounded subscription was given up, or the Track finished.
+- The subscriber FINs its Subscribe Stream once it wants nothing more, such as after settling its tail past SUBSCRIBE_FINISH, and the publisher FINs in reply; graceful session close waits for the subscriber's FIN or reset.
 
 ### Delivery {#changelog-07-delivery}
 

js/net/src/integration.test.ts

@@ -500,11 +500,11 @@ test("integration: lite-07 serves sparse and relative ranges", async () => {
 	}
 });
 
-// A lite-07 publisher ends a bounded subscription once it covered the range, releasing its
-// demand, and the subscriber keeps the copy: the read ends with its window, the same range reads
-// again from the cache, and a wider one subscribes again for the rest. Mirrors the Rust
-// `a_covered_bounded_subscription_ends_on_lite07`; reads end as Rust's do since moq-dev/moq#7545.
-test("integration: lite-07 keeps a covered bounded subscription's copy", async () => {
+// A lite-07 publisher keeps a covered bounded subscription open for an update: the read ends
+// with its window, the same range reads again from the cache, and a wider one reaches upstream
+// for the rest, past a group that never existed. The demand goes with the readers. Mirrors the
+// Rust `a_covered_bounded_subscription_stays_open`.
+test("integration: lite-07 keeps a covered bounded subscription open", async () => {
 	const pair = createMockTransportPair(Lite.ALPN_07_WIP);
 	const origin = new OriginProducer();
 	const [client, server] = await Promise.all([
@@ -532,19 +532,20 @@ test("integration: lite-07 keeps a covered bounded subscription's copy", async (
 	try {
 		const reader = bounded(3n);
 		expect(await read(reader, 2)).toEqual([1n, 2n]);
-		// The publisher's demand ends with the covered subscription, and the read with its window.
-		while (producer.subscription.peek() !== undefined) await producer.subscription.changed();
+		// The read ends with its window, while the covered subscription stays open.
 		expect(await reader.recvGroup()).toBeUndefined();
+		expect(producer.subscription.peek()).toBeDefined();
 
 		const again = bounded(3n);
 		expect(await read(again, 2)).toEqual([1n, 2n]);
-		expect(producer.subscription.peek()).toBeUndefined();
 
 		const wider = bounded(6n);
 		expect(await read(wider, 4)).toEqual([1n, 2n, 4n, 5n]);
 		reader.close();
 		again.close();
 		wider.close();
+		// The demand goes with the readers.
+		while (producer.subscription.peek() !== undefined) await producer.subscription.changed();
 	} finally {
 		remote.close();
 		broadcast.close();

js/net/src/lite/connection.test.ts

@@ -32,6 +32,7 @@ import {
 	ALPN_06,
 	ALPN_07_WIP,
 	hasTrackStream,
+	keepsSubscribeOpen,
 	Version,
 	versionName,
 } from "./version.ts";
@@ -538,8 +539,10 @@ async function servedRawSubscription(alpn: string = ALPN_07_WIP, version: Versio
 	expect("start" in (await decodeSubscribeResponse(subscriber.reader, version))).toBe(true);
 	producer.finish();
 	producer.close();
-	// SUBSCRIBE_END, then the publisher's FIN.
-	await subscriber.reader.readAll();
+	// SUBSCRIBE_END, then before lite-07 the publisher's FIN; lite-07 keeps the stream open.
+	if (keepsSubscribeOpen(version)) {
+		while (!("end" in (await decodeSubscribeResponse(subscriber.reader, version)))) {}
+	} else await subscriber.reader.readAll();
 
 	let closed = false;
 	void pair.server.closed.then(() => {
@@ -715,7 +718,8 @@ test("close waits for a request served while withdrawals are in flight", async (
 		expect("start" in (await decodeSubscribeResponse(subscriber.reader, version))).toBe(true);
 		producer.finish();
 		producer.close();
-		await subscriber.reader.readAll();
+		// SUBSCRIBE_END; lite-07 keeps the stream open past it.
+		while (!("end" in (await decodeSubscribeResponse(subscriber.reader, version)))) {}
 
 		// The withdrawal completes, but the subscription still waits for the subscriber's FIN.
 		fin.release();

js/net/src/lite/publisher.test.ts

@@ -30,7 +30,7 @@ import {
 	SubscribeUpdate,
 } from "./subscribe.ts";
 import { Track as TrackMessage } from "./track.ts";
-import { ALPN_05, ALPN_06, ALPN_07_WIP, Version } from "./version.ts";
+import { ALPN_05, ALPN_06, ALPN_07_WIP, keepsSubscribeOpen, Version } from "./version.ts";
 
 function publish(origin: OriginProducer, path: Path.Valid) {
 	const broadcastPublication = origin.createBroadcast(path);
@@ -892,8 +892,9 @@ test("lite-07 SUBSCRIBE_START names each resolved range start", async () => {
 });
 
 // A bounded window that resolves empty has no group to answer it, so the publisher answers it
-// at once and ends the subscription as covered, as a local read of it ends at once.
-test("lite-07 answers and ends a bounded window that resolves empty", async () => {
+// at once. The stream stays open for an update until the subscriber FINs, and the publisher
+// FINs in reply.
+test("lite-07 answers a bounded window that resolves empty and FINs in reply", async () => {
 	const version = Version.DRAFT_07;
 	const sub = await servedSubscription({
 		version,
@@ -905,6 +906,7 @@ test("lite-07 answers and ends a bounded window that resolves empty", async () =
 		const resp = await decodeSubscribeResponse(sub.client.reader, version);
 		if (!("start" in resp)) throw new Error("expected SUBSCRIBE_START");
 		expect(resp.start.starts).toEqual([undefined]);
+		sub.client.writer.close();
 		expect(await decodeSubscribeResponseMaybe(sub.client.reader, version)).toBeUndefined();
 	} finally {
 		await sub.close();
@@ -1041,6 +1043,8 @@ for (const version of [Version.DRAFT_06, Version.DRAFT_07]) {
 				const resp = await decodeSubscribeResponseMaybe(sub.client.reader, version);
 				if (!resp) break;
 				if ("drop" in resp) codes.push(resp.drop.error);
+				// Lite-07 keeps the stream open past SUBSCRIBE_END, after the sweep's drops.
+				if ("end" in resp && keepsSubscribeOpen(version)) break;
 			}
 			expect(codes).toEqual([version === Version.DRAFT_07 ? StreamCode.NotFound : 0]);
 		} finally {
@@ -2124,7 +2128,8 @@ test("blocked DROP responses stop opening later groups", async () => {
 		track.finish();
 		track.close();
 		let dropped = 0;
-		for (;;) {
+		// Lite-07 keeps the stream open past SUBSCRIBE_END, so read until every group is named.
+		while (dropped < 72) {
 			const response = await decodeSubscribeResponseMaybe(client.reader, Version.DRAFT_07);
 			if (!response) break;
 			if ("drop" in response) dropped += Number(response.drop.end - response.drop.start) + 1;