From 1ecbf3a04068adb8fea681262b9912a331cfca53 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 8 Oct 2026 16:30:06 -0700 Subject: [PATCH 1/7] quest: claim quest/m0/release-22/noq Co-Authored-By: Claude Opus 5.5 From 931b72cbf41ded835e3b959bf09fcd733273ff58 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 8 Oct 2026 16:53:25 -0700 Subject: [PATCH 2/7] fix(deps): bump moq-noq to 1.3.4 A raw QUIC peer without datagram support now gets a max datagram size of 0 instead of a panic (moq-dev/noq#29). Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 16 ++++++++-------- Cargo.toml | 6 +++--- 2 files changed, 11 insertions(+), 11 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 5ef52f29da..c6ff75f065 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4536,9 +4536,9 @@ dependencies = [ [[package]] name = "moq-noq" -version = "1.3.3" +version = "1.3.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b87d97674919e237b1ba6a7d7b1c9bdf5ac39cf94a3a86a307f3a31cdd5d83e7" +checksum = "90fc18fcbed942196f9124d7987848cbd5dd4a7407b1e353d1b6a2bfa57d0ddf" dependencies = [ "bytes", "cfg_aliases", @@ -4558,9 +4558,9 @@ dependencies = [ [[package]] name = "moq-noq-proto" -version = "1.3.3" +version = "1.3.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0992aa7804dfc60c7c5a4ba4d1dc720f32de38cd06fc7b317284f1f3d5f0bdf5" +checksum = "bd9ff00e72d7158264ff794f7da0955859bda424bb0154a5fc7d60aa084a5897" dependencies = [ "aes-gcm", "aws-lc-rs", @@ -4589,9 +4589,9 @@ dependencies = [ [[package]] name = "moq-noq-udp" -version = "1.3.3" +version = "1.3.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e038634392b7255a76310bf2d20f169c945556344b87882b4b969dd11c25543e" +checksum = "0094cf49d3c3a79e9f732edb263477ba70a750baf321f53502cace5ad2cdc120" dependencies = [ "cfg_aliases", "libc", @@ -9848,9 +9848,9 @@ dependencies = [ [[package]] name = "web-transport-moq" -version = "1.3.3" +version = "1.3.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6ba618c4bd4741eab697eb046515e34607e21b8cea28c762177ad897629887d8" +checksum = "c8051685a029b2b5257b2f0ca51eb385495b437437786c7442f00c36178de90f" dependencies = [ "bytes", "futures", diff --git a/Cargo.toml b/Cargo.toml index 6f10dc407c..884d703af2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -151,8 +151,8 @@ moq-mux = { version = "0.10.11", path = "rs/moq-mux" } moq-net = { version = "0.3.10", path = "rs/moq-net" } # The MoQ fork of noq (moq-dev/noq). iroh keeps upstream noq, so a build with the # iroh feature carries both stacks. -moq-noq-proto = { version = "1.3.3", default-features = false } -moq-noq-udp = "1.3.3" +moq-noq-proto = { version = "1.3.4", default-features = false } +moq-noq-udp = "1.3.4" # NVENC bindings, forked from ViliamVadocz/nvidia-video-codec-sdk to dlopen the # driver at runtime. Compiles on any platform (macOS included) but only actually # used by moq-video on Linux. @@ -236,7 +236,7 @@ web-transport-iroh = "0.7" # default-features off so the QUIC crypto provider is chosen by moq-tokio's # aws-lc-rs / ring features. # The WebTransport adapter released with moq-noq, from the same repository. -web-transport-moq = { version = "1.3.3", default-features = false } +web-transport-moq = { version = "1.3.4", default-features = false } web-transport-proto = "0.6" web-transport-trait = "0.4" web-transport-wasm = "0.6" From 210e2e9f725a0a32f59e8bc130fe7493ee36baec Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 8 Oct 2026 23:10:01 -0700 Subject: [PATCH 3/7] quest: claim m0/release-22/location-filter Co-Authored-By: Claude Opus 5.5 From b4c1bf38b85ace61a8406a665d709619ab380bc2 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 8 Oct 2026 23:03:19 -0700 Subject: [PATCH 4/7] fix(net): draft-22 LOCATION_FILTER carries its Location Filter Type (backport #5080) Backport of #5080 onto release. Draft-22 encodes LOCATION_FILTER as a Location Filter Type followed by only the fields it names, with no Length; drafts 20 and 21 keep the length-inferred field list. Adapted to release's Encode/Decode trait coding API and its JS parameter decode, which predate the per-draft rework on main. The quest file and the FILL_PARAMETERS group-order test from main are not part of this backport. (cherry picked from commit b8095029aff292dc0ccc03d4f1d112abe4badc9c) Co-Authored-By: Claude Opus 5.5 --- js/net/src/ietf/filter.test.ts | 83 ++++++++ js/net/src/ietf/filter.ts | 197 ++++++++++++++++--- js/net/src/ietf/ietf.test.ts | 49 +++-- js/net/src/ietf/parameters.ts | 59 ++++-- js/net/src/ietf/subscribe.ts | 5 +- rs/moq-net/src/ietf/fetch.rs | 39 ++-- rs/moq-net/src/ietf/filter.rs | 323 ++++++++++++++++++++++++------- rs/moq-net/src/ietf/subscribe.rs | 20 ++ rs/moq-net/src/ietf/version.rs | 73 ++++--- 9 files changed, 675 insertions(+), 173 deletions(-) diff --git a/js/net/src/ietf/filter.test.ts b/js/net/src/ietf/filter.test.ts index 91871de4f3..02289db85e 100644 --- a/js/net/src/ietf/filter.test.ts +++ b/js/net/src/ietf/filter.test.ts @@ -1,4 +1,6 @@ import { expect, test } from "bun:test"; +import { ProtocolViolation } from "../error.ts"; +import { Reader } from "../stream.ts"; import * as Filter from "./filter.ts"; import { joinFilter } from "./subscribe.ts"; import { Version } from "./version.ts"; @@ -182,3 +184,84 @@ test("fill skips allowed parameters it ignores", () => { const encoded = new Uint8Array([0x02, 0x20, 0x2a, 0x01, 0x01, 0x02]); expect(Filter.decodeFill(encoded, NEW)).toEqual({ filter: { kind: "relative", groups: 2n }, rangeFilters: false }); }); + +const TYPED = Version.DRAFT_22; + +// Draft-22 section 9.20.9: a Location Filter Type, then only the fields it names. +test("draft-22 carries the Location Filter Type", () => { + const cases: [Filter.Filter, number[]][] = [ + [{ kind: "unfiltered" }, [0x00]], + [{ kind: "relative", groups: 1n }, [0x01, 0x01]], + [{ kind: "absolute", startGroup: 7n, startObject: 3n }, [0x02, 0x07, 0x03]], + [{ kind: "absolute", startGroup: 7n, startObject: 3n, endGroup: 9n }, [0x03, 0x07, 0x03, 0x02]], + [ + { kind: "absolute", startGroup: 7n, startObject: 3n, endGroup: 9n, endObject: 4n }, + [0x04, 0x07, 0x03, 0x02, 0x04], + ], + [{ kind: "nextObject" }, [0x05]], + // The type says absolute, so {0, 0} no longer has to borrow the unfiltered spelling. + [{ kind: "absolute", startGroup: 0n, startObject: 0n }, [0x02, 0x00, 0x00]], + ]; + for (const [filter, bytes] of cases) { + expect([...Filter.encode(filter, TYPED)]).toEqual(bytes); + // No Length: the parameter value is the type and its fields. + expect([...Filter.encodeParam(filter, TYPED)]).toEqual(bytes); + expect(Filter.decode(new Uint8Array(bytes), TYPED)).toEqual(filter); + } + + // Draft-21 still frames the value with a Length and infers the meaning from it. + expect([...Filter.encodeParam({ kind: "nextObject" }, Version.DRAFT_21)]).toEqual([0x02, 0x00, 0x00]); +}); + +// With no Length, the type alone says where the value ends, so the bytes after it are left +// for the next parameter. +test("draft-22 reads only the named fields", async () => { + const r = new Reader(undefined, new Uint8Array([0x05, 0x01, 0x02]), TYPED); + expect(await Filter.decodeParam(r, TYPED)).toEqual({ kind: "nextObject" }); + expect(await r.u62()).toBe(1n); + expect(await r.u62()).toBe(2n); +}); + +test("draft-22 rejects an unknown Location Filter Type", async () => { + for (const type of [0x06, 0x3f]) { + expect(() => Filter.decode(new Uint8Array([type, 0x00, 0x00]), TYPED)).toThrow(ProtocolViolation); + const r = new Reader(undefined, new Uint8Array([type, 0x00, 0x00]), TYPED); + await expect(Filter.decodeParam(r, TYPED)).rejects.toThrow(ProtocolViolation); + } +}); + +// The nested LOCATION_FILTER carries its type too, and the whole-track form the spec calls +// zero-length is type 0x00. +test("draft-22 fill carries the Location Filter Type", () => { + // A two-byte group, so the draft-20 Length (4) and the draft-22 type (3) differ. + const range: Filter.Filter = { kind: "absolute", startGroup: 128n, startObject: 0n, endGroup: 130n }; + expect([...Filter.encodeFill({ filter: range, rangeFilters: false }, NEW)]).toEqual([ + 0x01, 0x21, 0x04, 0x80, 0x80, 0x00, 0x02, + ]); + expect([...Filter.encodeFill({ filter: range, rangeFilters: false }, TYPED)]).toEqual([ + 0x01, 0x21, 0x03, 0x80, 0x80, 0x00, 0x02, + ]); + expect([...Filter.encodeFill({ filter: { kind: "nextObject" }, rangeFilters: false }, TYPED)]).toEqual([ + 0x01, 0x21, 0x05, + ]); + expect([...Filter.encodeFill({ filter: { kind: "unfiltered" }, rangeFilters: false }, TYPED)]).toEqual([ + 0x01, 0x21, 0x00, + ]); + + const fills: Filter.Fill[] = [ + { filter: undefined, rangeFilters: false }, + { filter: { kind: "unfiltered" }, rangeFilters: false }, + { filter: { kind: "nextObject" }, rangeFilters: false }, + { filter: { kind: "relative", groups: 1n }, rangeFilters: false }, + { filter: range, rangeFilters: false }, + ]; + for (const fill of fills) { + expect(Filter.decodeFill(Filter.encodeFill(fill, TYPED), TYPED)).toEqual(fill); + } + + // A uint8, a typed filter, then another uint8: the filter must stop at its own fields. + expect(Filter.decodeFill(new Uint8Array([0x03, 0x20, 0x80, 0x01, 0x05, 0x01, 0x01]), TYPED)).toEqual({ + filter: { kind: "nextObject" }, + rangeFilters: false, + }); +}); diff --git a/js/net/src/ietf/filter.ts b/js/net/src/ietf/filter.ts index d3157b1a7f..9d4511f59f 100644 --- a/js/net/src/ietf/filter.ts +++ b/js/net/src/ietf/filter.ts @@ -4,6 +4,7 @@ * @module */ +import { ProtocolViolation } from "../error.ts"; import type { Reader } from "../stream.ts"; import * as Varint from "../varint.ts"; import { type IetfVersion, Version } from "./version.ts"; @@ -48,6 +49,24 @@ export function isDraft20(version: IetfVersion): boolean { ); } +/** + * Whether this is draft-22 or newer. + * + * Draft-22 replaced the length-inferred field list with a Location Filter Type that names + * the fields that follow, so the value carries no Length. + */ +export function isDraft22(version: IetfVersion): boolean { + return isDraft20(version) && version !== Version.DRAFT_20 && version !== Version.DRAFT_21; +} + +/** The Location Filter Type of draft-22. */ +const TYPE_NONE = 0x0n; +const TYPE_RELATIVE_START = 0x1n; +const TYPE_ABSOLUTE_START = 0x2n; +const TYPE_ABSOLUTE_GROUP_END = 0x3n; +const TYPE_ABSOLUTE_RANGE = 0x4n; +const TYPE_NEXT_OBJECT = 0x5n; + /** Whether the Range Filters (draft-19) exist on this draft. */ export function hasRangeFilters(version: IetfVersion): boolean { return isDraft20(version) || version === Version.DRAFT_19; @@ -88,11 +107,117 @@ function endDelta(startGroup: bigint, endGroup: bigint): bigint { return endGroup - startGroup; } -/** Encode a filter as its raw parameter value. Unfiltered encodes to zero bytes. */ +/** + * Encode a filter as its raw parameter value, without the Length that frames it through + * draft-21. Unfiltered encodes to zero bytes on draft-20 and draft-21. + */ export function encode(filter: Filter, version: IetfVersion): Uint8Array { + if (isDraft22(version)) return encodeTyped(filter, version); return isDraft20(version) ? encodeFields(filter, version) : encodeTag(filter, version); } +/** Encode LOCATION_FILTER's value as it follows the parameter type from draft-17 on. */ +export function encodeParam(filter: Filter, version: IetfVersion): Uint8Array { + const value = encode(filter, version); + // Draft-22's Location Filter Type says where the value ends, so it needs no Length. + return isDraft22(version) ? value : encodeLengthPrefixed(value, version); +} + +/** Read LOCATION_FILTER's value as it follows the parameter type from draft-17 on. */ +export async function decodeParam(r: Reader, version: IetfVersion): Promise { + if (!isDraft22(version)) { + return decode(await r.read(await r.u53()), version); + } + const type = await r.u62(); + const fields: bigint[] = []; + for (let i = typedFieldCount(type); i > 0; i--) { + fields.push(await r.u62()); + } + return fromTyped(type, fields); +} + +function encodeTyped(filter: Filter, version: IetfVersion): Uint8Array { + switch (filter.kind) { + case "unfiltered": + return varint(TYPE_NONE, version); + case "nextObject": + return varint(TYPE_NEXT_OBJECT, version); + case "relative": + return concat([varint(TYPE_RELATIVE_START, version), varint(filter.groups, version)]); + case "absolute": { + const type = + filter.endGroup === undefined + ? TYPE_ABSOLUTE_START + : filter.endObject === undefined + ? TYPE_ABSOLUTE_GROUP_END + : TYPE_ABSOLUTE_RANGE; + const parts = [ + varint(type, version), + varint(filter.startGroup, version), + varint(filter.startObject, version), + ]; + if (filter.endGroup !== undefined) { + parts.push(varint(endDelta(filter.startGroup, filter.endGroup), version)); + if (filter.endObject !== undefined) { + parts.push(varint(filter.endObject, version)); + } + } + return concat(parts); + } + } +} + +/** How many fields follow a draft-22 Location Filter Type. */ +function typedFieldCount(type: bigint): number { + switch (type) { + case TYPE_NONE: + case TYPE_NEXT_OBJECT: + return 0; + case TYPE_RELATIVE_START: + return 1; + case TYPE_ABSOLUTE_START: + return 2; + case TYPE_ABSOLUTE_GROUP_END: + return 3; + case TYPE_ABSOLUTE_RANGE: + return 4; + default: + throw new ProtocolViolation(`unknown Location Filter Type: ${type}`); + } +} + +/** Build a filter from a draft-22 Location Filter Type and the fields it named. */ +function fromTyped(type: bigint, fields: bigint[]): Filter { + switch (type) { + case TYPE_NONE: + return { kind: "unfiltered" }; + case TYPE_NEXT_OBJECT: + return { kind: "nextObject" }; + case TYPE_RELATIVE_START: + return { kind: "relative", groups: fields[0] }; + default: { + const [startGroup, startObject, delta, endObject] = fields; + const endGroup = delta === undefined ? undefined : startGroup + delta; + if (endGroup !== undefined && endGroup > 0xffff_ffff_ffff_ffffn) { + throw new ProtocolViolation("LOCATION_FILTER end group exceeds 2^64 - 1"); + } + return { kind: "absolute", startGroup, startObject, endGroup, endObject }; + } + } +} + +/** Read a draft-22 filter from the front of `data`, returning what follows it. */ +function readTyped(data: Uint8Array, version: IetfVersion): [Filter, Uint8Array] { + let [type, rest] = unvarint(data, version); + const fields: bigint[] = []; + for (let i = typedFieldCount(type); i > 0; i--) { + const [field, next] = unvarint(rest, version); + fields.push(field); + rest = next; + } + return [fromTyped(type, fields), rest]; +} + function encodeFields(filter: Filter, version: IetfVersion): Uint8Array { switch (filter.kind) { case "unfiltered": @@ -152,8 +277,13 @@ function encodeTag(filter: Filter, version: IetfVersion): Uint8Array { } } -/** Decode a filter from its raw parameter value. */ +/** Decode a filter from its raw parameter value, which must be consumed whole. */ export function decode(data: Uint8Array, version: IetfVersion): Filter { + if (isDraft22(version)) { + const [filter, rest] = readTyped(data, version); + expectEmpty(rest); + return filter; + } return isDraft20(version) ? decodeFields(data, version) : decodeTag(data, version); } @@ -264,8 +394,8 @@ function expectEmpty(rest: Uint8Array): void { export interface Fill { /** * The range to fill. `undefined` means the Location Filter was omitted, which inherits - * the subscription's own filter; `unfiltered` (a zero length filter) means the whole - * track up to Largest Object. + * the subscription's own filter; `unfiltered` (a zero length filter, or type 0x00 from + * draft-22) means the whole track up to Largest Object. */ filter?: Filter; @@ -280,8 +410,11 @@ export interface Fill { /** LOCATION_FILTER, the only parameter we act on inside a fill. */ const FILL_LOCATION_FILTER = 0x21n; +type Framing = "byte" | "varint" | "bytes"; + /** - * The parameters draft-20 allows inside FILL_PARAMETERS, and how each frames its value. + * The parameters draft-20 allows inside FILL_PARAMETERS besides LOCATION_FILTER, and how + * each frames its value. * * Tabulated rather than derived, because neither shortcut is right. The Key-Value-Pair rule * keys framing off the id's parity, but the Range Filters (0x25-0x28) carry an explicit @@ -289,20 +422,9 @@ const FILL_LOCATION_FILTER = 0x21n; * than a varint, so reading it as one misparses any value with a leading 1-bit. Either * mistake desyncs every parameter after it. */ -/** - * Maps each to whether it carries a length prefix. - * - * The Key-Value-Pair rule keys the framing off the id's parity, but the Range Filters - * (0x25-0x28) are written with an explicit `Length` field despite two of them having even - * ids. Their own definition wins, so the framing is tabulated rather than derived: reading - * 0x26 or 0x28 as a bare varint would desync every parameter after it. - */ -type Framing = "byte" | "varint" | "bytes"; - const FILL_ALLOWED = new Map([ [0x0an, "varint"], // FILL_TIMEOUT [0x20n, "byte"], // SUBSCRIBER_PRIORITY, a uint8 - [FILL_LOCATION_FILTER, "bytes"], [0x22n, "byte"], // GROUP_ORDER, a uint8 [0x25n, "bytes"], // SUBGROUP_FILTER [0x26n, "bytes"], // OBJECTID_FILTER, length prefixed despite an even id @@ -317,7 +439,8 @@ const FILL_ALLOWED = new Map([ */ export function encodeFill(fill: Fill, version: IetfVersion): Uint8Array { // An omitted filter inherits the subscription's, so the scope is empty. An explicit - // unfiltered still encodes, as a zero length filter meaning the whole track. + // unfiltered still encodes, as a zero length filter (type 0x00 from draft-22) meaning + // the whole track. if (fill.filter === undefined) { return varint(0n, version); } @@ -325,7 +448,7 @@ export function encodeFill(fill: Fill, version: IetfVersion): Uint8Array { varint(1n, version), // The first type in a scope is not delta encoded, so this is the raw id. varint(FILL_LOCATION_FILTER, version), - encodeLengthPrefixed(encode(fill.filter, version), version), + encodeParam(fill.filter, version), ]); } @@ -346,6 +469,21 @@ export function decodeFill(data: Uint8Array, version: IetfVersion): Fill { prev = key; rest = afterType; + // Its framing depends on the draft, so the filter reads itself. + if (key === FILL_LOCATION_FILTER) { + if (filter !== undefined) { + throw new Error("duplicate LOCATION_FILTER inside FILL_PARAMETERS"); + } + if (isDraft22(version)) { + [filter, rest] = readTyped(rest, version); + } else { + let value: Uint8Array; + [value, rest] = readLengthPrefixed(rest, version); + filter = decode(value, version); + } + continue; + } + const framing = FILL_ALLOWED.get(key); if (framing === undefined) { throw new Error(`parameter ${key} is not allowed inside FILL_PARAMETERS`); @@ -367,19 +505,7 @@ export function decodeFill(data: Uint8Array, version: IetfVersion): Fill { continue; } - const [length, afterLength] = unvarint(rest, version); - if (BigInt(afterLength.length) < length) { - throw new Error("truncated value inside FILL_PARAMETERS"); - } - const value = afterLength.slice(0, Number(length)); - rest = afterLength.slice(Number(length)); - - if (key === FILL_LOCATION_FILTER) { - if (filter !== undefined) { - throw new Error("duplicate LOCATION_FILTER inside FILL_PARAMETERS"); - } - filter = decode(value, version); - } + [, rest] = readLengthPrefixed(rest, version); } if (rest.length !== 0) { @@ -389,6 +515,15 @@ export function decodeFill(data: Uint8Array, version: IetfVersion): Fill { return { filter, rangeFilters }; } +/** Split a length-prefixed value off the front of `data`, returning it and what follows. */ +function readLengthPrefixed(data: Uint8Array, version: IetfVersion): [Uint8Array, Uint8Array] { + const [length, afterLength] = unvarint(data, version); + if (BigInt(afterLength.length) < length) { + throw new Error("truncated value inside FILL_PARAMETERS"); + } + return [afterLength.slice(0, Number(length)), afterLength.slice(Number(length))]; +} + function encodeLengthPrefixed(value: Uint8Array, version: IetfVersion): Uint8Array { return concat([varint(BigInt(value.length), version), value]); } diff --git a/js/net/src/ietf/ietf.test.ts b/js/net/src/ietf/ietf.test.ts index 93ef4815a7..fef65c8f9f 100644 --- a/js/net/src/ietf/ietf.test.ts +++ b/js/net/src/ietf/ietf.test.ts @@ -1857,21 +1857,48 @@ test("Subscribe: FORWARD=0 decodes", async () => { }); // A draft-20 FETCH names the track up front and carries its range in LOCATION_FILTER. -test("Fetch v20: decodes every legal parameter", async () => { +// Groups 128 through 130: the group takes two bytes, so draft-20's Length (4) and +// draft-22's Location Filter Type (0x03) differ, and reading one as the other misframes +// every parameter after it. +for (const [version, locationFilter] of [ + [Version.DRAFT_20, [0x01, 0x04, 0x80, 0x80, 0x00, 0x02]], + [Version.DRAFT_21, [0x01, 0x04, 0x80, 0x80, 0x00, 0x02]], + [Version.DRAFT_22, [0x01, 0x03, 0x80, 0x80, 0x00, 0x02]], +] as const) { + test(`Fetch ${version}: decodes every legal parameter`, async () => { + const body = framed([ + ...TRACK_HEAD, + 0x07, // Number of Parameters + ...[0x03, 0x03, 0x03, 0x00, 0xaa], // AUTHORIZATION TOKEN + ...[0x07, 0x64], // FILL_TIMEOUT (0x0A) + ...[0x16, 0x40], // SUBSCRIBER_PRIORITY (0x20) + ...locationFilter, // LOCATION_FILTER (0x21) + ...[0x01, 0x01], // GROUP_ORDER (0x22) + ...[0x04, 0x02, 0x00, 0x05], // OBJECTID_FILTER (0x26) + ...[0x0f, 0x00], // INCLUDE_PROPERTIES (0x35) + ]); + + const msg = await decodeVersioned(body, Fetch.decode, version); + expect(msg.requestId).toBe(1n); + }); +} + +// Draft-22's LOCATION_FILTER is a Location Filter Type with no Length, so Next Object is the +// single byte 0x05 and the parameter after it starts right behind it. +test("Subscribe draft-22: Next Object is 0x21 0x05", async () => { const body = framed([ ...TRACK_HEAD, - 0x07, // Number of Parameters - ...[0x03, 0x03, 0x03, 0x00, 0xaa], // AUTHORIZATION TOKEN - ...[0x07, 0x64], // FILL_TIMEOUT (0x0A) - ...[0x16, 0x40], // SUBSCRIBER_PRIORITY (0x20) - ...[0x01, 0x03, 0x04, 0x00, 0x02], // LOCATION_FILTER (0x21) - ...[0x01, 0x01], // GROUP_ORDER (0x22) - ...[0x04, 0x02, 0x00, 0x05], // OBJECTID_FILTER (0x26) - ...[0x0f, 0x00], // INCLUDE_PROPERTIES (0x35) + 0x04, // Number of Parameters + ...[0x10, 0x01], // FORWARD (0x10) = 1 + ...[0x10, 0x40], // SUBSCRIBER_PRIORITY (0x20) = 64 + ...[0x01, 0x05], // LOCATION_FILTER (0x21): Next Object + ...[0x01, 0x02], // GROUP_ORDER (0x22) = Descending ]); - const msg = await decodeVersioned(body, Fetch.decode, Version.DRAFT_20); - expect(msg.requestId).toBe(1n); + const msg = await decodeVersioned(body, Subscribe.Subscribe.decode, Version.DRAFT_22); + expect(msg.filter).toEqual({ kind: "nextObject" }); + expect(msg.subscriberPriority).toBe(64); + expect(await encodeVersioned(msg, Version.DRAFT_22)).toEqual(body); }); // The tagged forms through draft-19, with a token. diff --git a/js/net/src/ietf/parameters.ts b/js/net/src/ietf/parameters.ts index 0530e27917..1d316a048e 100644 --- a/js/net/src/ietf/parameters.ts +++ b/js/net/src/ietf/parameters.ts @@ -1,5 +1,6 @@ import type { Reader, Writer } from "../stream.ts"; import * as Varint from "../varint.ts"; +import * as Filter from "./filter.ts"; import { type IetfVersion, Version } from "./version.ts"; /// Setup Option key constants (separate namespace from Message Parameters). @@ -236,7 +237,7 @@ const MSG_PARAM_REPEATABLE: readonly bigint[] = [ MSG_PARAM_TRACK_PROPERTY_FILTER, ]; -type MessageParamKind = "varint" | "uint8" | "bool" | "location" | "bytes"; +type MessageParamKind = "varint" | "uint8" | "bool" | "location" | "filter" | "bytes"; /** A `{Group, Object}` pair carried by a message parameter, such as LARGEST_OBJECT. */ export type MessageLocation = { groupId: bigint; objectId: bigint }; @@ -260,8 +261,9 @@ function getMessageParamKind(id: bigint): MessageParamKind { return "bool"; case MSG_PARAM_LARGEST_OBJECT: return "location"; - case MSG_PARAM_AUTHORIZATION_TOKEN: case MSG_PARAM_SUBSCRIPTION_FILTER: + return "filter"; + case MSG_PARAM_AUTHORIZATION_TOKEN: case MSG_PARAM_FILL_PARAMETERS: case MSG_PARAM_HOP_PATH: return "bytes"; @@ -299,6 +301,8 @@ export class Parameters { vars: Map; bytes: Map; #locations: Map; + /** LOCATION_FILTER, kept decoded because its framing depends on the draft. */ + #filter: Filter.Filter | undefined; /** Every instance of a parameter that may repeat, decoded only; we never send one. */ #repeated: Map; @@ -414,18 +418,13 @@ export class Parameters { this.#locations.set(MSG_PARAM_LARGEST_OBJECT, { ...v }); } - /** - * LOCATION_FILTER, as its raw parameter value. - * - * Raw because the encoding is version dependent: a Filter Type tag through draft-19, a - * list of up to four varints from draft-20 on. See `filter.ts`. - */ - get subscriptionFilter(): Uint8Array | undefined { - return this.bytes.get(MSG_PARAM_SUBSCRIPTION_FILTER); + /** LOCATION_FILTER: which Objects the subscription delivers. See `filter.ts`. */ + get subscriptionFilter(): Filter.Filter | undefined { + return this.#filter; } - set subscriptionFilter(v: Uint8Array) { - this.bytes.set(MSG_PARAM_SUBSCRIPTION_FILTER, v); + set subscriptionFilter(v: Filter.Filter) { + this.#filter = v; } /** FILL_PARAMETERS: the draft-20 backfill request, as its raw parameter value. */ @@ -472,7 +471,8 @@ export class Parameters { } async encode(w: Writer, version: IetfVersion) { - await w.u53(this.vars.size + this.bytes.size + this.#locations.size); + const filter = this.#filter; + await w.u53(this.vars.size + this.bytes.size + this.#locations.size + (filter ? 1 : 0)); if (version === Version.DRAFT_14 || version === Version.DRAFT_15) { for (const [id, value] of this.vars) { @@ -492,12 +492,20 @@ export class Parameters { await w.u53(encoded.length); await w.write(encoded); } + + if (filter) { + const encoded = Filter.encode(filter, version); + await w.u62(MSG_PARAM_SUBSCRIPTION_FILTER); + await w.u53(encoded.length); + await w.write(encoded); + } } else { // d16+: Delta encoding, merge all parameter storage, sort by key - const all: { key: bigint; storage: "var" | "bytes" | "location" }[] = []; + const all: { key: bigint; storage: "var" | "bytes" | "location" | "filter" }[] = []; for (const id of this.vars.keys()) all.push({ key: id, storage: "var" }); for (const id of this.bytes.keys()) all.push({ key: id, storage: "bytes" }); for (const id of this.#locations.keys()) all.push({ key: id, storage: "location" }); + if (filter) all.push({ key: MSG_PARAM_SUBSCRIPTION_FILTER, storage: "filter" }); all.sort((a, b) => (a.key < b.key ? -1 : a.key > b.key ? 1 : 0)); let prevId = 0n; @@ -516,8 +524,10 @@ export class Parameters { storage === "bytes" ? // biome-ignore lint/style/noNonNullAssertion: key is guaranteed to exist in bytes map this.bytes.get(key)! - : // biome-ignore lint/style/noNonNullAssertion: key is guaranteed to exist in locations map - encodeLocation(this.#locations.get(key)!); + : storage === "filter" && filter + ? Filter.encode(filter, version) + : // biome-ignore lint/style/noNonNullAssertion: key is guaranteed to exist in locations map + encodeLocation(this.#locations.get(key)!); await w.u53(value.length); await w.write(value); } @@ -558,6 +568,11 @@ export class Parameters { await w.u62(location.objectId); break; } + case "filter": { + if (!filter) throw new Error(`invalid LOCATION_FILTER message parameter: ${key.toString()}`); + await w.write(Filter.encodeParam(filter, version)); + break; + } case "bytes": { const value = this.bytes.get(key); if (value === undefined) throw new Error(`invalid bytes message parameter: ${key.toString()}`); @@ -604,6 +619,12 @@ export class Parameters { throw new Error(`duplicate message parameter id: ${id.toString()}`); } params.#locations.set(id, decodeLocation(bytes)); + } else if (id === MSG_PARAM_SUBSCRIPTION_FILTER && version !== Version.DRAFT_14) { + // Draft-14 carries the filter in the SUBSCRIBE body, so 0x21 is undefined there. + if (params.#filter !== undefined) { + throw new Error(`duplicate message parameter id: ${id.toString()}`); + } + params.#filter = Filter.decode(bytes, version); } else { if (params.bytes.has(id)) { throw new Error(`duplicate message parameter id: ${id.toString()}`); @@ -620,7 +641,8 @@ export class Parameters { continue; } - if (params.vars.has(id) || params.bytes.has(id) || params.#locations.has(id)) { + const filter = id === MSG_PARAM_SUBSCRIPTION_FILTER && params.#filter !== undefined; + if (params.vars.has(id) || params.bytes.has(id) || params.#locations.has(id) || filter) { throw new Error(`duplicate message parameter id: ${id.toString()}`); } @@ -641,6 +663,9 @@ export class Parameters { params.#locations.set(id, { groupId, objectId }); break; } + case "filter": + params.#filter = await Filter.decodeParam(r, version); + break; case "bytes": { const size = await r.u53(); params.bytes.set(id, await r.read(size)); diff --git a/js/net/src/ietf/subscribe.ts b/js/net/src/ietf/subscribe.ts index 32cf1dba1e..518a3950d5 100644 --- a/js/net/src/ietf/subscribe.ts +++ b/js/net/src/ietf/subscribe.ts @@ -96,7 +96,7 @@ export class Subscribe { params.subscriberPriority = this.subscriberPriority; params.groupOrder = GROUP_ORDER; params.forward = this.forward; - params.subscriptionFilter = Filter.encode(this.filter, version); + params.subscriptionFilter = this.filter; // FILL_PARAMETERS and INCLUDE_PROPERTIES arrived in draft-20. An older peer reads // either as an unknown parameter, which is a protocol violation, so they are @@ -176,8 +176,7 @@ export class Subscribe { } // An absent LOCATION_FILTER means the subscription is unfiltered. - const raw = params.subscriptionFilter; - const filter = raw !== undefined ? Filter.decode(raw, version) : { kind: "unfiltered" as const }; + const filter = params.subscriptionFilter ?? { kind: "unfiltered" as const }; const rawFill = params.fillParameters; const fill = rawFill !== undefined ? Filter.decodeFill(rawFill, version) : undefined; diff --git a/rs/moq-net/src/ietf/fetch.rs b/rs/moq-net/src/ietf/fetch.rs index e0e1b09730..394857f4d8 100644 --- a/rs/moq-net/src/ietf/fetch.rs +++ b/rs/moq-net/src/ietf/fetch.rs @@ -885,19 +885,29 @@ mod tests { /// Reading a Fetch Type there instead takes the namespace's field count for one. #[test] fn test_fetch_v20_decodes_every_parameter() { + // Groups 128 through 130. The group takes two bytes, so draft-20's Length (4) and + // draft-22's Location Filter Type (0x03) differ, and reading one as the other + // misframes every parameter after it. #[rustfmt::skip] - let body = [FETCH_HEAD, &[ - 0x07, // Number of Parameters - 0x03, 0x03, 0x03, 0x00, 0xAA, // AUTHORIZATION TOKEN - 0x07, 0x64, // FILL_TIMEOUT (0x0A) = 100, a varint - 0x16, 0x40, // SUBSCRIBER_PRIORITY (0x20) = 64 - 0x01, 0x03, 0x04, 0x00, 0x02, // LOCATION_FILTER (0x21): groups 4 through 6 - 0x01, 0x01, // GROUP_ORDER (0x22) = Ascending - 0x04, 0x02, 0x00, 0x05, // OBJECTID_FILTER (0x26): SetID 0, from 5 - 0x0F, 0x00, // INCLUDE_PROPERTIES (0x35) = 0 - ]].concat(); + let cases: [(Version, &[u8]); 3] = [ + (Version::Draft20, &[0x01, 0x04, 0x80, 0x80, 0x00, 0x02]), + (Version::Draft21, &[0x01, 0x04, 0x80, 0x80, 0x00, 0x02]), + (Version::Draft22, &[0x01, 0x03, 0x80, 0x80, 0x00, 0x02]), + ]; + + for (version, location_filter) in cases { + #[rustfmt::skip] + let body = [FETCH_HEAD, &[ + 0x07, // Number of Parameters + 0x03, 0x03, 0x03, 0x00, 0xAA, // AUTHORIZATION TOKEN + 0x07, 0x64, // FILL_TIMEOUT (0x0A) = 100, a varint + 0x16, 0x40, // SUBSCRIBER_PRIORITY (0x20) = 64 + ], location_filter, &[ + 0x01, 0x01, // GROUP_ORDER (0x22) = Ascending + 0x04, 0x02, 0x00, 0x05, // OBJECTID_FILTER (0x26): SetID 0, from 5 + 0x0F, 0x00, // INCLUDE_PROPERTIES (0x35) = 0 + ]].concat(); - for version in [Version::Draft20, Version::Draft21, Version::Draft22] { let fetch: Fetch = decode_message(&body, version).unwrap_or_else(|e| panic!("{version}: {e}")); assert_eq!(fetch.request_id, RequestId(1)); assert_eq!(fetch.subscriber_priority, 64); @@ -910,8 +920,11 @@ mod tests { namespace: Path::new("live"), track: "video".into(), filter: Filter::Absolute { - start: Location { group: 4, object: 0 }, - end: Some(crate::ietf::EndLocation { group: 6, object: None }), + start: Location { group: 128, object: 0 }, + end: Some(crate::ietf::EndLocation { + group: 130, + object: None, + }), }, }, "{version}" diff --git a/rs/moq-net/src/ietf/filter.rs b/rs/moq-net/src/ietf/filter.rs index a909fb4458..009e5f8de5 100644 --- a/rs/moq-net/src/ietf/filter.rs +++ b/rs/moq-net/src/ietf/filter.rs @@ -59,11 +59,23 @@ mod tag { pub const ABSOLUTE_RANGE: u64 = 0x4; } +/// The Location Filter Type of draft-22, which names the fields that follow it. +mod filter_type { + pub const NONE: u64 = 0x0; + pub const RELATIVE_START: u64 = 0x1; + pub const ABSOLUTE_START: u64 = 0x2; + pub const ABSOLUTE_GROUP_END: u64 = 0x3; + pub const ABSOLUTE_RANGE: u64 = 0x4; + pub const NEXT_OBJECT: u64 = 0x5; +} + impl Filter { /// Whether this is draft-20 or newer. /// /// Draft-20 replaced the Filter Type tag with up to four optional varints, where the /// number present selects the meaning, and added the parameters that ride alongside it. + /// Draft-22 names the meaning with a Location Filter Type instead, but keeps those + /// parameters, so this still gates them. pub(crate) fn is_draft20(version: Version) -> bool { !matches!( version, @@ -96,20 +108,81 @@ impl Filter { start: Location { group: 0, object: 0 }, end: None, } => {} + Self::Absolute { start, end } => Self::encode_absolute(w, start, end, version)?, + } + Ok(()) + } + + /// Encode the start, then as much of the end as is set. + fn encode_absolute( + w: &mut W, + start: Location, + end: Option, + version: Version, + ) -> Result<(), EncodeError> { + start.group.encode(w, version)?; + start.object.encode(w, version)?; + if let Some(end) = end { + Self::end_delta(start.group, end.group)?.encode(w, version)?; + if let Some(object) = end.object { + object.encode(w, version)?; + } + } + Ok(()) + } + + /// Encode the draft-22 form: the Location Filter Type, then only the fields it names. + fn encode_typed(&self, w: &mut W, version: Version) -> Result<(), EncodeError> { + match *self { + Self::Unfiltered => filter_type::NONE.encode(w, version)?, + Self::NextObject => filter_type::NEXT_OBJECT.encode(w, version)?, + Self::Relative(groups) => { + filter_type::RELATIVE_START.encode(w, version)?; + groups.encode(w, version)?; + } Self::Absolute { start, end } => { - start.group.encode(w, version)?; - start.object.encode(w, version)?; - if let Some(end) = end { - Self::end_delta(start.group, end.group)?.encode(w, version)?; - if let Some(object) = end.object { - object.encode(w, version)?; - } + match end { + None => filter_type::ABSOLUTE_START, + Some(EndLocation { object: None, .. }) => filter_type::ABSOLUTE_GROUP_END, + Some(EndLocation { object: Some(_), .. }) => filter_type::ABSOLUTE_RANGE, } + .encode(w, version)?; + Self::encode_absolute(w, start, end, version)?; } } Ok(()) } + /// Decode the draft-22 form, reading only the fields its Location Filter Type names. + fn decode_typed(r: &mut R, version: Version) -> Result { + Ok(match u64::decode(r, version)? { + filter_type::NONE => Self::Unfiltered, + filter_type::RELATIVE_START => Self::Relative(u64::decode(r, version)?), + kind @ (filter_type::ABSOLUTE_START | filter_type::ABSOLUTE_GROUP_END | filter_type::ABSOLUTE_RANGE) => { + let start = Location { + group: u64::decode(r, version)?, + object: u64::decode(r, version)?, + }; + let end = match kind { + filter_type::ABSOLUTE_START => None, + _ => Some(EndLocation { + group: start + .group + .checked_add(u64::decode(r, version)?) + .ok_or(DecodeError::BoundsExceeded)?, + object: match kind { + filter_type::ABSOLUTE_RANGE => Some(u64::decode(r, version)?), + _ => None, + }, + }), + }; + Self::Absolute { start, end } + } + filter_type::NEXT_OBJECT => Self::NextObject, + _ => return Err(DecodeError::InvalidValue), + }) + } + /// Decode the draft-20 field list, which the caller has already delimited. fn decode_fields(r: &mut R, version: Version) -> Result { let mut fields = Vec::with_capacity(4); @@ -226,8 +299,6 @@ impl Param for Filter { } fn param_encode(&self, w: &mut W, version: Version) -> Result<(), EncodeError> { - let mut buf = Vec::new(); - // Inner varints use the draft-15 leading-ones encoding on the drafts that predate // the switch, matching the other length-prefixed parameters. let sv = match version { @@ -235,28 +306,36 @@ impl Param for Filter { _ => version, }; - if Self::is_draft20(version) { - self.encode_fields(&mut buf, sv)?; - } else { - self.encode_tag(&mut buf, sv)?; + let mut buf = Vec::new(); + match version { + Version::Draft14 + | Version::Draft15 + | Version::Draft16 + | Version::Draft17 + | Version::Draft18 + | Version::Draft19 => self.encode_tag(&mut buf, sv)?, + Version::Draft20 | Version::Draft21 => self.encode_fields(&mut buf, sv)?, + // Draft-22 drops the Length, since the Location Filter Type says which fields follow. + _ => return self.encode_typed(w, version), } buf.encode(w, version) } fn param_decode(r: &mut R, version: Version) -> Result { - let data = Vec::::decode(r, version)?; - let mut buf = bytes::Bytes::from(data); let sv = match version { Version::Draft14 | Version::Draft15 | Version::Draft16 => Version::Draft15, - _ => version, - }; - - if Self::is_draft20(version) { + Version::Draft17 | Version::Draft18 | Version::Draft19 => version, // The field count is what carries the meaning, so the value is consumed whole. - return Self::decode_fields(&mut buf, sv); - } + Version::Draft20 | Version::Draft21 => { + let data = Vec::::decode(r, version)?; + return Self::decode_fields(&mut bytes::Bytes::from(data), version); + } + _ => return Self::decode_typed(r, version), + }; + let data = Vec::::decode(r, version)?; + let mut buf = bytes::Bytes::from(data); let filter = Self::decode_tag(&mut buf, sv)?; if buf.has_remaining() { return Err(DecodeError::TrailingBytes); @@ -458,6 +537,85 @@ mod tests { } } + const TYPED: Version = Version::Draft22; + + /// The whole parameter value as it sits on the wire, which from draft-22 has no Length. + fn raw(filter: Filter, version: Version) -> Vec { + let mut buf = Vec::new(); + filter.param_encode(&mut buf, version).expect("encode"); + buf + } + + fn decode_raw(bytes: &[u8], version: Version) -> Result<(Filter, usize), DecodeError> { + let mut r = bytes; + let filter = Filter::param_decode(&mut r, version)?; + Ok((filter, r.remaining())) + } + + /// Draft-22 section 9.20.9: a Location Filter Type, then only the fields it names. + #[test] + fn draft22_carries_the_type() { + let start = Location { group: 7, object: 3 }; + for (filter, bytes) in [ + (Filter::Unfiltered, vec![0x00]), + (Filter::Relative(1), vec![0x01, 0x01]), + (Filter::Absolute { start, end: None }, vec![0x02, 0x07, 0x03]), + ( + Filter::Absolute { + start, + end: Some(EndLocation { group: 9, object: None }), + }, + vec![0x03, 0x07, 0x03, 0x02], + ), + ( + Filter::Absolute { + start, + end: Some(EndLocation { + group: 9, + object: Some(4), + }), + }, + vec![0x04, 0x07, 0x03, 0x02, 0x04], + ), + (Filter::NextObject, vec![0x05]), + // The type says absolute, so {0, 0} no longer has to borrow the unfiltered spelling. + ( + Filter::Absolute { + start: Location::default(), + end: None, + }, + vec![0x02, 0x00, 0x00], + ), + ] { + assert_eq!(raw(filter, TYPED), bytes, "{filter:?}"); + assert_eq!(decode_raw(&bytes, TYPED).unwrap(), (filter, 0), "{filter:?}"); + } + } + + /// With no Length, the type alone says where the value ends, so the parameter after it + /// must be left untouched. Draft-21 still reads a Length first. + #[test] + fn draft22_reads_only_the_named_fields() { + assert_eq!(decode_raw(&[0x05, 0x22, 0x01], TYPED).unwrap(), (Filter::NextObject, 2)); + assert_eq!( + decode_raw(&[0x01, 0x01, 0x22, 0x01], TYPED).unwrap(), + (Filter::Relative(1), 2) + ); + + // The same Next Object on draft-21 is a Length of 2 and two zero fields. + assert_eq!(raw(Filter::NextObject, Version::Draft21), vec![0x02, 0x00, 0x00]); + } + + #[test] + fn draft22_rejects_an_unknown_type() { + for kind in [0x06u8, 0x3f] { + assert!(matches!( + decode_raw(&[kind, 0x00, 0x00], TYPED), + Err(DecodeError::InvalidValue) + )); + } + } + #[test] fn rejects_too_many_fields() { let mut buf = Vec::new(); @@ -495,8 +653,8 @@ enum Framing { #[derive(Default, Clone, Copy, Debug, PartialEq, Eq)] pub struct Fill { /// The range to fill. `None` means the Location Filter was omitted, which inherits the - /// subscription's own filter; [`Filter::Unfiltered`] (a zero-length filter) means the - /// whole track up to Largest Object. + /// subscription's own filter; [`Filter::Unfiltered`] (a zero-length filter, or type 0x00 + /// from draft-22) means the whole track up to Largest Object. pub filter: Option, /// Whether the scope carried a Range Filter (0x25-0x28). Those narrow which objects @@ -510,9 +668,9 @@ impl Fill { /// LOCATION_FILTER, the only nested parameter that changes what we deliver. const LOCATION_FILTER: u64 = 0x21; - /// The parameters the draft allows inside FILL_PARAMETERS, and how each frames its - /// value. Anything absent from this table is a protocol violation rather than something - /// to skip. + /// The parameters the draft allows inside FILL_PARAMETERS besides LOCATION_FILTER, and + /// how each frames its value. Anything else is a protocol violation rather than + /// something to skip. /// /// The framing is tabulated rather than derived, because neither shortcut is right. The /// Key-Value-Pair rule keys it off the id's parity, but the Range Filters (0x25-0x28) @@ -522,12 +680,11 @@ impl Fill { const ALLOWED: &'static [(u64, Framing)] = &[ (0x0A, Framing::Varint), // FILL_TIMEOUT (0x20, Framing::Byte), // SUBSCRIBER_PRIORITY, a uint8 - (Self::LOCATION_FILTER, Framing::Bytes), - (0x22, Framing::Byte), // GROUP_ORDER, a uint8 - (0x25, Framing::Bytes), // SUBGROUP_FILTER - (0x26, Framing::Bytes), // OBJECTID_FILTER, length prefixed despite an even id - (0x27, Framing::Bytes), // PRIORITY_FILTER - (0x28, Framing::Bytes), // OBJECT_PROPERTY_FILTER, likewise + (0x22, Framing::Byte), // GROUP_ORDER, a uint8 + (0x25, Framing::Bytes), // SUBGROUP_FILTER + (0x26, Framing::Bytes), // OBJECTID_FILTER, length prefixed despite an even id + (0x27, Framing::Bytes), // PRIORITY_FILTER + (0x28, Framing::Bytes), // OBJECT_PROPERTY_FILTER, likewise ]; } @@ -537,7 +694,8 @@ impl Param for Fill { // A nested scope is encoded like a message's parameters: a count, then the KVPs. // An omitted filter inherits the subscription's, so the scope is empty. An explicit - // Unfiltered still encodes, as a zero-length filter meaning the whole track. + // Unfiltered still encodes, as a zero-length filter (type 0x00 from draft-22) + // meaning the whole track. match self.filter { None => 0u64.encode(&mut buf, version)?, Some(filter) => { @@ -572,10 +730,7 @@ impl Param for Fill { }; prev = key; - let Some((_, framing)) = Self::ALLOWED.iter().find(|(id, _)| *id == key) else { - return Err(DecodeError::InvalidValue); - }; - + // Its framing depends on the draft, so the filter reads itself. if key == Self::LOCATION_FILTER { if filter.is_some() { return Err(DecodeError::Duplicate); @@ -584,6 +739,10 @@ impl Param for Fill { continue; } + let Some((_, framing)) = Self::ALLOWED.iter().find(|(id, _)| *id == key) else { + return Err(DecodeError::InvalidValue); + }; + // A Range Filter changes which objects the fill may contain, so its presence // is recorded even though its value is not interpreted. range_filters |= (0x25..=0x28).contains(&key); @@ -617,34 +776,48 @@ mod fill_tests { const NEW: Version = Version::Draft20; - fn round_trip(fill: Fill) -> Fill { + fn round_trip(fill: Fill, version: Version) -> Fill { let mut buf = Vec::new(); - fill.param_encode(&mut buf, NEW).expect("encode"); + fill.param_encode(&mut buf, version).expect("encode"); let mut bytes = bytes::Bytes::from(buf); - let decoded = Fill::param_decode(&mut bytes, NEW).expect("decode"); + let decoded = Fill::param_decode(&mut bytes, version).expect("decode"); assert!(!bytes.has_remaining()); decoded } + /// The scope's value, without the length prefix FILL_PARAMETERS carries. + fn value(filter: Filter, version: Version) -> Vec { + let fill = Fill { + filter: Some(filter), + range_filters: false, + }; + let mut buf = Vec::new(); + fill.param_encode(&mut buf, version).expect("encode"); + let mut bytes = bytes::Bytes::from(buf); + Vec::::decode(&mut bytes, version).expect("length prefix") + } + #[test] fn round_trips() { - for filter in [ - // An omitted filter (inherit the subscription's) and an explicit zero-length - // one (the whole track) are distinct spellings and must stay distinct. - None, - Some(Filter::Unfiltered), - Some(Filter::Relative(1)), - Some(Filter::Relative(3)), - Some(Filter::Absolute { - start: Location { group: 4, object: 0 }, - end: Some(EndLocation { group: 9, object: None }), - }), - ] { - let fill = Fill { - filter, - range_filters: false, - }; - assert_eq!(round_trip(fill).filter, filter, "{filter:?}"); + for version in [Version::Draft20, Version::Draft22] { + for filter in [ + // An omitted filter (inherit the subscription's) and an explicit zero-length + // one (the whole track) are distinct spellings and must stay distinct. + None, + Some(Filter::Unfiltered), + Some(Filter::Relative(1)), + Some(Filter::Relative(3)), + Some(Filter::Absolute { + start: Location { group: 4, object: 0 }, + end: Some(EndLocation { group: 9, object: None }), + }), + ] { + let fill = Fill { + filter, + range_filters: false, + }; + assert_eq!(round_trip(fill, version).filter, filter, "{version} {filter:?}"); + } } } @@ -652,16 +825,34 @@ mod fill_tests { /// that starts one group back. #[test] fn current_group_join() { - let fill = Fill { - filter: Some(Filter::Relative(1)), - range_filters: false, - }; - let mut buf = Vec::new(); - fill.param_encode(&mut buf, NEW).expect("encode"); // count=1, type=0x21, len=1, StartGroup=1 - let mut bytes = bytes::Bytes::from(buf); - let value = Vec::::decode(&mut bytes, NEW).expect("length prefix"); - assert_eq!(value, vec![0x01, 0x21, 0x01, 0x01]); + assert_eq!(value(Filter::Relative(1), NEW), vec![0x01, 0x21, 0x01, 0x01]); + // count=1, type=0x21, Relative Start, StartGroup=1: the same bytes by coincidence. + assert_eq!( + value(Filter::Relative(1), Version::Draft22), + vec![0x01, 0x21, 0x01, 0x01] + ); + } + + /// Draft-22's nested LOCATION_FILTER carries its type instead of a Length, and the + /// whole-track form the spec calls zero-length is type 0x00. + #[test] + fn draft22_nested_filter_carries_its_type() { + // A two-byte group, so the draft-20 Length (4) and the draft-22 type (3) differ. + let range = Filter::Absolute { + start: Location { group: 128, object: 0 }, + end: Some(EndLocation { + group: 130, + object: None, + }), + }; + assert_eq!(value(range, NEW), vec![0x01, 0x21, 0x04, 0x80, 0x80, 0x00, 0x02]); + assert_eq!( + value(range, Version::Draft22), + vec![0x01, 0x21, 0x03, 0x80, 0x80, 0x00, 0x02] + ); + assert_eq!(value(Filter::NextObject, Version::Draft22), vec![0x01, 0x21, 0x05]); + assert_eq!(value(Filter::Unfiltered, Version::Draft22), vec![0x01, 0x21, 0x00]); } /// A parameter the draft does not allow inside the scope is a violation, not something diff --git a/rs/moq-net/src/ietf/subscribe.rs b/rs/moq-net/src/ietf/subscribe.rs index 83be556125..f7c584cd90 100644 --- a/rs/moq-net/src/ietf/subscribe.rs +++ b/rs/moq-net/src/ietf/subscribe.rs @@ -1363,6 +1363,26 @@ mod tests { assert!(decode_message::(&body, Version::Draft20).is_err()); } + /// Draft-22's LOCATION_FILTER is a Location Filter Type with no Length, so Next Object + /// is the single byte 0x05 and the parameter after it starts right behind it. + #[test] + fn subscribe_decodes_a_draft22_next_object() { + #[rustfmt::skip] + let body = subscribe_body(&[ + 0x04, // Number of Parameters + 0x10, 0x01, // FORWARD (0x10) = 1 + 0x10, 0x40, // SUBSCRIBER_PRIORITY (0x20) = 64 + 0x01, 0x05, // LOCATION_FILTER (0x21): Next Object + 0x01, 0x01, // GROUP_ORDER (0x22) = Ascending + ]); + + let msg: Subscribe = decode_message(&body, Version::Draft22).unwrap(); + assert_eq!(msg.filter, Filter::NextObject); + assert_eq!(msg.subscriber_priority, 64); + assert_eq!(msg.group_order, GroupOrder::Ascending); + assert_eq!(encode_message(&msg, Version::Draft22), body); + } + /// FORWARD=0 is legal on every draft. It decodes so the request can be refused, /// rather than closing the session. #[test] diff --git a/rs/moq-net/src/ietf/version.rs b/rs/moq-net/src/ietf/version.rs index 9e1923a055..8badb11300 100644 --- a/rs/moq-net/src/ietf/version.rs +++ b/rs/moq-net/src/ietf/version.rs @@ -70,12 +70,12 @@ mod tests { buf } - /// Draft-21 and draft-22 are editorial: they restructure the document and drop the - /// Connection URL, Stream Cancellation, and Examples sections, leaving every codepoint - /// and field layout as draft-20 defined them. So they have to encode byte for byte the - /// same, and this pins that rather than trusting each version branch to fall forward. + /// Draft-21 is editorial: it restructures the document and drops the Connection URL, + /// Stream Cancellation, and Examples sections, leaving every codepoint and field layout + /// as draft-20 defined them. So it has to encode byte for byte the same, and this pins + /// that rather than trusting each version branch to fall forward. #[test] - fn draft21_and_draft22_match_draft20_on_the_wire() { + fn draft21_matches_draft20_on_the_wire() { let properties = Properties { timescale: Some(Timescale::new(90_000).unwrap()), priority: Some(64), @@ -143,32 +143,41 @@ mod tests { flags: GroupFlags::default(), }; - for version in [Version::Draft21, Version::Draft22] { - assert_eq!( - message(&subscribe, Version::Draft20), - message(&subscribe, version), - "SUBSCRIBE {version}" - ); - assert_eq!( - message(&subscribe_ok, Version::Draft20), - message(&subscribe_ok, version), - "SUBSCRIBE_OK {version}" - ); - assert_eq!( - message(&publish, Version::Draft20), - message(&publish, version), - "PUBLISH {version}" - ); - assert_eq!( - message(&fetch, Version::Draft20), - message(&fetch, version), - "FETCH {version}" - ); - assert_eq!( - field(&group, Version::Draft20), - field(&group, version), - "SUBGROUP_HEADER {version}" - ); - } + let (old, new) = (Version::Draft20, Version::Draft21); + assert_eq!(message(&subscribe, old), message(&subscribe, new), "SUBSCRIBE"); + assert_eq!(message(&subscribe_ok, old), message(&subscribe_ok, new), "SUBSCRIBE_OK"); + assert_eq!(message(&publish, old), message(&publish, new), "PUBLISH"); + assert_eq!(message(&fetch, old), message(&fetch, new), "FETCH"); + assert_eq!(field(&group, old), field(&group, new), "SUBGROUP_HEADER"); + } + + /// Draft-22's only wire change is LOCATION_FILTER, which drops its Length for a Location + /// Filter Type. So a SUBSCRIBE differs from draft-20 in those bytes alone. + #[test] + fn draft22_changes_only_the_location_filter() { + let subscribe = Subscribe { + request_id: RequestId(1), + track_namespace: Path::new("broadcast"), + track_name: "video".into(), + subscriber_priority: 128, + group_order: GroupOrder::Descending, + filter: Filter::NextObject, + fill: None, + properties_wanted: true, + forward: true, + range_filters: false, + }; + + // Draft-20 spells Next Object as a Length of 2 and two zero fields; draft-22 as type 0x05. + let old = message(&subscribe, Version::Draft20); + let at = old + .windows(4) + .position(|w| w == [0x01, 0x02, 0x00, 0x00]) + .expect("draft-20 LOCATION_FILTER"); + let mut expected = old[..at].to_vec(); + expected.extend_from_slice(&[0x01, 0x05]); + expected.extend_from_slice(&old[at + 4..]); + + assert_eq!(message(&subscribe, Version::Draft22), expected); } } From 6c2b733bbb6fd37d529c5539c626bf88b312e49a Mon Sep 17 00:00:00 2001 From: OneTooMany Date: Fri, 9 Oct 2026 10:12:22 +0200 Subject: [PATCH 5/7] fix(net): a warm copy's expiry no longer aborts the relay's latest group A front that parks a track adopts the relay copy's groups into a warm copy, sharing their producers. Once the warm copy ends, expire_closed aborted every idle group, including the latest group the relay's live copy still serves, so new subscribers skipped it and waited for a group that a steady catalog never writes. Abort only groups the ended track alone owns. Ports #4923's regression test to release's rejoin.rs on tokio's paused clock. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/track.rs | 6 ++- rs/moq-net/tests/rejoin.rs | 82 +++++++++++++++++++++++++++++++++++ 2 files changed, 86 insertions(+), 2 deletions(-) diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index b9b9765d44..641edba19c 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -753,7 +753,9 @@ impl TrackState { /// Expire an ended track's idle groups, whose closed channel refuses the write /// [`Self::evict_expired_scan`] takes: abort them in place, releasing their frames, - /// and leave the slots, which every read path already skips. + /// and leave the slots, which every read path already skips. A group still shared with a + /// live track (a warm copy adopts the relay copy's groups) is left to that track, whose + /// own eviction keeps its latest group. pub(super) fn expire_closed(&self, scan: ExpiryScan) { for (sequence, stamp) in &self.evict { let Some(slot) = self.lookup.get(sequence) else { @@ -766,7 +768,7 @@ impl TrackState { .cache_accessed_tick(scan.gc.then_some(scan.now)) .is_some_and(|tick| scan.now.saturating_sub(tick) > scan.max_ticks) { - let _ = slot.group.clone().abort(Error::Old); + let _ = slot.group.abort_if_last(Error::Old); } } } diff --git a/rs/moq-net/tests/rejoin.rs b/rs/moq-net/tests/rejoin.rs index a7391e4620..a0bd6702e5 100644 --- a/rs/moq-net/tests/rejoin.rs +++ b/rs/moq-net/tests/rejoin.rs @@ -108,3 +108,85 @@ async fn rejoin_recovers_the_group_reset_on_leave() { } } } + +/// A reader leaving long after its last read leaves the relay holding the latest group. +/// +/// A catalog's latest group can sit unread past the cache's idle expiry. A downstream +/// session that read it and then leaves must not take the relay's copy of that group with +/// it: a reader that joins afterwards needs it, and no newer group may ever come. +#[tokio::test(start_paused = true)] +async fn leaving_after_the_cache_window_keeps_the_latest_group() { + for version in [ + "moq-lite-05", + "moq-lite-06", + "moq-lite-07-wip", + "moq-transport-19", + "moq-transport-22", + ] { + tokio::time::timeout(TEST_TIMEOUT + moq_net::cache::DEFAULT_EXPIRY * 3, async { + let version: Version = version.parse().unwrap(); + let publisher = produce_origin(1); + let relay = produce_origin(2); + + let broadcast = publisher.create_broadcast("bench").unwrap(); + let track = broadcast.create_track("catalog", None).unwrap(); + broadcast.announce(Default::default()).unwrap(); + let mut group = track.append_group().unwrap(); + group.write_frame(Timestamp::ZERO, b"c0".as_ref()).unwrap(); + group.finish().unwrap(); + + let mut options = MockConnectOptions::new(version); + options.server_publish = Some(publisher.consume()); + options.client_subscribe = Some(relay.clone()); + let _upstream = connect_mock(options).await; + + // Each session gets its own hop, so the relay serves it through its own front. + let session = |hop| { + let relay = relay.clone(); + async move { + let client = produce_origin(hop); + let mut options = MockConnectOptions::new(version); + options.server_publish = Some(relay.consume()); + options.client_subscribe = Some(client.clone()); + let pair = connect_mock(options).await; + let consumer = client.consume(); + consumer.routed("bench").await.unwrap(); + let remote = consumer.request_broadcast("bench").await.unwrap(); + (remote, pair) + } + }; + + // Holds the relay's upstream subscription, and its copy, open throughout. + let (holder, _holder) = session(3).await; + let mut held = holder.track("catalog").unwrap().subscribe(None).await.unwrap(); + assert_eq!(held.recv_group().await.unwrap().unwrap().sequence, 0, "{version}"); + + let (leaver, leaver_session) = session(4).await; + let mut sub = leaver.track("catalog").unwrap().subscribe(None).await.unwrap(); + let mut group = sub.recv_group().await.unwrap().unwrap(); + assert_eq!(read_all(&mut group).await.unwrap(), [b"c0"], "{version}"); + tokio::time::sleep(moq_net::cache::DEFAULT_EXPIRY * 2).await; + drop((group, sub, leaver, leaver_session)); + + // Right after the leave, and again once the leaver's front has lingered and let go + // of the track (`IDLE_LINGER` is as long as the cache window). + for (hop, after) in [ + (5, Duration::from_millis(100)), + (6, moq_net::cache::DEFAULT_EXPIRY + Duration::from_secs(1)), + ] { + tokio::time::sleep(after).await; + let (later, _later) = session(hop).await; + let mut sub = later.track("catalog").unwrap().subscribe(None).await.unwrap(); + let group = tokio::time::timeout(Duration::from_secs(1), sub.recv_group()).await; + let mut group = group + .unwrap_or_else(|_| panic!("{version}: reader {hop} never got the latest group")) + .unwrap() + .unwrap(); + assert_eq!(group.sequence, 0, "{version}"); + assert_eq!(read_all(&mut group).await.unwrap(), [b"c0"], "{version}"); + } + }) + .await + .unwrap_or_else(|_| panic!("{version}: timed out")); + } +} From 2564923af42194bf57fe845c1b2a5f3f3fd18420 Mon Sep 17 00:00:00 2001 From: OneTooMany Date: Fri, 9 Oct 2026 11:43:14 +0200 Subject: [PATCH 6/7] test(net): pin when ended tracks sharing a group release it Two ended tracks that share a group each see the other's handle, so neither aborts it; it expires once one of them lets go. Say so where expire_closed skips a shared group. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/track.rs | 44 ++++++++++++++++++++++++++++++++--- 1 file changed, 41 insertions(+), 3 deletions(-) diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 641edba19c..e129507b9e 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -753,9 +753,10 @@ impl TrackState { /// Expire an ended track's idle groups, whose closed channel refuses the write /// [`Self::evict_expired_scan`] takes: abort them in place, releasing their frames, - /// and leave the slots, which every read path already skips. A group still shared with a - /// live track (a warm copy adopts the relay copy's groups) is left to that track, whose - /// own eviction keeps its latest group. + /// and leave the slots, which every read path already skips. A group another handle + /// still owns (a warm copy shares the relay copy's) is left to that owner, which keeps + /// it while it is a live track's latest; a later sweep retries it once this track holds + /// the last handle. pub(super) fn expire_closed(&self, scan: ExpiryScan) { for (sequence, stamp) in &self.evict { let Some(slot) = self.lookup.get(sequence) else { @@ -7279,6 +7280,43 @@ mod test { ); } + /// A group two ended tracks share expires once one of them lets go. Until then each + /// sees the other's handle, so neither aborts it for both. + #[tokio::test] + async fn closed_tracks_sharing_a_group_expire_it_once_one_lets_go() { + let pool = cache::Pool::new(cache::Config::default().with_expiry(Duration::from_secs(1))); + + let first = track_producer_pooled("first", pool.clone()); + let group = first.append_group().unwrap(); + group.finish().unwrap(); + let mut second = track_producer_pooled("second", pool.clone()); + second.adopt_group(group.clone(), true).unwrap(); + drop(group); + + first.finish().unwrap(); + second.finish().unwrap(); + let stale_first = first.consume(); + let stale_second = second.consume(); + drop((first, second)); + + // The first pass dates the activity it has not seen yet; the next one expires it. + let sweep = || { + for _ in 0..2 { + crate::model::clock::advance(Duration::from_secs(2)); + pool.sweep(); + } + }; + sweep(); + assert!( + stale_second.peek_group(0).is_some(), + "a shared group outlives the sweep" + ); + + drop(stale_first); + sweep(); + assert!(stale_second.peek_group(0).is_none(), "the last owner expires it"); + } + /// An abort after every group below the declared end finished leaves the end standing. #[tokio::test] async fn abort_after_the_end_settles_ends_clean() { From f9b7982e018f40a225149579dd4ca61ecf9cef13 Mon Sep 17 00:00:00 2001 From: OneTooMany Date: Fri, 9 Oct 2026 12:02:59 +0200 Subject: [PATCH 7/7] fix(net): an ended track expires a shared group unless it is a live edge abort_if_last kept an ended warm copy from aborting the relay's live edge, but it also kept every group two ended tracks share: each saw the other's handle, so stale consumers on both pinned it. Count on the group the live tracks holding it as their latest, through a guard in the latest slot that goes on demotion, removal or close_cache. expire_closed aborts an idle group unless that count is above zero, which is the pool's rule that only a live track's latest group is exempt, applied across the tracks that share it. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/cache.rs | 27 ++++++++++++ rs/moq-net/src/model/group.rs | 10 +++++ rs/moq-net/src/model/track.rs | 77 ++++++++++++++++++++++------------- 3 files changed, 86 insertions(+), 28 deletions(-) diff --git a/rs/moq-net/src/model/cache.rs b/rs/moq-net/src/model/cache.rs index b4f85cea3c..dd86719ca3 100644 --- a/rs/moq-net/src/model/cache.rs +++ b/rs/moq-net/src/model/cache.rs @@ -672,6 +672,8 @@ pub(crate) struct Charge { pub(crate) struct Access { stamp: AtomicU64, expires: AtomicU64, + // Live tracks holding the group as their latest; see [`Protection`]. + protected: AtomicUsize, } impl Access { @@ -679,9 +681,15 @@ impl Access { Self { stamp: AtomicU64::new(stamp), expires: AtomicU64::new(u64::MAX), + protected: AtomicUsize::new(0), } } + /// Whether a live track holds the group as its latest. + pub(crate) fn is_protected(&self) -> bool { + self.protected.load(Ordering::Relaxed) > 0 + } + /// The stamp, tie-breaking bits included. pub(crate) fn get(&self) -> u64 { self.stamp.load(Ordering::Relaxed) @@ -715,6 +723,25 @@ impl Access { } } +/// A live track holding a group as its latest, for as long as this lives. +/// +/// Tracks can share a group (a warm copy adopts the relay copy's), so the group itself +/// counts these: an ended track sharing it must not expire another track's live edge. +pub(crate) struct Protection(Arc); + +impl Protection { + pub(crate) fn new(access: &Arc) -> Self { + access.protected.fetch_add(1, Ordering::Relaxed); + Self(access.clone()) + } +} + +impl Drop for Protection { + fn drop(&mut self) { + self.0.protected.fetch_sub(1, Ordering::Relaxed); + } +} + impl Charge { /// Charge `n` more payload bytes, counting them as written. /// diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index db1e74f5a7..2f16d11182 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -911,6 +911,16 @@ impl Producer { self.alive.access.tick(now) } + /// Hold the group as a live track's latest until the guard drops. + pub(crate) fn cache_protect(&self) -> cache::Protection { + cache::Protection::new(&self.alive.access) + } + + /// Whether a live track holds the group as its latest. + pub(crate) fn cache_protected(&self) -> bool { + self.alive.access.is_protected() + } + /// Enter the group into the evictable population: demoted from the live edge, /// or inserted behind it. Idempotent; a no-op once the group is closed. pub(crate) fn cache_demote(&self) { diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index e129507b9e..4b258985b9 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -284,6 +284,10 @@ struct Slot { // Whether this incarnation came from the live publisher and can replace older // subscription content. Fetch-only backfill stays cached but never anchors drift. visible: bool, + + // Held while this is the live edge (see `TrackState::protects`), so an ended track + // sharing the group leaves it alone. + protection: Option, } /// Heap the track keeps per cached group, excluding the group itself @@ -753,10 +757,9 @@ impl TrackState { /// Expire an ended track's idle groups, whose closed channel refuses the write /// [`Self::evict_expired_scan`] takes: abort them in place, releasing their frames, - /// and leave the slots, which every read path already skips. A group another handle - /// still owns (a warm copy shares the relay copy's) is left to that owner, which keeps - /// it while it is a live track's latest; a later sweep retries it once this track holds - /// the last handle. + /// and leave the slots, which every read path already skips. A group another track + /// still holds as its live edge stays, since tracks can share a group (a warm copy + /// adopts the relay copy's). pub(super) fn expire_closed(&self, scan: ExpiryScan) { for (sequence, stamp) in &self.evict { let Some(slot) = self.lookup.get(sequence) else { @@ -764,12 +767,13 @@ impl TrackState { }; if slot.stamp == *stamp && !slot.group.is_aborted() + && !slot.group.cache_protected() && slot .group .cache_accessed_tick(scan.gc.then_some(scan.now)) .is_some_and(|tick| scan.now.saturating_sub(tick) > scan.max_ticks) { - let _ = slot.group.abort_if_last(Error::Old); + let _ = slot.group.clone().abort(Error::Old); } } } @@ -864,9 +868,10 @@ impl TrackState { return; } if let Some(latest) = self.latest_group - && let Some(slot) = self.lookup.get(&latest) + && let Some(slot) = self.lookup.get_mut(&latest) { slot.group.cache_demote(); + slot.protection = None; self.evict.push_back((latest, slot.stamp)); } } @@ -948,9 +953,10 @@ impl TrackState { // pool's access average) like any other cached group. if let Some(latest) = self.latest_group && sequence > latest - && let Some(prev) = self.lookup.get(&latest) + && let Some(prev) = self.lookup.get_mut(&latest) { prev.group.cache_demote(); + prev.protection = None; self.evict.push_back((latest, prev.stamp)); } self.latest_group = Some(sequence); @@ -967,6 +973,7 @@ impl TrackState { group: group.clone(), stamp, visible, + protection: self.protects(sequence).then(|| group.cache_protect()), }, ); if visible { @@ -7280,25 +7287,12 @@ mod test { ); } - /// A group two ended tracks share expires once one of them lets go. Until then each - /// sees the other's handle, so neither aborts it for both. + /// An ended track that shares a group with a live one leaves it while it is the live + /// track's latest, and expires it like any other idle group once it is not. A group + /// only ended tracks share expires even while stale consumers keep them all. #[tokio::test] - async fn closed_tracks_sharing_a_group_expire_it_once_one_lets_go() { + async fn closed_track_expires_a_shared_group_unless_live_edge() { let pool = cache::Pool::new(cache::Config::default().with_expiry(Duration::from_secs(1))); - - let first = track_producer_pooled("first", pool.clone()); - let group = first.append_group().unwrap(); - group.finish().unwrap(); - let mut second = track_producer_pooled("second", pool.clone()); - second.adopt_group(group.clone(), true).unwrap(); - drop(group); - - first.finish().unwrap(); - second.finish().unwrap(); - let stale_first = first.consume(); - let stale_second = second.consume(); - drop((first, second)); - // The first pass dates the activity it has not seen yet; the next one expires it. let sweep = || { for _ in 0..2 { @@ -7306,15 +7300,42 @@ mod test { pool.sweep(); } }; + let adopt = |name: &str, group: &group::Producer| { + let mut track = track_producer_pooled(name, pool.clone()); + track.adopt_group(group.clone(), true).unwrap(); + track.finish().unwrap(); + let stale = track.consume(); + drop(track); + stale + }; + + let live = track_producer_pooled("live", pool.clone()); + let group = live.append_group().unwrap(); + group.finish().unwrap(); + let stale = adopt("ended", &group); + drop(group); + sweep(); + assert!(live.consume().peek_group(0).is_some(), "the live edge stays"); assert!( - stale_second.peek_group(0).is_some(), - "a shared group outlives the sweep" + stale.peek_group(0).is_some(), + "and so does the ended track's share of it" ); - drop(stale_first); + live.append_group().unwrap().finish().unwrap(); + sweep(); + assert!(stale.peek_group(0).is_none(), "once demoted, it expires"); + + let group = live.append_group().unwrap(); + group.finish().unwrap(); + let (first, second) = (adopt("first", &group), adopt("second", &group)); + drop(group); + live.finish().unwrap(); + drop(live); + sweep(); - assert!(stale_second.peek_group(0).is_none(), "the last owner expires it"); + assert!(first.peek_group(2).is_none(), "ended tracks alone don't keep it"); + assert!(second.peek_group(2).is_none(), "for each other"); } /// An abort after every group below the declared end finished leaves the end standing.