feat(net)!: routes scoped to interested streams, presented incrementally; hop chains private

dart/moq/lib/src/aliases.dart

@@ -67,7 +67,7 @@ typedef AnnounceConfig = MoqAnnounceConfig;
 /// Follows a specific broadcast path across the publisher instances that serve it.
 typedef Following = MoqFollowing;
 
-/// The read-only facts on a received route: hops, a static cost, and the stream's instance.
+/// The read-only facts on a received route: a static cost and the stream's instance.
 typedef RouteInfo = MoqRouteInfo;
 
 /// A route over a prefix: its origin-relative path, required wildcard captures (empty without wildcards), and route metadata.

dart/moq_ffi/lib/src/moq.dart

@@ -3007,16 +3007,9 @@ class FfiConverterMoqOriginConfig {
 }
 
 class MoqRouteInfo {
-  final List<int> hops;
   final int cost;
-  final bool anonymous;
   final String instance;
-  MoqRouteInfo({
-    required this.hops,
-    required this.cost,
-    required this.anonymous,
-    required this.instance,
-  });
+  MoqRouteInfo({required this.cost, required this.instance});
 }
 
 class FfiConverterMoqRouteInfo {
@@ -3026,42 +3019,25 @@ class FfiConverterMoqRouteInfo {
 
   static LiftRetVal<MoqRouteInfo> read(Uint8List buf) {
     int new_offset = buf.offsetInBytes;
-    final hops_lifted = FfiConverterSequenceUInt64.read(
-      Uint8List.view(buf.buffer, new_offset),
-    );
-    final hops = hops_lifted.value;
-    new_offset += hops_lifted.bytesRead;
     final cost_lifted = FfiConverterUInt64.read(
       Uint8List.view(buf.buffer, new_offset),
     );
     final cost = cost_lifted.value;
     new_offset += cost_lifted.bytesRead;
-    final anonymous_lifted = FfiConverterBool.read(
-      Uint8List.view(buf.buffer, new_offset),
-    );
-    final anonymous = anonymous_lifted.value;
-    new_offset += anonymous_lifted.bytesRead;
     final instance_lifted = FfiConverterString.read(
       Uint8List.view(buf.buffer, new_offset),
     );
     final instance = instance_lifted.value;
     new_offset += instance_lifted.bytesRead;
     return LiftRetVal(
-      MoqRouteInfo(
-        hops: hops,
-        cost: cost,
-        anonymous: anonymous,
-        instance: instance,
-      ),
+      MoqRouteInfo(cost: cost, instance: instance),
       new_offset - buf.offsetInBytes,
     );
   }
 
   static RustBuffer lower(MoqRouteInfo value) {
     final total_length =
-        FfiConverterSequenceUInt64.allocationSize(value.hops) +
         FfiConverterUInt64.allocationSize(value.cost) +
-        FfiConverterBool.allocationSize(value.anonymous) +
         FfiConverterString.allocationSize(value.instance) +
         0;
     final buf = Uint8List(total_length);
@@ -3071,18 +3047,10 @@ class FfiConverterMoqRouteInfo {
 
   static int write(MoqRouteInfo value, Uint8List buf) {
     int new_offset = buf.offsetInBytes;
-    new_offset += FfiConverterSequenceUInt64.write(
-      value.hops,
-      Uint8List.view(buf.buffer, new_offset),
-    );
     new_offset += FfiConverterUInt64.write(
       value.cost,
       Uint8List.view(buf.buffer, new_offset),
     );
-    new_offset += FfiConverterBool.write(
-      value.anonymous,
-      Uint8List.view(buf.buffer, new_offset),
-    );
     new_offset += FfiConverterString.write(
       value.instance,
       Uint8List.view(buf.buffer, new_offset),
@@ -3091,9 +3059,7 @@ class FfiConverterMoqRouteInfo {
   }
 
   static int allocationSize(MoqRouteInfo value) {
-    return FfiConverterSequenceUInt64.allocationSize(value.hops) +
-        FfiConverterUInt64.allocationSize(value.cost) +
-        FfiConverterBool.allocationSize(value.anonymous) +
+    return FfiConverterUInt64.allocationSize(value.cost) +
         FfiConverterString.allocationSize(value.instance) +
         0;
   }
@@ -19724,49 +19690,6 @@ class FfiConverterSequenceString {
   }
 }
 
-class FfiConverterSequenceUInt64 {
-  static List<int> lift(RustBuffer buf) {
-    return FfiConverterSequenceUInt64.read(buf.asUint8List()).value;
-  }
-
-  static LiftRetVal<List<int>> read(Uint8List buf) {
-    List<int> res = [];
-    final length = buf.buffer.asByteData(buf.offsetInBytes).getInt32(0);
-    int offset = buf.offsetInBytes + 4;
-    for (var i = 0; i < length; i++) {
-      final ret = FfiConverterUInt64.read(Uint8List.view(buf.buffer, offset));
-      offset += ret.bytesRead;
-      res.add(ret.value);
-    }
-    return LiftRetVal(res, offset - buf.offsetInBytes);
-  }
-
-  static int write(List<int> value, Uint8List buf) {
-    buf.buffer.asByteData(buf.offsetInBytes).setInt32(0, value.length);
-    int offset = buf.offsetInBytes + 4;
-    for (var i = 0; i < value.length; i++) {
-      offset += FfiConverterUInt64.write(
-        value[i],
-        Uint8List.view(buf.buffer, offset),
-      );
-    }
-    return offset - buf.offsetInBytes;
-  }
-
-  static int allocationSize(List<int> value) {
-    return value
-            .map((l) => FfiConverterUInt64.allocationSize(l))
-            .fold(0, (a, b) => a + b) +
-        4;
-  }
-
-  static RustBuffer lower(List<int> value) {
-    final buf = Uint8List(allocationSize(value));
-    write(value, buf);
-    return toRustBuffer(buf);
-  }
-}
-
 class FfiConverterUInt16 {
   static int lift(int value) => value;
   static LiftRetVal<int> read(Uint8List buf) {

doc/bin/relay/cluster.md

@@ -204,9 +204,13 @@ connect = [
 ```
 
 Prices aren't static. A publisher can re-price a live announcement, which is
-how a standby transcoder pool seeds a high cost and drops it once it's working,
-and a relay receiving a GOAWAY re-prices every route learned from that peer to
-the maximum so new subscriptions go elsewhere while existing ones finish. A
+how a standby transcoder pool seeds a high cost and drops it once it's working.
+A GOAWAY is not a price: a relay receiving one ranks every route learned from
+that peer below every other eligible route, whatever the costs say. New
+subscriptions go elsewhere when another route serves them, and an existing one
+moves to another route of the same stream if there is one, or else finishes on
+the draining peer. The drain is that relay's
+own state, so it still advertises the routes onward at their real cost. A
 relay keeps its session to a peer draining to restart until that peer closes
 it, and redials only then, so it reaches the replacement rather than the
 draining process.

doc/concept/identity.md

@@ -71,7 +71,10 @@ A route resolving a path is chosen in this order:
 2. Within one instance, only its highest epoch is eligible, even over a
    cheaper route. Readers continue across an epoch change without a seam,
    because the sequence space continues.
-3. Among different instances: the preferred link, then an identified chain
+3. Among eligible routes: a route from a peer that has not sent GOAWAY
+   (a draining peer's routes rank below every other, whatever their preference
+   or cost, and serve only as a last resort), then the preferred link, then an
+   identified chain
    over an anonymous one, then the lowest cost, then a broadcast published on
    this origin, then the shortest hop chain, then a hash of the full requested
    path with the route's origin node, which moq-lite 07 carries, or else with

doc/lib/rs/moq-net.md

@@ -16,7 +16,7 @@ moq-transport at setup and presents one API either way. Media is a layer above
 What you use it for, beyond what the concept page already describes:
 
 - **An origin outlives the session.** `origin.route(path, options)` returns a `Route` that owns the broadcast to fill (`route.broadcast()`) and controls its announcement: `announce()`, `withdraw()`, `set_cost()`, and dropping the last clone retracts it. `origin.claim(prefix, options)` returns a `Claim` that serves requests beneath a prefix instead: `claim.requested()` yields each one, answered with `request.serve(held)`, `request.route(options)` (the path's own exact route) or `reject`. A `Claim` has the same announcement controls and no broadcast. A standalone broadcast cannot announce; publish one only through a claim's `serve`. A reconnect announces existing routes again, and closing the session does not delete them.
-- **Routes report what the mesh filled in.** An announcement's `route` gives its `instance()`, `cost()`, the hop ids it crossed (`hops()`, oldest first, 0 for an anonymous hop) and `is_anonymous()`; an application sets only `RouteOptions`. An origin's own hop id is `origin::Config::hop`, random unless a relay names one.
+- **Routes report what the mesh filled in.** An announcement's `route` gives its `instance()` and `cost()`; an application sets only `RouteOptions`. The hop chain a relay uses for loop detection stays internal. An origin's own hop id is `origin::Config::hop`, random unless a relay names one.
 - **Requests can pin an instance.** `consumer.request(path, Some(instance))` resolves only through a route announcing that [publisher instance](/concept/moq-lite#publisher-instances); `None` takes whichever route wins. The resolved consumer names its instance in `info().instance`, pinned or not. `origin.route(path, options)` serves a random instance minted when it is created and kept across `withdraw()` and `announce()`. Unless one is named, a claim moves to a fresh instance each time a job holding its current one ends, and a job takes the claim's. Set one with `RouteOptions::with_instance(instance, epoch)` only for a publisher that continues its group sequences across restarts and raises the epoch each time. `route.instance()` names it.
 - **Rank several CDNs.** Dial each with its own subscribe handle, `origin.clone().with_preference(origin::Preference::new(rank))` (higher wins; 0 is neutral and a negative rank demotes), all on one origin. Requests use the preferred link that has a route and fail over to the next; see [multiple CDNs](/concept/moq-lite#multiple-cdns).
 - **Request one instance, or follow them all.** `consumer.request(path, None)` resolves the instance serving a path now, and fails `Unroutable` only once discovery has synced with no route for it. `consumer.follow(path, None)` hands out each publisher instance serving a path from `available()`: the first once announced, then each one that takes over, `None` once the path goes offline, and whatever serves it next, like JS `follow(path)`. `consumer.announced()` lists paths only as present or absent. A broadcast already resolved stays on the old instance until dropped or its route goes. See [identity and recovery](/concept/identity) to choose between following and holding one instance.

drafts/draft-lcurley-moq-lite.md

@@ -497,7 +497,7 @@ It MUST NOT select an infeasible route, and a distance only ever falls within a
 A node MAY mark a session upstream.
 A route is down-only when it arrived on an upstream session or with its Down Only bit set; a node sends a down-only route with the bit set, and never on an upstream session.
 
-A node sends its selected route to every origin node it knows on each Announce Stream it answers, and MAY omit routes with no claim matching the stream's request.
+A node sends its selected route to an origin node on an Announce Stream it answers only while the route has a claim the stream's request presents (see [Announce](#announce)): it sends ROUTE_START before the first such claim, and once the last one is withdrawn, the batch carrying that withdrawal, closed by its floor (which certifies the withdrawal where the node can certify the request), then ROUTE_END. It starts that route on the stream again only once the subscriber echoed the ROUTE_END. A subscriber therefore holds no route for an origin node its request does not meet.
 A node that does not forward, such as a client, sends only its own route, and holds no feasibility distance for any other.
 An origin advances its Seqno whenever its own set of sessions changes.
 It never advertises a Seqno or Revision of 2^64-1: a change that would reach it instead rotates its Node ID, ending the old route everywhere and announcing its claims again under the new one.
@@ -2194,6 +2194,7 @@ The `Message Length` describes the payload size on the wire.
 - Added ROUTE_REQUEST (0xA), sent by the subscriber on the Announce Stream to ask for a route with a newer seqno when every route it has is infeasible.
 - Added ANNOUNCE_FLOOR (0x9), closing each batch of a route's claims with the origin Revision at which the stream's claims across the requested prefix are certified plus one, or 0 to certify nothing, so a stale copy never revives a claim whose withdrawal a newer one delivered. A floor certifies only the claims its own request shows.
 - A route presents only its most specific claim covering the requested prefix, moved to the next covering claim when it is withdrawn rather than ended.
+- A route goes on a stream only while it has a claim the request presents, and ends after the batch withdrawing its last claim. It starts on that stream again only once the subscriber echoed the ROUTE_END.
 - Added `Node ID` to TRACK, TRACKS, SUBSCRIBE, FETCH, and DATAGRAMS: the origin node the first hop selected, followed by every later hop.
 - Replaced Warm and Cold Route Cost with the origin's Announce Cost, unchanged in transit. ANNOUNCE_UPDATE carries only the Announce Cost.
 - The Cost parameter must be at least 1.

go/wrapper/moq_test.go

@@ -792,9 +792,6 @@ func TestLocalPublishConsumeAudio(t *testing.T) {
 	if ann.Prefix != "live" {
 		t.Fatalf("prefix = %q, want %q", ann.Prefix, "live")
 	}
-	if len(ann.Route.Hops) != 0 {
-		t.Fatalf("route hops = %v, want empty for local origin", ann.Route.Hops)
-	}
 
 	bc, err := consumer.Request(ctx, ann.Prefix, nil)
 	if err != nil {

go/wrapper/types.go

@@ -39,7 +39,7 @@ type (
 	// route of one instance together, so restarting the sequences under an old instance splices
 	// unrelated groups into their tracks.
 	RouteOptions = ffi.MoqRouteOptions
-	// RouteInfo is the read-only facts on a received route: the Instance of the stream it serves (routes with one instance are one continuous stream), the hop chain it took to reach an origin, and its static production and link cost (lower wins).
+	// RouteInfo is the read-only facts on a received route: the Instance of the stream it serves (routes with one instance are one continuous stream) and its static production and link cost (lower wins).
 	RouteInfo = ffi.MoqRouteInfo
 	// Announce is a route over a prefix: the origin-relative Prefix, what each filter
 	// wildcard matched (Captures), and the RouteInfo serving it. It carries no broadcast;

js/net/api/index.api.md

@@ -513,9 +513,7 @@ export namespace Origin {
 		withdraw(): void;
 	}
 	export interface RouteInfo {
-		readonly anonymous: boolean;
 		readonly cost: Origin.Cost;
-		readonly hops: readonly bigint[];
 		readonly instance?: Instance.Valid;
 	}
 	export interface RouteOptions {

js/net/src/announced.test.ts

@@ -43,7 +43,7 @@ test("a listing reports only a path becoming present or absent", async () => {
 	expect((await listing.next())?.kind).toBe("absent");
 });
 
-test("a listing shows a route's facts, never its epoch or credential", async () => {
+test("a listing shows a route's facts, never its epoch, hops, credential, or drain state", async () => {
 	const producer = new Announce.Producer();
 	const listing = Announce.list(producer.consume());
 
@@ -54,6 +54,7 @@ test("a listing shows a route's facts, never its epoch or credential", async ()
 		hops: [HopSchema.parse(7n), UNKNOWN_HOP],
 		cost: 5n,
 		auth: {} as Credential,
+		draining: true,
 	});
 	producer.append({ prefix: p("a"), captures: undefined, kind: "start", route });
 
@@ -62,9 +63,9 @@ test("a listing shows a route's facts, never its epoch or credential", async ()
 		kind: "present",
 		prefix: p("a"),
 		captures: undefined,
-		route: { instance, hops: [7n, 0n], cost: 5n, anonymous: true },
+		route: { instance, cost: 5n },
 	});
-	expect(Object.keys(event?.route ?? {}).sort()).toEqual(["anonymous", "cost", "hops", "instance"]);
+	expect(Object.keys(event?.route ?? {}).sort()).toEqual(["cost", "instance"]);
 });
 
 test("the consumer is an async iterable of the same events", async () => {

js/net/src/goaway-requests.test.ts

@@ -21,9 +21,6 @@ import { wireOf } from "./wire.ts";
 
 const url = new URL("https://relay.example/room");
 
-// Same ceiling Rust stamps as Cost::DRAIN, and the lite cost varint saturates at.
-const DRAIN = 2n ** 62n - 1n;
-
 async function settle() {
 	await yieldNow();
 }
@@ -164,9 +161,11 @@ async function handover(kind: "lite" | "ietf"): Promise<void> {
 		else await sendLiteGoaway(pair.server);
 		await client.draining;
 
+		// The route ranks last, at its real cost.
 		const update = await announced.next();
 		expect(update?.kind).toBe("update");
-		expect(update?.route.cost).toBe(DRAIN);
+		expect(update?.route.draining).toBe(true);
+		expect(update?.route.cost).toBe(start?.route.cost);
 		// Still the only route, so the subscription already open keeps its front.
 		expect(request.active.peek()).toBe(front);
 
@@ -185,10 +184,10 @@ async function handover(kind: "lite" | "ietf"): Promise<void> {
 		expect(await group.readString()).toBe("one");
 		group.close();
 
-		// A new announce-interest opens too, and sees the route at the drain cost.
+		// A new announce-interest opens too, and sees the route draining.
 		const interest = client.routes();
 		const drained = await interest.next();
-		expect(drained?.route.cost).toBe(DRAIN);
+		expect(drained?.route.draining).toBe(true);
 		interest.close();
 
 		expect(pair.client.sendStreams.bidi.length).toBeGreaterThan(opened);