diff --git a/doc/bin/cli.md b/doc/bin/cli.md index edd7ae8477..4a71053dc9 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -337,7 +337,7 @@ zero-based `frame` and padded standard base64. `` is the literal track name. `/fetch` splits its path on the last `/`, so the two agree only for names without one. Fetch only dials `--connect`, and -refuses any listener, cluster, auth, or `--hop` flag. It gives up after 30 +refuses any listener, cluster, auth, or `--epoch` flag. It gives up after 30 seconds, as `/fetch` does, and exits non-zero when the broadcast or group is not found (before writing anything), the relay refuses, or the deadline passes. @@ -390,11 +390,28 @@ moq --connect https://relay.example.com/anon \ ## Redundant publishers -Two publishers of the same broadcast name are interchangeable sources: -relays hold both routes and fail over between them mid-group. They must -produce identical tracks with aligned groups. A restarted encoder is the same -broadcast too, so one whose groups restart from 0 must publish under a new -name, or viewers wait for its sequence to catch up. +Each run announces a fresh epoch, so a restarted publisher replaces the old +broadcast instead of resuming into it. Two publishers that pass the same +`--epoch` (a lowercase, hyphenated UUIDv7, such as util-linux `uuidgen -7` +prints) are interchangeable sources: relays hold both routes and fail over +between them mid-group. They must produce identical tracks with aligned groups, +which no importer guarantees yet (see +[#4352](https://github.com/moq-dev/moq/issues/4352) and +[#4354](https://github.com/moq-dev/moq/issues/4354)). A member restarted with +the same epoch rejoins as the same content, so if its groups restart from 0, +viewers wait for its sequence to catch up. + +```bash +EPOCH=$(uuidgen -7) +# On each of the two hosts: +moq --connect https://relay.example.com/anon --broadcast event.hang --epoch "$EPOCH" import ts < feed.ts +``` + +`--epoch` applies to the sources announced once per run. The RTMP, SRT, and +WHIP ingests and `import ts --program all` announce their own, so they refuse it. + +`--epoch` replaces the removed `--hop`. `MOQ_HOP` is no longer read, so drop it +from any deployment and pass `MOQ_EPOCH` instead. ## Cluster diff --git a/doc/bin/relay/cluster.md b/doc/bin/relay/cluster.md index 5aba2c0ff5..625ba578ca 100644 --- a/doc/bin/relay/cluster.md +++ b/doc/bin/relay/cluster.md @@ -58,12 +58,6 @@ A route whose original publisher (its first hop) changes is updated in place on both wire protocols, so the broadcast never briefly vanishes downstream, and subscriptions in flight carry on through it. -A publisher whose protocol names no hop (moq-transport without the cluster -extension, moq-lite 01 through 03, or a peer that sends 0) gets a random first -hop from the relay it connects to, fresh for each connection, followed by a 0. -Its reconnect is therefore a new first hop downstream, a reprice on the same -connection stays in place, and the 0 keeps it ranked as anonymous. - ## Topology List the peers each relay dials. That's the whole topology: a relay dials only diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index ff5e6d433c..38c7f2fe1f 100644 --- a/doc/concept/moq-lite.md +++ b/doc/concept/moq-lite.md @@ -100,8 +100,7 @@ through, which is how forwarding loops are caught, and a cost, which is how a subscriber picks among several routes to the same broadcast. A route may also carry a [publisher epoch](#publisher-epochs), which says which routes serve the same bytes. A hop of 0 is -the anonymous mark and travels the chain unchanged; when it is the first hop, a -relay puts a random ID, fresh per connection, in front of it. A route that passed through an +the anonymous mark and travels the chain unchanged. A route that passed through an anonymous hop at any depth ranks below every fully identified route, whatever the costs say; among anonymous routes, cost keeps ordering. diff --git a/doc/setup/upgrade.md b/doc/setup/upgrade.md index de3dc8c821..739ce701e2 100644 --- a/doc/setup/upgrade.md +++ b/doc/setup/upgrade.md @@ -47,6 +47,16 @@ These land with the next breaking release, not the 2026-09-23 train. `producer.demand().used` and `.unused()`, as are `Group.Producer`'s, and `Allocator.reserve` takes `producer.demand()`, replacing the `Bandwidth.Demand` interface. +- **`moq --hop` is removed; `--epoch` replaces it.** A redundant pair shares an + epoch, a UUIDv7 such as `uuidgen -7` prints, instead of a Hop ID: pass the + same `--epoch` (or `MOQ_EPOCH`) to both. Unlike a rename, `--hop` is now an + unknown flag and `MOQ_HOP` is silently ignored, so remove it from any + deployment: a pair still keyed on `MOQ_HOP` mints an epoch per process, and + whenever either member starts or restarts, its new epoch replaces the other's + broadcast and restarts its viewers. `--cluster-id` no longer falls + back to `--hop`, so a node that pinned its Hop ID with `--hop` passes + `--cluster-id`. Relays no longer put a random hop in front of a route that + names no publisher; it keeps its 0. - **The `"auto"` delay is measured, not derived from RTT** (#4162). It is sized from how late frames arrive (see [audio jitter](/concept/audio-jitter)) in `@moq/watch` and `moq play`, which now defaults `--delay` to `auto` instead of @@ -185,7 +195,7 @@ variables follow the flag (`MOQ_SERVER_BIND` is `MOQ_LISTEN`). | `--cluster-linger` | removed; a broadcast closes when its last publisher is lost | | `--cluster-connect host:port` | a full URL, `https://host/?jwt=TOKEN` | | `--cluster-mesh`, TOML `mesh` | removed; list every peer with `--cluster-connect` or `--cluster-connect-api` | -| `moq --origin`, `--name`, `--latency-max` | `--hop`, `--broadcast`, `--max-age` | +| `moq --origin`, `--name`, `--latency-max` | `--hop` (removed for `--epoch` after [Unreleased](#unreleased)), `--broadcast`, `--max-age` | | `moq publish`, `moq subscribe` | `moq import`, `moq export` | | `moq token`, the `moq-token` binary | `moq auth` | diff --git a/drafts/draft-lcurley-moq-cluster.md b/drafts/draft-lcurley-moq-cluster.md index abf04226e8..91f1b13ed2 100644 --- a/drafts/draft-lcurley-moq-cluster.md +++ b/drafts/draft-lcurley-moq-cluster.md @@ -108,7 +108,6 @@ It stands for an endpoint that did not negotiate this extension, and an endpoint Since any number of endpoints can be 0, it identifies nothing: - **Loop detection**: 0 in a HOP_PATH is never a loop. A receiver whose own Hop ID is 0 cannot detect loops through itself and MUST NOT discard an advertisement merely because the path contains 0. -- **Unknown publisher**: an advertisement whose first entry is 0 has an unknown publisher, which a relay names by putting a stamp of its own in front ({{stamping}}). - **Filtering**: a peer that declared 0 gave the receiver nothing to filter that session on. The receiver MAY assign an ID of its own ({{assigned}}) as local selection state and MUST NOT write it into HOP_PATH. Duplicate *non-zero* Hop IDs in one HOP_PATH are a loop; duplicate zeros are not. @@ -120,7 +119,7 @@ It uses that ID as local selection state: as what it filters that session on, in The ID is the receiver's own, not the peer's, and MUST NOT be forwarded. An advertisement that arrives with its own HOP_PATH already names the sender there, as 0 if withheld. -An unnamed publisher is stamped instead ({{stamping}}), which is not an assigned identity. +A receiver writes 0 for an upstream that sent no HOP_PATH ({{bridging}}). An assigned ID MUST NOT be shared between peers not known to be the same endpoint. Sharing one suppresses each one's advertisements to the other, so two unrelated peers would starve each other of routes. @@ -128,14 +127,6 @@ Sharing one suppresses each one's advertisements to the other, so two unrelated A peer the receiver authenticated, or dialed and therefore chose, SHOULD get one stable ID, so its reconnects and redundant sessions are filtered as one peer; a fresh ID per connection would make one peer look like several. An anonymous accepted session cannot be correlated with anything, so it SHOULD get a distinct ID per session: not an identity, but enough to keep routes learned from it from being advertised back to it, which is the loop 0 cannot prevent. -## Stamping an Unknown Publisher {#stamping} -A relay that records an advertisement whose first HOP_PATH entry is 0 MUST insert a random non-zero Hop ID in front of it, picked once for the session the advertisement arrived on. -One that arrived with no HOP_PATH ({{bridging}}) becomes that stamp followed by 0. - -A fresh stamp per session keeps unnamed publishers on different sessions apart, while an update on the same session keeps the stamp and applies in place. -The 0 behind the stamp keeps the path ranked below fully identified ones ({{selection}}), since the unnamed hop may hide any depth. -A stamp names a session, not a peer: it MUST NOT be reused across sessions or taken as the peer's identity. - # Namespace Advertisements {#namespace} HOP_PATH and ROUTE_COST are Key-Value-Pair parameters ({{moqt}} Section 2.5). @@ -198,7 +189,7 @@ Unknown, out-of-budget, and saturated routes are outside the guarantee: a receiv # Relay Behavior A relay forwarding an advertisement MUST append its own Hop ID to the HOP_PATH it received, so its ID is always the last entry. -A received 0 is forwarded unchanged, behind a stamp when it is the first entry ({{stamping}}). +A received 0 is forwarded unchanged. A relay MUST discard an advertisement whose HOP_PATH already contains its own non-zero Hop ID: forwarding it would extend a loop, and subscribing through it would route the relay back to itself. This check catches loops of any length and is the only loop defense required. @@ -206,7 +197,7 @@ A conforming sender never sends one ({{selection}}), so a receiver MAY close the ## Bridging {#bridging} An upstream that did not negotiate the extension sends no HOP_PATH. -The relay creates one holding its stamp for that session followed by 0 ({{stamping}}), then appends its own Hop ID. +The relay creates one with a single 0 entry for that upstream ({{zero}}), then appends its own Hop ID. The identity a receiver assigned that upstream ({{assigned}}) is local selection state and MUST NOT appear in HOP_PATH. ## Accumulating Cost {#accumulating} @@ -324,7 +315,6 @@ The Key-Value-Pair parity is load-bearing: HOP_PATH is odd, so its value is a le - Defined request resolution against the longest covering prefix; every refusal is terminal, including between several publishers of one namespace, and capacity is signaled only by withdrawing or re-pricing the advertisement. - A relay does not advertise a namespace because it resolved it; the publisher advertises the concrete namespace once producing. - A change of original publisher is an ordinary update instead of a withdrawal and a new advertisement. -- A relay puts a random Hop ID, picked per session, in front of an advertisement whose first HOP_PATH entry is 0, and writes that stamp followed by 0 for one with no HOP_PATH. ## moq-cluster-01 - Assigned identities are local selection state and MUST NOT be forwarded. diff --git a/drafts/draft-lcurley-moq-lite.md b/drafts/draft-lcurley-moq-lite.md index b5d6ab02fe..097b7c3f4a 100644 --- a/drafts/draft-lcurley-moq-lite.md +++ b/drafts/draft-lcurley-moq-lite.md @@ -411,9 +411,6 @@ This is the only loop defense moq-lite requires, and it catches loops of any len A conforming sender never sends one (see below), so a receiver MAY instead close the session with a protocol violation; discarding is what keeps a mesh working when one member does not conform. A Hop ID of 0 means unknown and never matches anything; withholding an ID trades loop detection for privacy. A receiver MAY assign an identity of its own to a peer that declared 0, as local selection state for filtering that session; it MUST NOT forward that identity as a Hop ID. -A relay that records an announcement whose reconstructed path starts with 0 MUST insert a random non-zero Hop ID in front of it, picked once for the session it arrived on; an empty path becomes that stamp followed by 0. -Nothing on the wire says whether an unnamed publisher that reconnects is the same one, so its reconnect reads downstream as a new first hop, while an ANNOUNCE_UPDATE on the same session keeps the stamp. -The 0 behind the stamp keeps the path ranked below fully identified ones. A publisher MUST NOT advertise a path whose entries contain the Hop ID the subscriber declared in its SETUP (see [Hop Parameter](#hop-parameter)). The receiver can only discard it, and acting on it would form a loop, so sending one is never useful. @@ -822,7 +819,7 @@ Both endpoints send it and the two values need not match: the parameter prices t A declared cost is an assertion, not an instruction: a receiver MAY charge a locally configured value instead, so a peer cannot reprice its neighbours by declaring itself cheap. ### Hop Parameter {#hop-parameter} -The Hop Parameter declares the sender's Hop ID: the identity it stamps onto announcements it forwards. +The Hop Parameter declares the sender's Hop ID: the identity that names it in the paths of announcements it forwards. The Parameter Value is a variable-length integer; a value of 0 carries no identity and is equivalent to omitting the parameter. Declaring it at setup gives the receiver the peer's identity before any other stream arrives, so route selection applies the same exclusion to the peer's subscriptions as to its announcements (see [Routing](#routing)), even on a session that never opens an Announce Stream. @@ -935,10 +932,9 @@ A receiver MUST close the stream with a PROTOCOL_VIOLATION if the Hop Count does A unique identifier for each relay in the path from the origin publisher, ordered from origin to the upstream of the responding publisher. The responding publisher's own Hop ID is NOT included in the resolved list; it is carried once in ANNOUNCE_OK, so the total path length is the resolved list's length plus 1 (`Hop Count + Hop Keep + 1`). When forwarding an announcement received from an upstream peer, a relay MUST append the upstream peer's ANNOUNCE_OK `Hop ID` to the resolved list, since that ID is no longer implicit downstream. -The first entry of the reconstructed path identifies the endpoint that originated the route. A Hop ID value of 0 means the hop is unknown: either it was never assigned or a relay deliberately withholds it (see [Routing](#routing)). -A received 0 is forwarded unchanged, behind a stamp when it is the first entry (see [Routing](#routing)). -When bridging an announcement from an upstream that sent no hop list, a relay writes its stamp followed by 0 for that hop. +A received 0 is forwarded unchanged. +When bridging an announcement from an upstream that sent no hop list, a relay writes 0 for that hop. An identity a receiver assigned that upstream is local selection state and MUST NOT be forwarded as a Hop ID. A receiver MUST close the session with a PROTOCOL_VIOLATION if a non-zero Hop ID appears twice in the resolved list. @@ -1395,7 +1391,6 @@ The `Message Length` describes the payload size on the wire. - Added the Spread Hash tie-break after the shortest path: a hash of the requested path and the route's Hop IDs, so equal-cost advertisers of one prefix share its paths. - Added announce compression: ANNOUNCE_START gains `Path Base` and `Path Keep` to copy the head of a live advertisement's suffix, and ANNOUNCE_START and ANNOUNCE_UPDATE gain `Hop Base` and `Hop Keep` to copy the tail of a live advertisement's Hop ID list. - Capped the SETUP Message Length at 65,536 bytes. -- A relay puts a random Hop ID, picked per session, in front of an announcement whose reconstructed path starts with 0, and writes that stamp followed by 0 for an empty path. ## moq-lite-06 diff --git a/js/net/src/hop.ts b/js/net/src/hop.ts index fe38b8de6c..736a543e40 100644 --- a/js/net/src/hop.ts +++ b/js/net/src/hop.ts @@ -67,20 +67,6 @@ export function randomHop(): Hop { return HopSchema.parse(raw === 0n ? 1n : raw); } -/** - * Name an unknown original publisher: put `stamp`, the receiving connection's own random id, - * in front of a chain that starts with 0, and turn an empty chain into `[stamp, 0]`. A - * publisher that reconnects then reads downstream as a new first hop, while the 0 after the - * stamp keeps the route ranked as anonymous. `undefined` if the chain is full or already holds - * `stamp`. Mirrors `Hops::stamp` in rs/moq-net. - */ -export function stampHops(hops: readonly Hop[], stamp: Hop): Hop[] | undefined { - if (hops.length === 0) return [stamp, UNKNOWN_HOP]; - if (hops[0] !== UNKNOWN_HOP) return [...hops]; - if (hops.length >= MAX_HOPS || hops.includes(stamp)) return undefined; - return [stamp, ...hops]; -} - /** * What pulling content via a route costs, in two magnitudes accumulated together * and compared in that order: lower {@link Cost.warm} wins, and {@link Cost.cold} diff --git a/js/net/src/ietf/subscriber.test.ts b/js/net/src/ietf/subscriber.test.ts index ff6415c9d5..8ff3f945d8 100644 --- a/js/net/src/ietf/subscriber.test.ts +++ b/js/net/src/ietf/subscriber.test.ts @@ -116,30 +116,21 @@ test("an unsolicited announcement lands", async () => { }); /** - * A session without the Cluster extension names no publisher, so each connection stamps its - * own random Hop ID in front of a 0: a publisher that reconnects reads as a new one, and the 0 - * keeps it ranked below identified routes. + * A session without the Cluster extension names no publisher, so its advertisement carries + * only the anonymous mark: identity is the epoch's job, and nothing goes in front of the 0. */ -test("an advertisement with no path is stamped per connection", async () => { - const stamp = async () => { - const pair = createMockTransportPair(ALPN.DRAFT_19); - const subscriber = new Subscriber({ session: new NativeSession(pair.server, VERSION, true) }); - const announced = subscriber.announced(); - expect(await nextStream(pair.client)).toBeDefined(); - - const stream = await Stream.open(pair.server, { version: VERSION }); - void subscriber.runPublishNamespace( - new PublishNamespace({ requestId: 0n, trackNamespace: Path.from("legacy") }), - stream, - ); - const hops = (await announced.next())?.route.hops ?? []; - expect(hops).toHaveLength(2); - expect(hops[0]).not.toBe(UNKNOWN_HOP); - expect(hops[1]).toBe(UNKNOWN_HOP); - return hops[0]; - }; +test("an advertisement with no path is anonymous", async () => { + const pair = createMockTransportPair(ALPN.DRAFT_19); + const subscriber = new Subscriber({ session: new NativeSession(pair.server, VERSION, true) }); + const announced = subscriber.announced(); + expect(await nextStream(pair.client)).toBeDefined(); - expect(await stamp()).not.toBe(await stamp()); + const stream = await Stream.open(pair.server, { version: VERSION }); + void subscriber.runPublishNamespace( + new PublishNamespace({ requestId: 0n, trackNamespace: Path.from("legacy") }), + stream, + ); + expect((await announced.next())?.route.hops).toEqual([UNKNOWN_HOP]); }); /** diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts index dbffb1c1f7..f8464052d7 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/js/net/src/ietf/subscriber.ts @@ -13,7 +13,7 @@ import { sessionCause, } from "../error.ts"; import * as netGroup from "../group.ts"; -import { Cost, type Route, randomHop, routesEqual, stampHops, UNKNOWN_HOP } from "../hop.ts"; +import { Cost, type Route, routesEqual, UNKNOWN_HOP } from "../hop.ts"; import { hiddenBelow, hooks, scopeCaptures, scopeHead, scopeOverlaps } from "../internal.ts"; import * as Path from "../path.ts"; import type { Cursor, Reader, Stream } from "../stream.ts"; @@ -130,10 +130,6 @@ export class Subscriber { // Dedup consumed broadcasts per path: repeat consume() calls share one subscription. #consumes = new BroadcastCache(); - // A random Hop ID of this connection's own, written as the first hop of any path that - // names no publisher, so a publisher that reconnects reads as a new one. - #stamp = randomHop(); - // Paths with a legacy PUBLISH_NAMESPACE request in flight, reserved synchronously. // The count below is only taken once the OK is written, and two requests that both // got past the duplicate check before either attached would both take one. @@ -197,16 +193,10 @@ export class Subscriber { return advert !== undefined && this.#cluster !== undefined && Cluster.loops(advert, this.#cluster.self); } - /** - * The route an advertisement carries; one without a path is free. A path that names no - * publisher, or none at all, gets this connection's stamp in front of a 0. - */ + /** The route an advertisement carries; one without a path is anonymous and free. */ #route(advert: Cluster.Advert | undefined): Route { - if (advert === undefined) return { hops: [this.#stamp, UNKNOWN_HOP], cost: Cost.zero }; - // A full chain, or a stamp colliding with an entry (a 1-in-2^53 draw), keeps the - // path as sent. - const hops = stampHops(advert.hops, this.#stamp) ?? [...advert.hops]; - return { hops, cost: { warm: advert.cost, cold: advert.cost } }; + if (advert === undefined) return { hops: [UNKNOWN_HOP], cost: Cost.zero }; + return { hops: advert.hops, cost: { warm: advert.cost, cold: advert.cost } }; } /** diff --git a/js/net/src/lite/subscriber.test.ts b/js/net/src/lite/subscriber.test.ts index 1df08f571b..ca3deea098 100644 --- a/js/net/src/lite/subscriber.test.ts +++ b/js/net/src/lite/subscriber.test.ts @@ -187,35 +187,24 @@ test("an unidentified responder keeps hop 0 on a nonempty chain", async () => { subscriber.close(); }); -test("a received hop list naming no publisher is stamped per connection", async () => { - const first = async () => { - const { subscriber, send, settle } = announceHarness(Version.DRAFT_06); - const announced = subscriber.announced(); - await settle(); - - await send((w) => new AnnounceOk(UNKNOWN_HOP, 0).encode(w, Version.DRAFT_06)); - await send((w) => - encodeAnnounceBroadcast(w, { status: "active", suffix: Path.from("room"), hops: [] }, Version.DRAFT_06), - ); - const update = await announced.next(); - expect(update).toMatchObject({ prefix: Path.from("room"), kind: "start" }); - const hops = update?.route.hops ?? []; - expect(hops).toHaveLength(2); - expect(hops[0]).not.toBe(UNKNOWN_HOP); - // The 0 after the stamp keeps the unnamed publisher ranked as anonymous. - expect(hops[1]).toBe(UNKNOWN_HOP); - expect(update && isAnonymous(update.route)).toBe(true); - - announced.close(); - subscriber.close(); - return hops[0]; - }; +test("a received hop list naming no publisher stays anonymous", async () => { + const { subscriber, send, settle } = announceHarness(Version.DRAFT_06); + const announced = subscriber.announced(); + await settle(); - // A reconnect is a new connection, so the same unnamed publisher reads as a new one. - expect(await first()).not.toBe(await first()); + await send((w) => new AnnounceOk(UNKNOWN_HOP, 0).encode(w, Version.DRAFT_06)); + await send((w) => + encodeAnnounceBroadcast(w, { status: "active", suffix: Path.from("room"), hops: [] }, Version.DRAFT_06), + ); + const update = await announced.next(); + expect(update).toMatchObject({ prefix: Path.from("room"), kind: "start", route: { hops: [UNKNOWN_HOP] } }); + expect(update && isAnonymous(update.route)).toBe(true); + + announced.close(); + subscriber.close(); }); -test("a received chain starting with hop 0 keeps it behind the stamp", async () => { +test("a received chain starting with hop 0 is forwarded unchanged", async () => { const { subscriber, send, settle } = announceHarness(Version.DRAFT_06); const announced = subscriber.announced(); await settle(); @@ -230,9 +219,7 @@ test("a received chain starting with hop 0 keeps it behind the stamp", async () ); const update = await announced.next(); expect(update).toMatchObject({ prefix: Path.from("room"), kind: "start" }); - const [stamp, ...rest] = update?.route.hops ?? []; - expect(stamp).not.toBe(UNKNOWN_HOP); - expect(rest).toEqual([UNKNOWN_HOP, PUBLISHER_A, PEER]); + expect(update?.route.hops).toEqual([UNKNOWN_HOP, PUBLISHER_A, PEER]); announced.close(); subscriber.close(); @@ -353,9 +340,8 @@ test("a lite-05 duplicate announce follows the same restart rule", async () => { }); // A responder that withholds its Hop ID sends the reserved 0, and an empty chain means it -// originated the path itself, so the advertisement names nobody and the connection stamps it. -// A restart on the same connection keeps that stamp, so it is a reprice that updates in place -// and keeps the shared broadcast, as does one that names a publisher. +// originated the path itself, so the advertisement names nobody. A restart of it is a reprice +// that updates in place and keeps the shared broadcast, as does one that names a publisher. test("a restart from an unidentified publisher updates in place", async () => { const { subscriber, send, settle } = announceHarness(Version.DRAFT_06); const announced = subscriber.announced(); diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index 2a2ddb5f99..691f37bb4a 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -6,7 +6,7 @@ import { BroadcastCache } from "../consume.ts"; import type * as Epoch from "../epoch.ts"; import { controlTimeout, error, ProtocolViolation, reason, StreamCode, StreamError, sessionCause } from "../error.ts"; import * as netGroup from "../group.ts"; -import { Cost, type Hop, MAX_HOPS, type Route, randomHop, routesEqual, stampHops, UNKNOWN_HOP } from "../hop.ts"; +import { Cost, type Hop, MAX_HOPS, type Route, routesEqual, UNKNOWN_HOP } from "../hop.ts"; import { groupBounds, hiddenBelow, scopeCaptures, scopeHead, scopeOverlaps } from "../internal.ts"; import * as Path from "../path.ts"; import { type OpenOptions, type Reader, Stream } from "../stream.ts"; @@ -143,10 +143,6 @@ export class Subscriber { // once and asks for it on every request, so a later epoch never feeds an older handle. #epochs = new Map(); - // A random Hop ID of this connection's own, written as the first hop of any chain that - // names no publisher, so a publisher that reconnects reads as a new one. - #stamp: Hop; - // Dedup in-flight one-shot fetches, keyed by [broadcast, track, sequence]. Concurrent (or // repeat, while still open) fetchGroup() calls for the same group share one FETCH stream and // each get an independent mirror; the entry is evicted once the group closes. @@ -181,7 +177,6 @@ export class Subscriber { this.#quic = quic; this.version = version; this.hop = hop; - this.#stamp = randomHop(); this.#probe = probe; this.#peerSetup = peerSetup; } @@ -271,13 +266,13 @@ export class Subscriber { // they go on record and obey the same one-per-path rule: the initial set // naming a path twice is the same violation as two ANNOUNCE_STARTs for it, // and the record is what catches either. Draft01/02 carry no hop ids and no - // ANNOUNCE_OK, so this connection's stamp is the first hop. + // ANNOUNCE_OK, so nothing names the publisher. for (const suffix of init.suffixes) { const path = Path.join(prefix, suffix); if (advertised.has(path)) { throw new ProtocolViolation(`duplicate announce for ${path}`); } - const route = { hops: [this.#stamp, UNKNOWN_HOP], cost: Cost.zero }; + const route = { hops: [UNKNOWN_HOP], cost: Cost.zero }; const live = visible(path); const captures = scopeCaptures(scope, path); advertised.set(path, { live, route, captures }); @@ -413,18 +408,15 @@ export class Subscriber { continue; } - // The first hop is the original publisher; an empty chain means the peer itself - // originated it. One that names nobody (lite-01..03, or a peer reporting 0) - // gets this connection's stamp in front of its 0. - const fullHops = stampHops( + const fullHops = hops !== undefined && responderOrigin !== undefined ? [...hops, responderOrigin] - : [...(hops ?? [])], - this.#stamp, - ); + : [...(hops ?? [])]; + // A received empty list is the anonymous mark, not a local announcement. + if (fullHops.length === 0) fullHops.push(UNKNOWN_HOP); // Appending a withheld AnnounceOk(0) onto a 32-entry list is the same // drop Rust's Hops::push makes: do not expose an overlong chain. - if (fullHops === undefined || fullHops.length > MAX_HOPS) { + if (fullHops.length > MAX_HOPS) { console.debug(`announced: broadcast=${path} dropped (hop chain at MAX_HOPS)`); advertised.set(path, { live: false, diff --git a/js/net/src/origin.test.ts b/js/net/src/origin.test.ts index f1254022dd..73ecfa06fe 100644 --- a/js/net/src/origin.test.ts +++ b/js/net/src/origin.test.ts @@ -3,7 +3,7 @@ import { getter, Once, race, Signal } from "@moq/signals"; import { type Consumer as BroadcastConsumer, Producer as BroadcastProducer } from "./broadcast.ts"; import * as Epoch from "./epoch.ts"; import { StreamCode, StreamError } from "./error.ts"; -import { HopSchema, Route, stampHops } from "./hop.ts"; +import { HopSchema, Route } from "./hop.ts"; import { spreadHash } from "./internal.ts"; import type { Consumer, Table } from "./origin.ts"; import { Producer } from "./origin.ts"; @@ -121,23 +121,6 @@ test("unscoped broadcast getters share one fresh snapshot per mutation", () => { origin.close(); }); -test("a stamped route ranks below an identified one of the same length", () => { - const origin = new Producer(); - const path = Path.from("room/alice"); - const stamped = stampHops([], HopSchema.parse(5n)); - if (!stamped) throw new Error("an empty chain always has room"); - const identified = [HopSchema.parse(11n), PEER]; - expect(stamped).toHaveLength(identified.length); - - const legacy = wireOf(origin).receive(path, { hops: stamped, cost: 0n }); - const named = wireOf(origin).receive(path, { hops: identified, cost: 9n }); - expect(origin.consume().broadcasts().peek().get(path)).toEqual(Route.normalize({ hops: identified, cost: 9n })); - - legacy.close(); - named.close(); - origin.close(); -}); - test("an announced local path competes with a received route on cost", async () => { const origin = new Producer(); const consumer = origin.consume(); diff --git a/quest/m0/broadcast-epoch/README.md b/quest/m0/broadcast-epoch/README.md index 8f4ce2b0d0..7cefe29c1b 100644 --- a/quest/m0/broadcast-epoch/README.md +++ b/quest/m0/broadcast-epoch/README.md @@ -16,7 +16,7 @@ their routes carry no epoch, and see a restart as an end and start at the same path. Non-goals: pooling, which needs nothing here; a redundant pair shares an -explicit epoch through [`--hop` removal](/quest/m0/broadcast-epoch/hop-removal.md). +explicit epoch through `moq --epoch`. Also out of scope: trusting the publisher's clock (a far-future epoch wins until its route goes away). @@ -75,6 +75,5 @@ timeout, and killing the newest epoch falls back to a still-live older one. - [Gateways](/quest/m0/broadcast-epoch/gateways.md) - RTMP, SRT, and WHIP ingest mint an epoch per incoming connection, so an encoder reconnect is a clean takeover - [TS restart](/quest/m0/broadcast-epoch/ts-restart.md) - a signalled backward TS discontinuity finishes the broadcast and continues the same input under a fresh epoch - [Bindings](/quest/m0/broadcast-epoch/bindings.md) - moq-ffi and every wrapper expose the epoch and let a publisher announce one -- [Remove `--hop`](/quest/m0/broadcast-epoch/hop-removal.md) - `moq` takes an optional `--epoch` instead of `--hop`, a plain publisher declares a random Hop ID, and the per-session hop stamp is gone - [Stats epochs](/quest/m0/broadcast-epoch/stats-epoch.md) - moq-stats publishes each group announcement under its own epoch, so neither a restarted node nor a returning idle group stalls its viewers - [Stats totals and prefix tracks](/quest/m0/broadcast-epoch/stats-split.md) - the same release retires the per-path stats maps for totals and on-demand prefix tracks (decided 2026-10-05) diff --git a/quest/m0/broadcast-epoch/hop-removal.md b/quest/m0/broadcast-epoch/hop-removal.md deleted file mode 100644 index 56f5f310f9..0000000000 --- a/quest/m0/broadcast-epoch/hop-removal.md +++ /dev/null @@ -1,50 +0,0 @@ -# [M] Redundant publishers share an epoch, not a hop - -## Goal - -Nothing names a publisher's identity outside its epoch. `moq` loses `--hop`, -`MOQ_HOP`, and the hidden `--origin` alias; an optional `--epoch` takes their -place for publishing. Omitted, `moq` itself mints a fresh epoch per run, since -nothing mints one by default; a redundant pair passes the same value. A plain publisher declares -a random Hop ID per process with no flag, as the bindings and js already do, -so a multi-homed publisher still catches its own loops. `--cluster-id` names -a node and no longer falls back to `--hop`. Relays stop stamping an unnamed -publisher's hop chain. - -## Plan - -- Drop the per-session hop stamp for unnamed publishers (decided 2026-10-03: - split horizon rests on `via` and own-hop checks, and the stamp never - matched any relay's id, so it only existed for publisher identity). Delete - `Hops::stamp` (`rs/moq-net/src/model/origin.rs`), the `stamp` fields in - `rs/moq-net/src/{lite,ietf}/subscriber.rs`, `stampHops` in - `js/net/src/hop.ts` and its callers in `js/net/src/{lite,ietf}/subscriber.ts`, - and their tests, including `rs/moq-net/tests/legacy_reconnect.rs`. Keep the - leading 0, so an unnamed chain still ranks anonymous. -- Keep request-time exclusion (decided 2026-10-06): the relay still serves - each session through `excluding` its declared or assigned hop. Dropping it - would let a shared front hairpin through a peer in a cluster. Viewers share - a front anyway, because `Horizon::effective` ignores a hop no route chain - names (#4922). -- Drafts: lite-07 and cluster-02 are in progress, so edit their text and - changelogs in place. In `drafts/draft-lcurley-moq-lite.md`: the stamping - rule, "the identity it stamps", - and "the first entry identifies the endpoint that originated the route". In - `drafts/draft-lcurley-moq-cluster.md`: the unknown-publisher stamping text, - the relay behavior that stamps, and the changelog bullets. -- CLI: `rs/moq-cli/src/{args,complete,fetch,announced}.rs`. Relays keep `cluster.id` - as their node id. -- Docs: `doc/bin/cli.md` ("Redundant publishers" uses `--epoch`), the - migration row in `doc/setup/upgrade.md`, the stamping paragraph in - `doc/bin/relay/cluster.md`, and the stamping sentence in - `doc/concept/moq-lite.md`. Search the repo for `--hop`, `MOQ_HOP`, and - `--cluster-id` examples, including demo recipes. - -Open: whether anything declares an incumbent dead faster than the QUIC -idle timeout (a failover service that retracts it, or active-active delivery -to the relay). Neither is required to land this. - -Public API: removes `--hop`, `MOQ_HOP`, and `--origin`, adds `--epoch`, and -`--cluster-id` stops reading `--hop`. -Wire: relays stop stamping. - diff --git a/quest/m0/broadcast-epoch/unepoched-takeover.md b/quest/m0/broadcast-epoch/unepoched-takeover.md index 735090ff57..c551edb8d3 100644 --- a/quest/m0/broadcast-epoch/unepoched-takeover.md +++ b/quest/m0/broadcast-epoch/unepoched-takeover.md @@ -68,4 +68,3 @@ ends it when another route wins, and announces the change. ## Related - [Apps](/quest/m0/broadcast-epoch/apps.md) - #4970 drives `@moq/watch` and `moq play` from announcements, which this relies on -- [Remove `--hop`](/quest/m0/broadcast-epoch/hop-removal.md) - drops the per-session hop stamp, so restarted routes often have identical metadata diff --git a/quest/m1/cluster-routing/README.md b/quest/m1/cluster-routing/README.md index e2764f8ca2..1dceeb222a 100644 --- a/quest/m1/cluster-routing/README.md +++ b/quest/m1/cluster-routing/README.md @@ -139,7 +139,6 @@ Once every child has landed: ## Related - [P2P](/quest/m3/p2p/README.md) - browser and native peers that become routing nodes over this layer -- [Remove `--hop`](/quest/m0/broadcast-epoch/hop-removal.md) - redundant publishers share an explicit epoch - [One route cost](/quest/m1/route-cost.md) - Warm and Cold collapse to one static cost - [Same-epoch importers](/quest/m1/hop-aligned-import.md) - the importer half of a redundant pair under one explicit epoch - [Broadcast epochs](/quest/m0/broadcast-epoch/README.md) - a redundant pair shares one epoch diff --git a/quest/m1/hop-aligned-import.md b/quest/m1/hop-aligned-import.md index 3a1d425cf3..b589952e45 100644 --- a/quest/m1/hop-aligned-import.md +++ b/quest/m1/hop-aligned-import.md @@ -51,7 +51,6 @@ issues' 1+1 setup (one relay, two `import ts` sharing one `--epoch`, two ## Required - [Shared import clock](/quest/m1/shared-clock.md) - the `Input`/offset API this supplies an input-derived anchor through -- [Remove `--hop`](/quest/m0/broadcast-epoch/hop-removal.md) - the shared `--epoch` this keys a redundant pair on ## Closes diff --git a/quest/m1/lite07-finalize.md b/quest/m1/lite07-finalize.md index c783cabfb8..89c0e937a0 100644 --- a/quest/m1/lite07-finalize.md +++ b/quest/m1/lite07-finalize.md @@ -68,7 +68,6 @@ is published; older versions are unchanged. ## Required - [Cluster routing](/quest/m1/cluster-routing/README.md) - lite-07 carries the route layer: ROUTE per origin node and path-less ANNOUNCE, with the hop list gone -- [Remove `--hop`](/quest/m0/broadcast-epoch/hop-removal.md) - lite-07 loses NO_CAPACITY and stamping - [SUBSCRIBE_DROP](/quest/m1/subscribe-drop.md) - lite-07 restores SUBSCRIBE_DROP in place of `Stream Count` - [Subscribe ranges](/quest/m1/subscribe-ranges/README.md) - SUBSCRIBE carries ranges and an order and lite FETCH is gone, in Rust and JS - [Live media time](/quest/m1/subscribe-live-time.md) - SUBSCRIBE_OK carries the publisher's live media time diff --git a/rs/moq-cli/src/announced.rs b/rs/moq-cli/src/announced.rs index eafe3508e9..4c289d82a8 100644 --- a/rs/moq-cli/src/announced.rs +++ b/rs/moq-cli/src/announced.rs @@ -176,7 +176,7 @@ mod tests { use tokio::io::{AsyncBufReadExt, BufReader, DuplexStream, Lines}; const TIMEOUT: Duration = Duration::from_secs(10); - const ENV: &[&str] = &["MOQ_CONNECT", "MOQ_HOP", "MOQ_BROADCAST"]; + const ENV: &[&str] = &["MOQ_CONNECT", "MOQ_EPOCH", "MOQ_BROADCAST"]; /// A running relay holding `demo/a`, `demo/b`, and `other/c` from one publisher. struct Fixture { diff --git a/rs/moq-cli/src/archive.rs b/rs/moq-cli/src/archive.rs index 6aa0653993..e31ae999f7 100644 --- a/rs/moq-cli/src/archive.rs +++ b/rs/moq-cli/src/archive.rs @@ -100,11 +100,16 @@ pub async fn export( } } -/// Republish the recording at `args.store` as the broadcast `name`. +/// Republish the recording at `args.store` as the broadcast `name`, announced under `epoch`. /// /// Each track's timeline replays as a live track and every other track's groups are served on /// request. -pub async fn import(origin: moq_net::origin::Producer, name: String, args: ImportArgs) -> anyhow::Result<()> { +pub async fn import( + origin: moq_net::origin::Producer, + name: String, + args: ImportArgs, + epoch: moq_net::Epoch, +) -> anyhow::Result<()> { let store = open(&args.store)?; // A second handle lists the recording's timelines while the reader owns the first. let listing = open(&args.store)?; @@ -119,7 +124,7 @@ pub async fn import(origin: moq_net::origin::Producer, name: String, args: Impor .with_context(|| format!("no readable recording at {}", args.store))?; let serve = reader.serve(); broadcast - .announce(moq_tokio::moq_net::origin::Route::default().with_epoch(moq_tokio::moq_net::Epoch::mint())) + .announce(moq_net::origin::Route::default().with_epoch(epoch)) .context("failed to announce broadcast")?; tracing::info!(%name, store = %args.store, "replaying"); @@ -400,7 +405,12 @@ mod tests { store: url, follow: None, }; - let serving = tokio::spawn(import(replay.clone(), "replay.hang".into(), args)); + let serving = tokio::spawn(import( + replay.clone(), + "replay.hang".into(), + args, + moq_net::Epoch::mint(), + )); let consumer = replay.consume().routed_broadcast("replay.hang").await.unwrap(); let audio = consumer.track("audio").unwrap(); diff --git a/rs/moq-cli/src/args.rs b/rs/moq-cli/src/args.rs index f010234f5a..b501879fac 100644 --- a/rs/moq-cli/src/args.rs +++ b/rs/moq-cli/src/args.rs @@ -284,6 +284,28 @@ impl Invocation { /// Called before anything binds a port or dials out, so a refused invocation has /// no side effects to unwind. pub fn validate(&self) -> anyhow::Result<()> { + if self.moq.epoch.is_some() { + let mut publishes = false; + for command in &self.stages { + match command { + Command::Import(import) => { + anyhow::ensure!( + import.source.takes_epoch(), + "--epoch names one publisher instance, but the RTMP, SRT, and WHIP ingests and `import ts --program all` announce their own per connection or program" + ); + publishes = true; + } + #[cfg(feature = "transcode")] + Command::Transcode(_) => publishes = true, + _ => {} + } + } + anyhow::ensure!( + publishes, + "--epoch names what this process publishes, but nothing here publishes" + ); + } + for command in &self.stages { let Command::Export(export) = command else { continue; @@ -377,21 +399,14 @@ pub struct MoqSide { #[usage(long = "name", hide = true)] name: Option, - /// Fix this process's Hop ID instead of minting a fresh random one. + /// Announce under this epoch (a UUIDv7) instead of minting a fresh one per run. /// - /// The Hop ID is the first hop of every announcement this process - /// publishes, and relays treat it as the broadcast's content identity: - /// redundant publishers of the same broadcast share an id so relays fail - /// over between them at a group boundary. Leave unset outside a redundant - /// (1+1) chain; the default fresh id per run is what makes a restarted - /// publisher look like new content instead of silently splicing. - #[usage(long, env = "MOQ_HOP", help_heading = "MoQ")] - pub hop: Option, - - /// The released spelling of [`Self::hop`], kept in the parser only so a - /// process that still passes it is told what to pass instead. - #[usage(name = "origin", long = "origin", env = "MOQ_ORIGIN", hide = true)] - pub origin: Option, + /// Relays resume a subscription only between routes with the same epoch, so + /// redundant publishers of one broadcast pass the same value and fail over + /// seamlessly. Leave it unset otherwise: a fresh epoch per run is what makes a + /// restarted publisher replace the old broadcast instead of resuming into it. + #[usage(long, env = "MOQ_EPOCH", help_heading = "MoQ")] + pub epoch: Option, /// MoQ client config (`--connect`, `--connect-bind`, `--connect-tls-*`, ...). #[usage(flatten)] @@ -432,9 +447,6 @@ impl MoqSide { found.extend(self.server.deprecated()); found.extend(self.cluster.deprecated()); found.extend(self.auth.deprecated()); - if self.origin.is_some() { - found.flag("--origin", Some("MOQ_ORIGIN"), "--hop / MOQ_HOP"); - } if self.name.is_some() { found.flag("--name", None, "--broadcast"); } @@ -442,18 +454,10 @@ impl MoqSide { } /// The cluster this process publishes and subscribes on. Built once; the - /// origin is its origin. `--hop` fills `--cluster-id` when the latter is - /// unset; they must agree when both are set. + /// origin is its origin, whose Hop ID is random per run unless `--cluster-id` + /// names the node. pub fn cluster(&self) -> anyhow::Result { - let mut config = self.cluster.clone(); - match (config.id, self.hop) { - (None, Some(hop)) => config.id = Some(hop), - (Some(id), Some(hop)) if id != hop => { - anyhow::bail!("--hop {hop} and --cluster-id {id} must agree") - } - _ => {} - } - moq_relay::cluster::Cluster::new(moq_relay::cluster::Options::new(config)) + moq_relay::cluster::Cluster::new(moq_relay::cluster::Options::new(self.cluster.clone())) } /// Whether `--cluster-lan` asked this process to mesh over the LAN. @@ -755,6 +759,17 @@ impl ImportSource { _ => return None, }) } + + /// Whether this source announces one publisher instance per run, which `--epoch` + /// can name. The ingest gateways and `ts --program all` announce their own. + pub fn takes_epoch(&self) -> bool { + match self { + Self::Ts(args) => args.program != Some(TsProgram::All), + Self::Rtmp(_) | Self::Srt(_) => false, + Self::Rtc(rtc) => rtc.listen.is_none(), + _ => true, + } + } } /// The MPEG-TS stdin container: which programs of a multiplex to publish. @@ -1229,31 +1244,31 @@ mod tests { } } - /// `moq-cli`'s own rename rides the same refusal as the flags it flattens from - /// `moq-tokio`, and lands in the same message. - /// - /// Both halves matter: `--origin` must not silently pin a Hop ID onto the field - /// `--hop` now owns, and the migration has to name the environment variable too, - /// since a deployment that sets `MOQ_ORIGIN` never typed the flag. + /// `--epoch` reaches the stages that announce once per run, and is refused where + /// it would be ignored: an ingest that announces per connection, or no publisher. #[test] - fn the_released_origin_spelling_is_refused_with_a_migration() { - let Err(err) = Invocation::try_parse_from([ - "moq", - "--origin", - "42", - "--connect", - "http://relay/anon", - "export", - "ts", - ]) else { - panic!("--origin must not start a run"); + fn epoch_applies_only_to_a_once_per_run_publisher() { + let _env = crate::test_env::EnvGuard::clear(&["MOQ_EPOCH"]); + let epoch = hang::moq_net::Epoch::mint().to_string(); + let parse = |stage: &[&str]| { + let argv = ["moq", "--connect", "http://relay/anon", "--epoch", &epoch].into_iter(); + Invocation::try_parse_from(argv.chain(stage.iter().copied())).expect("parse") }; - let reported = err.to_string(); - assert!( - reported.contains("--origin / MOQ_ORIGIN -> --hop / MOQ_HOP"), - "missing the migration from {reported}" - ); + let cli = parse(&["import", "fmp4"]); + cli.validate().expect("a stdin import takes an epoch"); + assert_eq!(cli.moq.epoch.map(|epoch| epoch.to_string()), Some(epoch.clone())); + + for refused in [ + &["import", "rtmp", "--listen", "[::]:1935"][..], + &["import", "ts", "--program", "all"], + &["export", "fmp4"], + ] { + assert!(parse(refused).validate().is_err(), "{refused:?} accepted --epoch"); + } + + let malformed = Invocation::try_parse_from(["moq", "--epoch", "42", "import", "fmp4"]); + assert!(malformed.is_err(), "--epoch accepted a value that is not a UUIDv7"); } /// A stage carries config of its own, and the check has to reach it. @@ -1857,7 +1872,7 @@ mod tests { &["--listen-tcp-bind", "127.0.0.1:0"], &["--cluster-node", "https://self.example"], &["--auth-public", "**"], - &["--hop", "1"], + &["--epoch", "0192a3b4-c5d6-7e8f-9a0b-1c2d3e4f5a6b"], ]; let dial: &[&[&str]] = &[ &["--connect", "https://relay.example"], diff --git a/rs/moq-cli/src/complete.rs b/rs/moq-cli/src/complete.rs index 20748d5f5a..f9d87f63e1 100644 --- a/rs/moq-cli/src/complete.rs +++ b/rs/moq-cli/src/complete.rs @@ -344,7 +344,7 @@ mod tests { // A stage offers its own flags, and none of the globals it would refuse. let staged = complete("moq --connect http://x/y import fmp4 -- export fmp4 --").await; assert!(!staged.is_empty(), "a later stage completed nothing"); - for global in ["--connect", "--hop", "--broadcast"] { + for global in ["--connect", "--epoch", "--broadcast"] { assert!( !staged.iter().any(|candidate| candidate == global), "{global} leaked into a stage that refuses it: {staged:?}" diff --git a/rs/moq-cli/src/fetch.rs b/rs/moq-cli/src/fetch.rs index 7dfeae88a0..bddf5b1f6c 100644 --- a/rs/moq-cli/src/fetch.rs +++ b/rs/moq-cli/src/fetch.rs @@ -327,7 +327,7 @@ mod tests { } } - const ENV: &[&str] = &["MOQ_CONNECT", "MOQ_HOP"]; + const ENV: &[&str] = &["MOQ_CONNECT", "MOQ_EPOCH"]; /// A known sequence prints its frames back to back, the same bytes as `/fetch`. #[tokio::test(start_paused = true)] diff --git a/rs/moq-cli/src/hls.rs b/rs/moq-cli/src/hls.rs index 9d754ef295..f902c56d44 100644 --- a/rs/moq-cli/src/hls.rs +++ b/rs/moq-cli/src/hls.rs @@ -43,8 +43,9 @@ pub struct ExportArgs { pub cors: crate::web::Cors, } -/// Pull a remote HLS/LL-HLS playlist (URL or file path) into the Origin under `target.name`. -pub async fn import(target: ImportTarget, playlist: String) -> anyhow::Result<()> { +/// Pull a remote HLS/LL-HLS playlist (URL or file path) into the Origin under `target.name`, +/// announced under `epoch`. +pub async fn import(target: ImportTarget, playlist: String, epoch: moq_net::Epoch) -> anyhow::Result<()> { let ImportTarget { origin, name, @@ -60,7 +61,7 @@ pub async fn import(target: ImportTarget, playlist: String) -> anyhow::Result<() .with_bandwidth(bandwidth); let catalog = moq_mux::catalog::Producer::new(&mut producer, config)?; producer - .announce(moq_tokio::moq_net::origin::Route::default().with_epoch(moq_tokio::moq_net::Epoch::mint())) + .announce(moq_net::origin::Route::default().with_epoch(epoch)) .context("failed to announce broadcast")?; let playlist = playlist_url(&playlist)?; diff --git a/rs/moq-cli/src/main.rs b/rs/moq-cli/src/main.rs index 4d4fddbeb9..898f43ac7e 100644 --- a/rs/moq-cli/src/main.rs +++ b/rs/moq-cli/src/main.rs @@ -526,6 +526,9 @@ async fn run_stages(moq: MoqSide, stages: Vec, net: Net) -> anyhow::Res let mut stdin = None; let mut stdout = None; + // One publisher instance per run, unless redundant publishers named a shared one. + let epoch = moq.epoch.clone().unwrap_or_else(moq_net::Epoch::mint); + for stage in stages { let name = stage.broadcast(&moq); match stage { @@ -533,7 +536,9 @@ async fn run_stages(moq: MoqSide, stages: Vec, net: Net) -> anyhow::Res if import.source.stdin_format().is_some() { claim("stdin", &mut stdin, &name)?; } - if let Some(publish) = spawn_import(&origin, import, name, bandwidth.clone(), &mut tasks)? { + if let Some(publish) = + spawn_import(&origin, import, name, epoch.clone(), bandwidth.clone(), &mut tasks)? + { locals.push(publish); } } @@ -608,12 +613,16 @@ fn display_name(name: &str) -> &str { /// Route one stage's source INTO the shared Origin, exposing it to the MoQ network. /// +/// The sources announced once per run take `epoch`; the ingest gateways and +/// `ts --program all` mint their own per connection or program. +/// /// Returns the pipeline that has to run on the caller's thread, for the sources /// that have one (the stdin containers and capture). fn spawn_import( origin: &moq_net::origin::Producer, import: Import, name: String, + epoch: moq_net::Epoch, bandwidth: moq_net::bandwidth::Allocator, tasks: &mut JoinSet>, ) -> anyhow::Result> { @@ -650,13 +659,13 @@ fn spawn_import( let broadcast = origin.create_broadcast(&name).context("failed to create broadcast")?; Publish::new(broadcast, &format, config)? }; - publish.announce()?; + publish.announce(epoch)?; local = Some(publish); } else { match import.source { ImportSource::Hls(hls) => { warn_if_missing_format(&name); - tasks.spawn(hls::import(target(name), hls.playlist)); + tasks.spawn(hls::import(target(name), hls.playlist, epoch)); } ImportSource::Rtmp(rtmp) => { if let Some(addr) = rtmp.listen { @@ -689,20 +698,20 @@ fn spawn_import( }, )); } else if let Some(url) = rtc.connect { - tasks.spawn(rtc::connect_import(target(name), url)); + tasks.spawn(rtc::connect_import(target(name), url, epoch)); } } ImportSource::Archive(args) => { // A replay serves the retention the recording was made with. anyhow::ensure!(max_age.is_none(), "`--max-age` does not apply to `import archive`"); - tasks.spawn(archive::import(origin.clone(), name, args)); + tasks.spawn(archive::import(origin.clone(), name, args, epoch)); } #[cfg(feature = "capture")] ImportSource::Capture(capture) => { warn_if_missing_format(&name); let broadcast = origin.create_broadcast(&name).context("failed to create broadcast")?; let publish = Publish::capture(broadcast, &capture, bandwidth.clone(), max_age)?; - publish.announce()?; + publish.announce(epoch)?; local = Some(publish); } _ => unreachable!("container formats are handled by stdin_format above"), diff --git a/rs/moq-cli/src/publish.rs b/rs/moq-cli/src/publish.rs index c18499aaff..9ecec60e88 100644 --- a/rs/moq-cli/src/publish.rs +++ b/rs/moq-cli/src/publish.rs @@ -394,13 +394,13 @@ impl Publish { }) } - /// Advertise the broadcast's path, now that the catalog tracks are in place. - pub fn announce(&self) -> anyhow::Result<()> { + /// Advertise the broadcast's path under `epoch`, now that the catalog tracks are in place. + pub fn announce(&self, epoch: moq_net::Epoch) -> anyhow::Result<()> { let Some(broadcast) = &self.broadcast else { return Ok(()); }; broadcast - .announce(moq_tokio::moq_net::origin::Route::default().with_epoch(moq_tokio::moq_net::Epoch::mint())) + .announce(moq_net::origin::Route::default().with_epoch(epoch)) .context("failed to announce broadcast") } diff --git a/rs/moq-cli/src/rtc.rs b/rs/moq-cli/src/rtc.rs index d9e55666e0..21621bc9bb 100644 --- a/rs/moq-cli/src/rtc.rs +++ b/rs/moq-cli/src/rtc.rs @@ -87,8 +87,9 @@ fn scope_producer(origin: &moq_net::origin::Producer, name: &str) -> anyhow::Res .with_context(|| format!("failed to scope origin to broadcast `{name}`")) } -/// WHEP client: pull a remote broadcast into the Origin under `target.name` (import). -pub async fn connect_import(target: ImportTarget, url: Url) -> anyhow::Result<()> { +/// WHEP client: pull a remote broadcast into the Origin under `target.name` (import), +/// announced under `epoch`. +pub async fn connect_import(target: ImportTarget, url: Url, epoch: moq_net::Epoch) -> anyhow::Result<()> { let name = &target.name; let producer = target .origin @@ -97,7 +98,7 @@ pub async fn connect_import(target: ImportTarget, url: Url) -> anyhow::Result<() // The WHEP pull fills the tracks as they arrive; announce up front so viewers // can discover the broadcast while it connects. producer - .announce(moq_tokio::moq_net::origin::Route::default().with_epoch(moq_tokio::moq_net::Epoch::mint())) + .announce(moq_net::origin::Route::default().with_epoch(epoch)) .context("failed to announce broadcast")?; tracing::info!(url = %RedactedUrl::new(&url), %name, "WHEP client pulling"); diff --git a/rs/moq-cli/src/transcode.rs b/rs/moq-cli/src/transcode.rs index dc25e4cabe..0235f219ff 100644 --- a/rs/moq-cli/src/transcode.rs +++ b/rs/moq-cli/src/transcode.rs @@ -166,7 +166,7 @@ pub async fn run(moq: MoqSide, args: Args, net: Net) -> anyhow::Result<()> { .create_broadcast(&output_path) .context("failed to create the derivative broadcast")?; output - .announce(moq_tokio::moq_net::origin::Route::default().with_epoch(moq_tokio::moq_net::Epoch::mint())) + .announce(moq_net::origin::Route::default().with_epoch(moq.epoch.clone().unwrap_or_else(moq_net::Epoch::mint))) .context("failed to announce the derivative broadcast")?; tracing::info!(source = %source_path, output = %output_path, "transcoding"); diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index f144d58419..852a3bd44d 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -565,10 +565,6 @@ pub(super) struct Subscriber { // that withheld an identity is not named on the wire. A server answers it per // accepted session; a client only when it knows the peer. session_origin: crate::Hop, - // A random Hop ID of this connection's own, written as the first hop of any path - // that arrives naming no publisher, so a publisher that reconnects reads downstream - // as a new one. Fresh per connection, unlike `session_origin`. - stamp: crate::Hop, // Our own Hop ID, which an advertisement must not already contain: one that does // looped back through us. self_origin: crate::Hop, @@ -657,7 +653,6 @@ where origin, control, session_origin: peer_hop.unwrap_or(crate::Hop::UNKNOWN), - stamp: crate::Hop::random(), self_origin, peer_setup, cost, @@ -715,11 +710,10 @@ where /// The route for an advertisement that carries no path of its own. /// - /// Base moq-transport has no hops on the wire, so the chain is this connection's - /// stamp, naming the unknown publisher for as long as the connection lasts, then - /// the anonymous 0 that keeps it ranked below identified routes. - /// The session's assigned identity stays on `via` for split-horizon; putting it in - /// the chain would publish a name for a peer that declined to give one. + /// Base moq-transport has no hops on the wire, so the chain is a single 0: the + /// anonymous mark, forwarded unchanged. The session's assigned identity stays + /// on `via` for split-horizon; putting it in the chain would publish a name for + /// a peer that declined to give one. /// /// The link is charged all the same. Such an advertisement carries no ROUTE_COST, /// which reads as 0, but the draft charges every advertisement for the direction it @@ -728,11 +722,13 @@ where /// /// It is charged only one hop, though the chain it stands for may be arbitrarily /// long: a peer that carries no hop ids hides its depth, so this route understates - /// its true length. Price such a link with [`crate::Client::with_cost`]. + /// its true length. An anonymous route already ranks below every identified one, + /// so that understatement cannot beat a real path. Price such a link with + /// [`crate::Client::with_cost`] among other anonymous routes. fn session_route(&self, peer: &cluster::Peer) -> crate::origin::Route { let mut hops = crate::Hops::new(); - hops.stamp(self.stamp) - .expect("an empty hop chain has room for the stamp and its 0"); + hops.push(crate::Hop::UNKNOWN) + .expect("an empty hop chain has room for one entry"); crate::origin::Route::default() .with_hops(hops) .with_via(self.via(peer)) @@ -745,8 +741,8 @@ where /// /// A negotiated peer supplies the path and cost, so the route is what the mesh /// actually knows: the full chain, and the accumulated cost plus this link's price. - /// A path starting with 0 gets this connection's stamp in front of it; the 0s stay. An advertisement whose path already contains our own Hop ID looped back, - /// and neither forwarding it nor subscribing through it is safe. + /// A received 0 stays 0. An advertisement whose path already contains our own Hop + /// ID looped back, and neither forwarding it nor subscribing through it is safe. fn route(&self, advert: Option<&cluster::Advert>, peer: &cluster::Peer) -> Option { let Some(advert) = advert else { return Some(Advertised { @@ -758,11 +754,11 @@ where return None; } - let mut route = advert - .route(cluster::link_cost(self.cost, peer)) - .with_via(self.via(peer)); - route.hops.stamp(self.stamp).ok()?; - Some(Advertised { route }) + Some(Advertised { + route: advert + .route(cluster::link_cost(self.cost, peer)) + .with_via(self.via(peer)), + }) } /// Bind the alias the publisher chose for this subscription. @@ -5257,9 +5253,9 @@ mod tests { assert!(!table.map.contains_key(&0), "the oldest tombstone is forgotten first"); } - /// moq-transport carries no hop ids, so a peer's broadcasts are named by the - /// connection's own random stamp. An identity assigned via `Client::with_peer_hop` - /// is stored as `via` for split-horizon and never written into the chain. + /// moq-transport carries no hop ids, so a peer's broadcasts are marked + /// anonymous (hop 0). An identity assigned via `Client::with_peer_hop` is + /// stored as `via` for split-horizon and never written into the chain. #[moq_net_sim::test] async fn assigned_peer_hop_attributes_announces() { let session = crate::lite::test_transport::SinkSession::new(Default::default()); @@ -5290,10 +5286,8 @@ mod tests { let mut announced = consumer.announced(); let route = announced.assert_next_active("room/host"); let hops: Vec<_> = route.hops.iter().copied().collect(); - assert_eq!(hops, vec![subscriber.stamp, crate::Hop::UNKNOWN]); - assert_ne!(subscriber.stamp, crate::Hop::UNKNOWN); - assert!(route.is_anonymous(), "the 0 after the stamp still ranks it as unknown"); - assert_ne!(subscriber.stamp, assigned, "the assigned identity stays off the chain"); + assert_eq!(hops, vec![crate::Hop::UNKNOWN]); + assert!(route.is_anonymous()); let mut hidden = consumer.excluding(assigned).announced(); hidden.assert_next_wait(); @@ -5356,11 +5350,11 @@ mod tests { assert_eq!(route.hops, upstream); } - /// Two sessions assigned the same identity still stamp their own first hop, since - /// a wire with no hop ids cannot say the content continued: the reconnect reads as a - /// new source, while split-horizon keeps filtering on the shared identity. + /// Two sessions assigned the same identity announce the same anonymous chain. + /// The cursor treats that as an identical re-announce, so a reconnect is + /// invisible and retracting the stale session leaves the fresh route standing. #[moq_net_sim::test] - async fn reconnecting_peer_is_a_new_first_hop() { + async fn reconnecting_peer_joins_the_front_it_replaces() { let peer = crate::Hop::new(777).unwrap(); let self_origin = crate::Hop::new(1).unwrap(); @@ -5392,18 +5386,20 @@ mod tests { }; let first = connect(); - let first_stamp = first.stamp; - let route = announced.assert_next_active("room/host"); - assert_eq!(route.hops.iter().next(), Some(&first_stamp)); + announced.assert_next_active("room/host"); - // The peer reconnects before the old session is retired, under its own stamp. - let second = connect(); - assert_ne!(second.stamp, first_stamp); - assert!(routed_now(&consumer, "room/host").is_some()); + // The peer reconnects before the old session is retired: an identical route + // from the fresh session joins without any consumer-visible churn. + let _second = connect(); + announced.assert_next_wait(); // Neither route is offered back to the peer they both came from. - consumer.excluding(peer).announced().assert_next_wait(); + consumer.clone().excluding(peer).announced().assert_next_wait(); + + // The stale session finally retracting leaves the fresh route standing. drop(first); + announced.assert_next_wait(); + assert!(routed_now(&consumer, "room/host").is_some()); } fn cluster_subscriber( diff --git a/rs/moq-net/src/lite/setup.rs b/rs/moq-net/src/lite/setup.rs index c907da029d..446b75b7ba 100644 --- a/rs/moq-net/src/lite/setup.rs +++ b/rs/moq-net/src/lite/setup.rs @@ -176,10 +176,10 @@ pub struct Setup { /// Directional: it prices the sender's own egress, so both ends declare their own /// and the two need not match. `None` means the default cost of 1. pub cost: Option, - /// This endpoint's Hop ID, the identity it stamps onto forwarded - /// announcements. The peer uses it to serve this endpoint's subscriptions from - /// a route that does not flow through it (the same split horizon the announce - /// filter applies). `None` when the endpoint has no meaningful identity (a + /// This endpoint's Hop ID, the identity forwarded announcements name it by. + /// The peer uses it to serve this endpoint's subscriptions from a route that + /// does not flow through it (the same split horizon the announce filter + /// applies). `None` when the endpoint has no meaningful identity (a /// leaf that never forwards); a wire value of 0 decodes as `None`. pub hop: Option, } diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 389aa9ebcf..0dd7552ad7 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -68,11 +68,6 @@ pub(super) struct Subscriber { // another session's, while a client only assigns one it knows out of band. The // assigned id stays local and is never written into a hop chain. session_origin: crate::Hop, - // A random Hop ID of this connection's own, written as the first hop of any chain - // that names no publisher (Lite01/02, a Lite03 placeholder, or a peer reporting 0), - // so a publisher that reconnects reads downstream as a new one. Fresh per - // connection, unlike `session_origin`. - stamp: crate::Hop, subscribes: Lock>, /// Why this session ended, once it has. A track still waiting on TRACK_INFO is /// not in [`Self::subscribes`], so dropping its request reads this instead of @@ -116,7 +111,6 @@ impl Subscriber { recv_bandwidth: config.recv_bandwidth, self_origin, session_origin: config.peer_hop.unwrap_or(crate::Hop::UNKNOWN), - stamp: crate::Hop::random(), subscribes: Default::default(), ended: Default::default(), next_id: Default::default(), @@ -323,12 +317,13 @@ impl Subscriber { return Ok(false); } - // Lite01/02 send no list at all, and Lite03 carries only UNKNOWN placeholders, - // so the publisher may be unnamed: this connection's stamp goes in front to name - // it, and the 0s stay so the route still ranks as anonymous. - if hops.stamp(self.stamp).is_err() { - tracing::debug!(route = %self.log_path(&path), "dropping announce; no room to stamp the chain"); - return Ok(false); + // Lite03 carries its hop count as UNKNOWN placeholders rather than real + // ids; they stay 0 and count as anonymous. Lite01/02 send no list at all. + // Either way the chain must have at least the anonymous mark so a + // downstream hop can see that this path passed through an unidentified hop. + if hops.is_empty() { + hops.push(crate::Hop::UNKNOWN) + .expect("an empty hop chain always has room for one entry, and repeats nothing"); } tracing::debug!(route = %self.log_path(&path), hops = hops.len(), "announce"); @@ -429,11 +424,9 @@ impl Subscriber { return Ok(false); } - // Named by this connection's stamp when the chain names no publisher, exactly as - // the announce was, so a restart of the same unnamed publisher stays in place. - if hops.stamp(self.stamp).is_err() { - announced.declined(path); - return Ok(false); + if hops.is_empty() { + hops.push(crate::Hop::UNKNOWN) + .expect("an empty hop chain always has room for one entry, and repeats nothing"); } tracing::debug!(route = %self.log_path(&path), hops = hops.len(), "restart"); @@ -2865,9 +2858,8 @@ mod tests { ); } - /// A peer that declares no identity has its publisher named by the connection's - /// stamp. The assigned identity stays on `via` for split-horizon and is never - /// written into the chain. + /// A peer that declares no identity is marked anonymous (hop 0). The assigned + /// identity stays on `via` for split-horizon and is never written into the chain. #[moq_net_sim::test] async fn assigned_peer_hop_attributes_announces() { let session = SinkSession::new(Default::default()); @@ -2903,21 +2895,19 @@ mod tests { .unwrap(); assert!(accepted); - // The route is announced synchronously: the stamp in the chain, assigned id local. + // The route is announced synchronously: hop 0 on the wire, assigned id local. let mut cursor = consumer.announced(); let route = cursor.assert_next_active("room/host"); let hops: Vec<_> = route.hops.iter().copied().collect(); - assert_eq!(hops, vec![subscriber.stamp, crate::Hop::UNKNOWN]); - assert_ne!(subscriber.stamp, assigned); + assert_eq!(hops, vec![crate::Hop::UNKNOWN]); assert!(route.is_anonymous()); let mut hidden = consumer.excluding(assigned).announced(); hidden.assert_next_wait(); } - /// Lite03 hop-count placeholders name nobody, so the connection's stamp goes in front; - /// the placeholders stay 0 and count as anonymous. None is rewritten with the - /// assigned identity. + /// Lite03 hop-count placeholders stay 0 and count as anonymous; they are not + /// rewritten with the assigned identity. #[moq_net_sim::test] async fn lite03_placeholders_stay_anonymous() { let assigned = crate::Hop::new(777).unwrap(); @@ -2954,22 +2944,19 @@ mod tests { let mut cursor = consumer.announced(); let route = cursor.assert_next_active("room/host"); let hops: Vec<_> = route.hops.iter().copied().collect(); - assert_eq!(hops, vec![subscriber.stamp, crate::Hop::UNKNOWN, crate::Hop::UNKNOWN]); + assert_eq!(hops, vec![crate::Hop::UNKNOWN, crate::Hop::UNKNOWN]); assert!(route.is_anonymous()); } - /// A publisher that names nobody is stamped with a random id fresh per connection, - /// never 0, ahead of the 0 that keeps it ranked as unknown: a restart on the same connection keeps the stamp and updates in place, - /// while the same publisher reconnecting reads as a new first hop. + /// A publisher that names nobody keeps the anonymous mark, and nothing in front of it: + /// identity is the epoch's job, so the relay never names the publisher itself. A + /// reprice on the same connection keeps it and updates in place. #[moq_net_sim::test] - async fn an_unnamed_publisher_is_stamped_per_connection() { - let (mut first, consumer) = restart_subscriber(SinkSession::new(Default::default())); - let (second, _) = restart_subscriber(SinkSession::new(Default::default())); - assert_ne!(first.stamp, crate::Hop::UNKNOWN); - assert_ne!(first.stamp, second.stamp, "each connection stamps its own id"); + async fn an_unnamed_publisher_stays_anonymous() { + let (mut subscriber, consumer) = restart_subscriber(SinkSession::new(Default::default())); let mut announced = Announced::default(); - first + subscriber .start_announce( Path::new("room/host").to_owned(), None, @@ -2984,10 +2971,10 @@ mod tests { let mut cursor = consumer.announced(); let route = cursor.assert_next_active("room/host"); let hops: Vec<_> = route.hops.iter().copied().collect(); - assert_eq!(hops, vec![first.stamp, crate::Hop::UNKNOWN]); + assert_eq!(hops, vec![crate::Hop::UNKNOWN]); - // A reprice from the same unnamed publisher keeps the stamp, so it stays in place. - first + // A reprice from the same unnamed publisher stays anonymous and in place. + subscriber .restart_announce( Path::new("room/host").to_owned(), crate::Hops::new(), @@ -2999,7 +2986,7 @@ mod tests { .unwrap(); let route = cursor.assert_next_active("room/host"); let hops: Vec<_> = route.hops.iter().copied().collect(); - assert_eq!(hops, vec![first.stamp, crate::Hop::UNKNOWN]); + assert_eq!(hops, vec![crate::Hop::UNKNOWN]); } fn restart_subscriber(session: SinkSession) -> (Subscriber, crate::origin::Consumer) { diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index e1cf83b59e..fdcf181174 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -263,34 +263,6 @@ impl Hops { Ok(()) } - /// Name an unknown original publisher: prefix `stamp`, the receiving session's own - /// per-connection id, to a chain that starts with 0, and turn an empty chain into - /// `[stamp, 0]`. - /// - /// A publisher that reconnects then reads downstream as a new first hop, which is - /// all anyone can say about it, while the 0 after the stamp keeps the route ranked - /// as anonymous. Fails with [`InvalidHop::TooMany`] if the chain is full, and with - /// [`InvalidHop::Duplicate`] if `stamp` already appears in it. - pub(crate) fn stamp(&mut self, stamp: Hop) -> Result<(), InvalidHop> { - match self.0.first() { - None => { - self.push(stamp)?; - self.push(Hop::UNKNOWN) - } - Some(first) if *first == Hop::UNKNOWN => { - if self.0.len() >= MAX_HOPS { - return Err(InvalidHop::TooMany); - } - if self.0.contains(&stamp) { - return Err(InvalidHop::Duplicate); - } - self.0.insert(0, stamp); - Ok(()) - } - Some(_) => Ok(()), - } - } - /// Returns true if any entry matches `hop`. pub fn contains(&self, hop: &Hop) -> bool { self.0.contains(hop) @@ -482,8 +454,7 @@ pub struct Route { /// The chain of origins the route has traversed, oldest first. Each relay /// appends its own [`crate::Hop`] when forwarding; used for loop detection /// and as the selection tie-break. A 0 entry is the anonymous mark and - /// travels unchanged; a session receiving one as the first entry puts its own - /// per-connection stamp in front of it; see [`Self::is_anonymous`]. + /// travels unchanged; see [`Self::is_anonymous`]. pub hops: Hops, /// What pulling content via this route costs, accumulated per link: lower wins @@ -5506,46 +5477,6 @@ mod tests { assert_eq!(route.cost, Cost::new(5)); } - /// A stamped route names its session but keeps the 0 behind the stamp, so it still - /// loses to a fully identified route of the same length, however cheap it is. - #[moq_net_sim::test] - async fn a_stamped_route_ranks_below_an_identified_one() { - let producer = origin(1).produce(); - let mut announced = producer.consume().announced(); - - let mut stamped = Hops::new(); - stamped.stamp(origin(5)).unwrap(); - assert_eq!(stamped.as_slice(), &[origin(5), Hop::UNKNOWN]); - - let _legacy = producer - .announce("room", Route::default().with_hops(stamped).with_cost(0)) - .unwrap(); - let route = announced.assert_next_active("room"); - assert!(route.is_anonymous()); - - let _identified = producer - .announce("room", Route::default().with_hops(hops(&[10, 11])).with_cost(5)) - .unwrap(); - let route = announced.assert_next_active("room"); - assert_eq!(route.hops.as_slice(), hops(&[10, 11]).as_slice()); - } - - #[test] - fn stamping_keeps_the_leading_zero_and_names_nothing_else() { - let mut chain = hops(&[0, 7]); - chain.stamp(origin(5)).unwrap(); - assert_eq!(chain.as_slice(), hops(&[5, 0, 7]).as_slice()); - - // Already named: untouched. - let mut named = hops(&[7, 0]); - named.stamp(origin(5)).unwrap(); - assert_eq!(named.as_slice(), hops(&[7, 0]).as_slice()); - - // A full chain has no room for the stamp. - let mut full = Hops::try_from(vec![Hop::UNKNOWN; MAX_HOPS]).unwrap(); - assert_eq!(full.stamp(origin(5)), Err(InvalidHop::TooMany)); - } - #[test] fn anonymous_routes_order_by_cost() { let producer = origin(1).produce(); diff --git a/rs/moq-net/src/server.rs b/rs/moq-net/src/server.rs index a751afec72..4f615ed1bf 100644 --- a/rs/moq-net/src/server.rs +++ b/rs/moq-net/src/server.rs @@ -440,8 +440,8 @@ pub struct Handshake { role: Option, origin: Option, token: Option, - /// The identity this session's routes are stamped with when the peer declares none - /// on the wire. Fresh per request unless the caller overrides it + /// The identity this session's routes are attributed to for split horizon when the + /// peer declares none on the wire; it never enters a hop chain. Fresh per request unless the caller overrides it /// ([`Handshake::with_peer_hop`]). assigned_hop: crate::Hop, // Taken by `ok`/`close`; `Drop` rejects the handshake if neither ran. diff --git a/rs/moq-net/tests/legacy_reconnect.rs b/rs/moq-net/tests/legacy_reconnect.rs deleted file mode 100644 index adaf2147b9..0000000000 --- a/rs/moq-net/tests/legacy_reconnect.rs +++ /dev/null @@ -1,80 +0,0 @@ -//! A publisher on a wire with no hop ids, reconnecting through a relay, is a new first -//! hop downstream. -//! -//! moq-transport without the Cluster extension names no publisher, so the relay it -//! connects to stamps each connection with a random Hop ID of its own. A reconnect is a -//! new connection and so a new first hop. - -mod support; - -use std::time::Duration; - -use moq_net::{Hop, Version, announce, origin}; -use support::harness::{MockConnectOptions, MockPair, connect_mock, peer}; - -const TEST_TIMEOUT: Duration = Duration::from_secs(10); -const PATH: &str = "room/cam"; - -fn produce_origin(hop: u64) -> origin::Producer { - let (producer, driver) = origin::Producer::new(origin::Config::new(Hop::new(hop).unwrap())); - support::harness::spawn(driver); - producer -} - -/// Connect `publisher` to `relay` over draft-14, which carries no hop ids. -async fn connect_legacy(publisher: &origin::Producer, relay: &origin::Producer) -> MockPair { - let mut options = MockConnectOptions::new("moq-transport-14".parse::().unwrap()); - options.client_publish = Some(publisher.consume()); - options.server_subscribe = Some(relay.clone()); - connect_mock(options).await -} - -/// The first hop of the next active update for [`PATH`]. A withdrawal in between is -/// skipped: whether the relay saw the new connection before the old one dropped is a -/// race, and either way the path comes back. -async fn next_first_hop(announced: &mut announce::Consumer) -> Hop { - loop { - let update = match announced.next().await.expect("announce cursor ended") { - announce::Event::Start(update) | announce::Event::Update(update) => update, - announce::Event::End(_) => continue, - }; - if update.prefix.as_str() != PATH { - continue; - } - return *update.route.hops.iter().next().expect("a route names its first hop"); - } -} - -#[moq_net_sim::test] -async fn a_legacy_reconnect_is_a_new_first_hop_downstream() { - for mesh in ["moq-lite-06", "moq-transport-17"] { - moq_net_sim::timeout(TEST_TIMEOUT, async { - let publisher = produce_origin(9); - let relay = produce_origin(1); - let downstream = produce_origin(2); - let _mesh = peer(mesh.parse::().unwrap(), &relay, &downstream).await; - let mut announced = downstream.consume().announced(); - - let broadcast = publisher.create_broadcast(PATH).unwrap(); - broadcast.announce(Default::default()).unwrap(); - - let first = connect_legacy(&publisher, &relay).await; - let before = next_first_hop(&mut announced).await; - assert_ne!(before, Hop::UNKNOWN, "{mesh}: the relay stamps an unnamed publisher"); - assert_ne!(before, Hop::new(9).unwrap(), "{mesh}: draft-14 never carried the id"); - - // The publisher reconnects: the same content, but nothing on the wire says so. - let second = connect_legacy(&publisher, &relay).await; - drop(first); - let mut after = next_first_hop(&mut announced).await; - while after == before { - after = next_first_hop(&mut announced).await; - } - assert_ne!(after, Hop::UNKNOWN, "{mesh}: the relay stamps the new connection too"); - - drop(second); - }) - .await - .expect("test timed out"); - } -} diff --git a/rs/moq-relay/tests/cluster_unknown.rs b/rs/moq-relay/tests/cluster_unknown.rs index 72492b264c..bf57b60b63 100644 --- a/rs/moq-relay/tests/cluster_unknown.rs +++ b/rs/moq-relay/tests/cluster_unknown.rs @@ -1,6 +1,6 @@ //! Regression for an external publisher whose protocol does not declare a Hop ID. -//! The relay stamps that publisher with a random Hop ID of the connection's own; -//! reflected cluster paths must not replace it while propagating around a redundant mesh. +//! The relay records that publisher as `Hop::UNKNOWN`; reflected cluster paths +//! must not replace it while propagating around a redundant mesh. use std::time::Duration; @@ -105,8 +105,8 @@ async fn publish_version(port: u16, version: &str) -> Publisher { } async fn publish_unknown(port: u16) -> Publisher { - // Draft-14 has no Cluster extension, so the accepting relay must name this - // external publisher with a stamp of its own. + // Draft-14 has no Cluster extension, so the accepting relay must represent + // this external publisher with the wire-defined UNKNOWN Hop ID. publish_version(port, "moq-transport-14").await } @@ -226,7 +226,7 @@ async fn assert_unknown_publisher_stays_announced(cluster_version: Option