diff --git a/Cargo.lock b/Cargo.lock index ee815167cb4..629f6f1f0b8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5054,6 +5054,7 @@ dependencies = [ "loom", "moq-clock", "moq-json", + "moq-net", "moq-pattern 0.1.2", "num_enum", "rand 0.10.3", diff --git a/js/net/MIGRATE.md b/js/net/MIGRATE.md index 77990b90505..2bdc084a66e 100644 --- a/js/net/MIGRATE.md +++ b/js/net/MIGRATE.md @@ -52,7 +52,7 @@ Only `publish` announces it: `prepared.broadcast.announce()` throws. `Prepared.reject(err)` is replaced by idempotent `close()` and `[Symbol.dispose]()`; either refuses the unpublished answer as `Unroutable` and closes its broadcast. Closing that broadcast also refuses the request. `using` handles normal scope exit and exceptions; after `publish()`, disposal leaves the returned producer live. Use `request.reject(err)` to refuse before preparing. A retained, unclosed `Prepared` keeps its request pending; garbage collection provides no prompt cleanup guarantee. Construct a `Prepared` only with `request.prepare()`. Its constructor and the `Request.publisher`, `publishPrepared`, and `rejectPrepared` completion hooks are no longer public. -`Dynamic.update` and `Request.accept` are no longer part of the public API, and `accept` throws on a request from `dynamic`. +`Dynamic.update` is no longer part of the public API, and `Request` has no `accept` (#6496): only the built-in sessions answer with a broadcast they already hold. Close the claim and claim again to change it, and close an answer's producer to end that output. ### `largest()` reports within a reader's `setGroups` bounds diff --git a/js/net/bench/broadcasts.ts b/js/net/bench/broadcasts.ts index e491fe7e48b..e500038b24d 100644 --- a/js/net/bench/broadcasts.ts +++ b/js/net/bench/broadcasts.ts @@ -6,6 +6,7 @@ import { Producer as BroadcastProducer } from "../src/broadcast.ts"; import { BroadcastCache } from "../src/consume.ts"; import * as Epoch from "../src/epoch.ts"; import { HopSchema, Route } from "../src/hop.ts"; +import { hooks } from "../src/internal.ts"; import { AnnounceRequest } from "../src/lite/announce.ts"; import { Publisher } from "../src/lite/publisher.ts"; import { Subscriber } from "../src/lite/subscriber.ts"; @@ -252,7 +253,10 @@ for (const pathCount of routeCounts) { Array.from({ length: readerCount }, () => origin.consume().request(Path.from(`room/${index}`))), ).flat(); const broadcasts = Array.from({ length: pathCount }, () => new BroadcastProducer()); - for (const broadcast of broadcasts) (await pending.next()).value?.accept(broadcast); + for (const broadcast of broadcasts) { + const request = (await pending.next()).value; + if (request) hooks.acceptRequest(request, broadcast); + } await Promise.resolve(); const before = requests.map((request) => request.active.peek()); const start = performance.now(); diff --git a/js/net/bench/requests.ts b/js/net/bench/requests.ts index 32aa2666a3f..4afeb144fa2 100644 --- a/js/net/bench/requests.ts +++ b/js/net/bench/requests.ts @@ -10,6 +10,7 @@ import { SubscribeNamespaceEntryDone, } from "../src/ietf/subscribe_namespace"; import { ALPN, Version } from "../src/ietf/version"; +import { hooks } from "../src/internal"; import { createMockTransportPair } from "../src/mock"; import { Producer } from "../src/origin"; import * as Path from "../src/path"; @@ -111,7 +112,8 @@ for (const failovers of [0, 1, 16, 128]) const held = origin.request(path); const source = new Broadcast(); sources.push(source); - (await ar.next()).value?.accept(source); + const answer = (await ar.next()).value; + if (answer) hooks.acceptRequest(answer, source); await settle(); entries.push({ primary, ar, br, held }); } @@ -148,7 +150,7 @@ for (const failovers of [0, 1, 16, 128]) for (let i = 0; i < entries.length; i++) { const source = new Broadcast(); sources.push(source); - pending[i].accept(source); + hooks.acceptRequest(pending[i], source); await settle(); if (!readsFrom(entries[i].held.active.peek(), source)) throw new Error("lost failover"); } diff --git a/js/net/src/connection/forward.ts b/js/net/src/connection/forward.ts index 684482c7977..7a3a91ab78b 100644 --- a/js/net/src/connection/forward.ts +++ b/js/net/src/connection/forward.ts @@ -8,7 +8,7 @@ import { Milli, Nano } from "@moq/clock"; import { type Dispose, race } from "@moq/signals"; import type * as Announce from "../announced.ts"; import { type Route, routesEqual } from "../hop.ts"; -import { coveringPrefixes } from "../internal.ts"; +import { coveringPrefixes, hooks } from "../internal.ts"; import type { Dynamic, Producer as OriginProducer, RequestSlot } from "../origin.ts"; import * as Path from "../path.ts"; import { after } from "../util/timer.ts"; @@ -276,7 +276,7 @@ async function drive(handle: Dynamic, conn: Established): Promise { const session = wireOf(conn); try { for await (const request of handle.requested()) { - request.accept(session.consume(request.path)); + hooks.acceptRequest(request, session.consume(request.path)); } } catch { handle.close(); diff --git a/js/net/src/ietf/publisher.test.ts b/js/net/src/ietf/publisher.test.ts index 0a7b0de99cd..d125d687742 100644 --- a/js/net/src/ietf/publisher.test.ts +++ b/js/net/src/ietf/publisher.test.ts @@ -7,6 +7,7 @@ import * as Epoch from "../epoch.ts"; import { error, ProtocolViolation, SessionCode, SessionError, StreamCode, unauthorized } from "../error.ts"; import { Producer as GroupProducer, MAX_GROUP_FRAMES } from "../group.ts"; import { type Hop, HopSchema } from "../hop.ts"; +import { hooks } from "../internal.ts"; import { createMockTransportPair } from "../mock.ts"; import { Producer as OriginProducer } from "../origin.ts"; import * as Path from "../path.ts"; @@ -435,7 +436,7 @@ test("a cancelled TRACK_STATUS releases its demand on every draft", async () => expect(early.written).toEqual([]); // Withdrawn, so a late answer is closed rather than served. const answer = new BroadcastProducer().consume(); - pending?.accept(answer); + hooks.acceptRequest(pending, answer); expect(answer.closed.peek()).toBeNull(); // The broadcast resolves, but its track is never answered. @@ -445,7 +446,7 @@ test("a cancelled TRACK_STATUS releases its demand on every draft", async () => if (!request) throw new Error("no request for room/late"); const served = new BroadcastProducer(); const trackRequest = wireOf(served).requested(); - request.accept(served); + hooks.acceptRequest(request, served); expect((await trackRequest)?.name).toBe("video"); expect(served.demand().used.peek()).toBe(true); late.cancel(); @@ -485,7 +486,7 @@ test("a SUBSCRIBE withdrawn during setup releases its demand on every draft", as expect(early.written).toEqual([]); // Withdrawn, so a late answer is closed rather than served. const answer = new BroadcastProducer().consume(); - pending?.accept(answer); + if (pending) hooks.acceptRequest(pending, answer); expect(answer.closed.peek()).toBeNull(); // The broadcast resolves, but its track is never answered. @@ -495,7 +496,7 @@ test("a SUBSCRIBE withdrawn during setup releases its demand on every draft", as if (!request) throw new Error("no request for room/late"); const served = new BroadcastProducer(); const trackRequest = wireOf(served).requested(); - request.accept(served); + hooks.acceptRequest(request, served); expect((await trackRequest)?.name).toBe("video"); expect(served.demand().used.peek()).toBe(true); late.withdraw(); diff --git a/js/net/src/integration.test.ts b/js/net/src/integration.test.ts index c1465e8cfac..16a5681dd6d 100644 --- a/js/net/src/integration.test.ts +++ b/js/net/src/integration.test.ts @@ -19,6 +19,7 @@ import { Parameters } from "./ietf/parameters.ts"; import { Publisher as IetfPublisher } from "./ietf/publisher.ts"; import { RequestOk } from "./ietf/request.ts"; import { SubscribeTracks } from "./ietf/subscribe_tracks.ts"; +import { hooks } from "./internal.ts"; import * as Lite from "./lite/index.ts"; import { Publisher as LitePublisher } from "./lite/publisher.ts"; import { createMockTransportPair, textFrame } from "./mock.ts"; @@ -2739,7 +2740,7 @@ test("a handle serves a request under live/** over the wire", async () => { for await (const request of handle.requested()) { const produced = new BroadcastProducer(); produced.createTrack("chat", { timescale: Timescale.MILLI }).writeFrame(textFrame("hello")); - request.accept(produced); + hooks.acceptRequest(request, produced); } })(); diff --git a/js/net/src/internal.ts b/js/net/src/internal.ts index 311219b09bf..f9c7c5f09d2 100644 --- a/js/net/src/internal.ts +++ b/js/net/src/internal.ts @@ -14,6 +14,7 @@ import type * as Epoch from "./epoch.ts"; import { toStreamCode } from "./error.ts"; import type { Frame, Consumer as GroupConsumer, Producer as GroupProducer } from "./group.ts"; import { isAnonymous, type Route } from "./hop.ts"; +import type { Request as OriginRequest } from "./origin.ts"; import * as Path from "./path.ts"; import type { Timestamp } from "./time.ts"; import type { Demand, Groups, Info, Position, Producer, Request, Subscriber, Subscription } from "./track.ts"; @@ -553,6 +554,11 @@ export const hooks: { stampEpoch: (target: BroadcastConsumer, epoch: Epoch.Valid | undefined) => void; /** Mark a broadcast handle served through a route claiming a shorter prefix: it lists nothing. */ stampUnlisted: (target: BroadcastConsumer) => void; + /** + * Resolve a session-forwarded request with its peer's routed broadcast. An application + * claim answers through `prepare` and `publish` instead. Assigned by `origin.ts`. + */ + acceptRequest: (request: OriginRequest, source: BroadcastProducer | BroadcastConsumer) => void; } = { endTrack: () => { throw new Error("track.ts not loaded"); @@ -734,6 +740,9 @@ export const hooks: { stampUnlisted: () => { throw new Error("broadcast.ts not loaded"); }, + acceptRequest: () => { + throw new Error("origin.ts not loaded"); + }, }; /** diff --git a/js/net/src/lite/publisher.test.ts b/js/net/src/lite/publisher.test.ts index b797842820c..d827efb477b 100644 --- a/js/net/src/lite/publisher.test.ts +++ b/js/net/src/lite/publisher.test.ts @@ -2503,7 +2503,7 @@ test("lite draft-06: a grant that shrinks while a TRACK resolves refuses it", as const request = await requests.next(); if (request.done) throw new Error("the handler never saw the request"); grant.set(grantOf("other")); - request.value.accept(produced); + hooks.acceptRequest(request.value, produced); await serving; expect(unauthorizedResets(resets)).toContain("unauthorized: live/cam"); @@ -2559,7 +2559,7 @@ test("lite draft-06: a grant that shrinks while TRACK_INFO is blocked resets it" const serving = publisher.runTrackInfo(new TrackMessage(Path.from("live/cam"), "video"), stream); const request = await requests.next(); if (request.done) throw new Error("the handler never saw the request"); - request.value.accept(produced); + hooks.acceptRequest(request.value, produced); await writing; grant.set(grantOf("other")); @@ -2608,7 +2608,7 @@ test("lite draft-06: a grant that shrinks while the broadcast resolves resets th if (request.done) throw new Error("the handler never saw the request"); grant.set(grantOf("other")); await Promise.resolve(); - request.value.accept(produced); + hooks.acceptRequest(request.value, produced); await serving; expect(unauthorizedResets(resets)).toContain("unauthorized: live/cam"); diff --git a/js/net/src/lite/tracks.test.ts b/js/net/src/lite/tracks.test.ts index 362180d38cb..5c7a9185085 100644 --- a/js/net/src/lite/tracks.test.ts +++ b/js/net/src/lite/tracks.test.ts @@ -7,6 +7,7 @@ import { accept, connect } from "../connection/index.ts"; import * as Epoch from "../epoch.ts"; import { fromTransport, NotFound, SessionCode, StreamCode, StreamError } from "../error.ts"; import { randomHop } from "../hop.ts"; +import { hooks } from "../internal.ts"; import { createMockTransportPair } from "../mock.ts"; import { Producer as OriginProducer } from "../origin.ts"; import * as Path from "../path.ts"; @@ -412,7 +413,7 @@ test("a path served through a covering prefix route is NOT_FOUND", async () => { for await (const request of dynamic.requested()) { const served = new BroadcastProducer(); served.createTrack("video"); - request.accept(served); + hooks.acceptRequest(request, served); } })(); const { list } = publisher(origin); @@ -429,7 +430,7 @@ test("a local read through a covering prefix route is NOT_FOUND, and through its for await (const request of dynamic.requested()) { const served = new BroadcastProducer(); served.createTrack("video"); - request.accept(served); + hooks.acceptRequest(request, served); } })(); } @@ -529,7 +530,7 @@ test("an epoch-pinned front served through a covering prefix route is NOT_FOUND" for await (const request of dynamic.requested()) { const served = new BroadcastProducer(); served.createTrack("video"); - request.accept(served); + hooks.acceptRequest(request, served); } })(); const request = origin.consume().request(Path.from("room/alice")); diff --git a/js/net/src/origin-demand.test.ts b/js/net/src/origin-demand.test.ts index 887af3e9288..89d2f5e207c 100644 --- a/js/net/src/origin-demand.test.ts +++ b/js/net/src/origin-demand.test.ts @@ -6,6 +6,7 @@ import { NativeSession } from "./ietf/adapter.ts"; import { Publisher as IetfPublisher } from "./ietf/publisher.ts"; import { Subscribe as IetfSubscribe } from "./ietf/subscribe.ts"; import { Version as IetfVersion } from "./ietf/version.ts"; +import { hooks } from "./internal.ts"; import { Fetch } from "./lite/fetch.ts"; import { Publisher as LitePublisher } from "./lite/publisher.ts"; import { Subscribe as LiteSubscribe } from "./lite/subscribe.ts"; @@ -128,11 +129,11 @@ test("last cancellation finishes the internal wait and discards a late consumer" const keeper = source.consume(); const late = source.consume(); const close = spyOn(late, "close"); - old.accept(late); + hooks.acceptRequest(old, late); expect(close).toHaveBeenCalledTimes(1); close.mockRestore(); expect(source.closed.peek()).toBeUndefined(); - fresh.accept(source); + hooks.acceptRequest(fresh, source); await settle(); expect(next.active.peek()?.closed).toBe(source.closed); next.close(); @@ -171,7 +172,7 @@ test.each([true, false])( await settle(); const late = new Broadcast(); const front = late.consume(); - old.accept(front); + hooks.acceptRequest(old, front); expect(late.closed.peek()).toBeNull(); const next = origin.request(path); const fresh = (await requests.next()).value; @@ -210,7 +211,7 @@ test.each([ try { hold.abort(); await first; - request.accept(source); + hooks.acceptRequest(request, source); await settle(); expect(other === "peer" ? (await second)?.closed : local?.active.peek()?.closed).toBe(source.closed); } finally { @@ -257,7 +258,7 @@ test.each(["accept", "reject", "cancel"] as const)( const request = (await requests.next()).value; if (!request) throw new Error("expected a pending request"); try { - if (action === "accept") request.accept(source); + if (action === "accept") hooks.acceptRequest(request, source); if (action === "reject") request.reject(new Error("denied")); if (action === "cancel") hold.abort(); const result = await wait; @@ -265,9 +266,9 @@ test.each(["accept", "reject", "cancel"] as const)( hold.abort(); if (action === "accept") { expect(result?.closed).toBe(source.closed); - request.accept(source); + hooks.acceptRequest(request, source); request.reject(new Error("late")); - } else request.accept(source); + } else hooks.acceptRequest(request, source); expect(source.closed.peek()).toBeUndefined(); } finally { remove.mockRestore(); @@ -295,9 +296,9 @@ test("a pending local request changing routes withdraws the old route's work", a const late = source.consume(); const close = spyOn(late, "close"); try { - old.accept(late); + hooks.acceptRequest(old, late); expect(close).toHaveBeenCalledTimes(1); - fresh.accept(source); + hooks.acceptRequest(fresh, source); await settle(); expect(local.active.peek()?.closed).toBe(source.closed); } finally { @@ -331,7 +332,7 @@ test("cancelling an old epoch waiter does not decrement a new epoch's demand", a const source = new Broadcast(); try { old.reject(new Error("obsolete")); - fresh.accept(source); + hooks.acceptRequest(fresh, source); expect((await after)?.closed).toBe(source.closed); } finally { origin.close(); @@ -352,7 +353,7 @@ test.each(["answer", "release"] as const)( const br = b.requested(); const held = origin.request(path); const source = new Broadcast(); - (await ar.next()).value?.accept(source); + hooks.acceptRequest((await ar.next()).value, source); await settle(); // Another instance wins new requests. The old handle stays on its own epoch. const c = origin.dynamicRoute(prefix, { epoch: Epoch.mint(), cost: 0n }); @@ -377,7 +378,7 @@ test.each(["answer", "release"] as const)( const front = late.consume(); const close = spyOn(front, "close"); try { - pending.accept(front); + hooks.acceptRequest(pending, front); await settle(); if (action === "release") expect(close).toHaveBeenCalledTimes(1); else { @@ -412,7 +413,7 @@ test.each([0, 1])( const held = origin.request(path); const source = new Broadcast(); sources.push(source); - (await ar.next()).value?.accept(source); + hooks.acceptRequest((await ar.next()).value, source); await settle(); entries.push({ primary, ar, br, held }); } @@ -436,9 +437,9 @@ test.each([0, 1])( // The slot is still detached until its microtask runs, but no longer owns demand. entries[released].held.close(); hold.abort(); - pending[released].accept(front); + hooks.acceptRequest(pending[released], front); expect(close).toHaveBeenCalledTimes(1); - pending[1 - released].accept(replacement); + hooks.acceptRequest(pending[1 - released], replacement); await wait; expect(readsFrom(entries[1 - released].held.active.peek(), replacement)).toBe(true); } finally { @@ -511,7 +512,7 @@ test.each(["accept", "reject"] as const)( const denied = new Error("denied"); try { hold.abort(cancelled); - if (answer === "accept") request.accept(source); + if (answer === "accept") hooks.acceptRequest(request, source); else request.reject(denied); expect(await first).toBe(cancelled); const result = await second; @@ -541,7 +542,7 @@ test.each([true, false])("last waiter cancellation and accept in one turn (cance const close = spyOn(front, "close"); try { if (cancelFirst) hold.abort(cancelled); - request.accept(front); + hooks.acceptRequest(request, front); if (!cancelFirst) hold.abort(cancelled); expect(await wait).toBe(cancelled); expect(close).toHaveBeenCalledTimes(cancelFirst ? 1 : 0); @@ -551,7 +552,7 @@ test.each([true, false])("last waiter cancellation and accept in one turn (cance const fresh = (await requests.next()).value; expect(fresh).not.toBe(request); if (!fresh) throw new Error("expected fresh request"); - fresh.accept(source); + hooks.acceptRequest(fresh, source); } const answer = await next; expect(answer?.closed).toBe(source.closed); @@ -586,7 +587,7 @@ test.each([true, false])("local demand survives same-turn cancellation only when const replacement = rejoin ? origin.request(path) : undefined; try { await settle(); - request.accept(front); + hooks.acceptRequest(request, front); expect(close).toHaveBeenCalledTimes(rejoin ? 0 : 1); if (replacement) { await settle(); @@ -596,7 +597,7 @@ test.each([true, false])("local demand survives same-turn cancellation only when const next = (await requests.next()).value; expect(next).not.toBe(request); if (!next) throw new Error("expected fresh request"); - next.accept(source); + hooks.acceptRequest(next, source); await settle(); expect(fresh.active.peek()?.closed).toBe(source.closed); fresh.close(); diff --git a/js/net/src/origin-request.test.ts b/js/net/src/origin-request.test.ts index 50cdbfb3bc5..47a9e024dfa 100644 --- a/js/net/src/origin-request.test.ts +++ b/js/net/src/origin-request.test.ts @@ -3,7 +3,8 @@ import { Producer as BroadcastProducer } from "./broadcast"; import * as Epoch from "./epoch"; import { StreamCode, StreamError } from "./error"; import { Route } from "./hop"; -import { Producer } from "./origin"; +import { hooks } from "./internal.ts"; +import { Producer, type Request } from "./origin"; import * as Path from "./path"; import { wireOf } from "./wire"; @@ -143,7 +144,7 @@ test("announced requests follow the serving prefix, ignoring descendants and re- const keeper = source.consume(); const prefix = origin.dynamicRoute(Path.from("pool")); const serving = (async () => { - for await (const request of prefix.requested()) request.accept(source.consume()); + for await (const request of prefix.requested()) hooks.acceptRequest(request, source.consume()); })(); const request = origin.request(Path.from("pool/job"), { announced: true }); await settle(); @@ -196,7 +197,7 @@ test("an announced request follows an epochless route replaced by an identical o const serve = (source: BroadcastProducer) => { const route = origin.dynamicRoute(path); void (async () => { - for await (const request of route.requested()) request.accept(source.consume()); + for await (const request of route.requested()) hooks.acceptRequest(request, source.consume()); })(); return route; }; @@ -227,7 +228,7 @@ test("an origin closing settles an announced request a served route resolved", a const source = new BroadcastProducer(); const route = origin.dynamicRoute(path); void (async () => { - for await (const request of route.requested()) request.accept(source.consume()); + for await (const request of route.requested()) hooks.acceptRequest(request, source.consume()); })(); const request = origin.request(path, { announced: true }); await settle(); @@ -277,7 +278,7 @@ for (const received of [false, true]) { const source = new BroadcastProducer(); const route = received ? wireOf(origin).receive(path, Route.default) : origin.dynamicRoute(path); void (async () => { - for await (const request of route.requested()) request.accept(source.consume()); + for await (const request of route.requested()) hooks.acceptRequest(request, source.consume()); })(); const request = origin.request(path); @@ -342,14 +343,15 @@ test("a prepared answer cannot announce itself before publish", async () => { origin.close(); }); -test("an application claim refuses an arbitrary broadcast", async () => { +test("an application claim cannot accept an arbitrary broadcast", async () => { const origin = new Producer(); const claim = origin.dynamic(Path.from("room")); const requests = claim.requested(); const request = origin.request(Path.from("room/alice")); - const pending = (await requests.next()).value; + const pending: Request | undefined = (await requests.next()).value; if (!pending) throw new Error("expected a request"); - expect(() => pending.accept(new BroadcastProducer())).toThrow("prepared and published"); + // @ts-expect-error Only sessions accept a request; an application prepares and publishes. + expect(pending.accept).toBeUndefined(); request.close(); claim.close(); origin.close(); diff --git a/js/net/src/origin.test.ts b/js/net/src/origin.test.ts index 1433b81dcde..6717f94274b 100644 --- a/js/net/src/origin.test.ts +++ b/js/net/src/origin.test.ts @@ -6,7 +6,7 @@ import * as Epoch from "./epoch.ts"; import { StreamCode, StreamError } from "./error.ts"; import { Producer as GroupProducer } from "./group.ts"; import { HopSchema, Route } from "./hop.ts"; -import { spreadHash } from "./internal.ts"; +import { hooks, spreadHash } from "./internal.ts"; import type { Consumer, Table } from "./origin.ts"; import { Producer } from "./origin.ts"; import * as Path from "./path.ts"; @@ -31,7 +31,7 @@ function serve(origin: Producer, prefix: Path.Valid, consume: () => BroadcastCon void (async () => { try { for await (const request of handle.requested()) { - request.accept(consume()); + hooks.acceptRequest(request, consume()); } } catch { handle.close(); @@ -1092,7 +1092,7 @@ test("a handle serves a request under live", async () => { const produced = new BroadcastProducer(); produced.createTrack("video", { timescale: Timescale.MILLI }); - req?.accept(produced); + hooks.acceptRequest(req, produced); await settle(); expect(request.active.peek()).toBeDefined(); @@ -1117,11 +1117,11 @@ test("an originated route refuses a broadcast a session delivered", async () => // What lite and IETF do when a session hands back a consumed broadcast. const foreign = new BroadcastProducer().consume(); overrideBroadcastWire(foreign, {}); - expect(() => req.accept(foreign)).toThrow("origin cannot serve a broadcast it did not produce"); + expect(() => hooks.acceptRequest(req, foreign)).toThrow("origin cannot serve a broadcast it did not produce"); const produced = new BroadcastProducer(); produced.createTrack("video"); - req.accept(produced.consume()); + hooks.acceptRequest(req, produced.consume()); expect(await pending).toBeDefined(); handle.close(); @@ -1172,7 +1172,7 @@ test("an accepted dynamic broadcast is retired when it closes", async () => { const first = await it.next(); const produced = new BroadcastProducer(); produced.createTrack("video", { timescale: Timescale.MILLI }); - first.value?.accept(produced); + hooks.acceptRequest(first.value, produced); await settle(); expect(request.active.peek()?.track("video").subscribe()).toBeDefined(); @@ -1184,7 +1184,7 @@ test("an accepted dynamic broadcast is retired when it closes", async () => { expect(second.value?.path).toBe(path); const replacement = new BroadcastProducer(); replacement.createTrack("video", { timescale: Timescale.MILLI }); - second.value?.accept(replacement); + hooks.acceptRequest(second.value, replacement); await settle(); expect(request.active.peek()?.track("video").subscribe()).toBeDefined(); @@ -1222,7 +1222,7 @@ test("a peer is offered and served the cheapest originated route", async () => { const pending = wireOf(consumer).demand(Path.from("live/cam")); const { value: req } = await cheap.requested().next(); const upstream = new BroadcastProducer(); - req?.accept(upstream.consume()); + hooks.acceptRequest(req, upstream.consume()); expect(await pending).toBeDefined(); // An exact-path local broadcast competes with them on cost too. @@ -1281,7 +1281,7 @@ test("a request refusal does not keep demand from asking again", async () => { const { value: second } = await it.next(); expect(second?.path).toBe(Path.from("live/cam")); const produced = new BroadcastProducer(); - second?.accept(produced); + hooks.acceptRequest(second, produced); await expect(pending).resolves.toBeDefined(); handle.close(); @@ -1305,7 +1305,7 @@ test("advancing requested without settling rejects the previous request", async await expect(firstDemand).rejects.toMatchObject({ code: StreamCode.Unroutable }); const produced = new BroadcastProducer(); - second.value?.accept(produced); + hooks.acceptRequest(second.value, produced); await expect(secondDemand).resolves.toBeDefined(); handle.close(); @@ -1371,7 +1371,7 @@ test("an outranked route keeps serving until its replacement answers", async () // Once it answers, the request's tracks read it, under the same broadcast. const replacement = new BroadcastProducer(); const { value: req } = await asked; - req?.accept(replacement.consume()); + hooks.acceptRequest(req, replacement.consume()); await settle(); expect(request.active.peek()).toBe(first); expect(readsFrom(first, replacement)).toBe(true); @@ -1705,7 +1705,7 @@ test("a refusal from a superseded route does not end the request", async () => { expect(request.closed.peek()).toBeUndefined(); const produced = new BroadcastProducer(); - req?.accept(produced); + hooks.acceptRequest(req, produced); await settle(); expect(request.active.peek()).toBeDefined(); @@ -1757,7 +1757,7 @@ test("close rejects queued requests as unroutable", async () => { expect((await handle.requested().next()).done).toBe(true); expect(request.active.peek()).toBeUndefined(); - req?.accept(new BroadcastProducer()); + hooks.acceptRequest(req, new BroadcastProducer()); await settle(); expect(request.active.peek()).toBeUndefined(); @@ -1798,7 +1798,7 @@ test("an equal-cost pool spreads its paths the same way on every node", async () void (async () => { for await (const request of handle.requested()) { served.set(request.path, hop); - request.accept(new BroadcastProducer()); + hooks.acceptRequest(request, new BroadcastProducer()); } })(); return handle; @@ -1837,7 +1837,7 @@ test("an equal-cost replacement at its own path wins whatever its hop", async () void (async () => { for await (const request of handle.requested()) { served.push(hop); - request.accept(new BroadcastProducer()); + hooks.acceptRequest(request, new BroadcastProducer()); } })(); return handle; @@ -2056,7 +2056,7 @@ test("a scoped dynamic only serves allowed paths and presents relative requests" await new Promise((resolve) => setImmediate(resolve)); expect(pending.value?.path).toBe(Path.from("room/alice/chat")); const broadcast = new BroadcastProducer(); - pending.value?.accept(broadcast); + hooks.acceptRequest(pending.value, broadcast); await new Promise((resolve) => setImmediate(resolve)); expect(request.active.peek()).toBeDefined(); dynamic.updateRoute({ cost: 2n }); @@ -2190,7 +2190,7 @@ test("broadcast handles carry the path they were created or requested at", async const request = origin.request(Path.from("live/bob")); const pending = await dynamic.requested().next(); const served = new BroadcastProducer(); - pending.value?.accept(served); + hooks.acceptRequest(pending.value, served); expect(request.active.peek()?.path).toBe(Path.from("live/bob")); clone?.close(); @@ -2237,12 +2237,12 @@ test("demand held across an epoch change asks again, unless it named the old epo // Another instance now serves the route: the old handler's answer is for the old one. handle.updateRoute({ ...handle.route, epoch: NEWER }); - req?.accept(new BroadcastProducer().consume()); + hooks.acceptRequest(req, new BroadcastProducer().consume()); await expect(pinned).rejects.toThrow("unroutable"); const next = new BroadcastProducer(); const track = next.createTrack("video"); - (await it.next()).value?.accept(next); + hooks.acceptRequest((await it.next()).value, next); const subscription = (await unpinned)?.track("video").subscribe(); track.writeGroup(new GroupProducer(7)); expect((await subscription?.recvGroup())?.sequence).toBe(7); @@ -2286,11 +2286,11 @@ test("demand held across an epoch round trip still asks again", async () => { // Released by the first change; the epoch coming back does not undo that. handle.updateRoute({ ...handle.route, epoch: NEWER }); handle.updateRoute({ ...handle.route, epoch: EPOCH }); - req?.accept(new BroadcastProducer().consume()); + hooks.acceptRequest(req, new BroadcastProducer().consume()); const next = new BroadcastProducer(); const track = next.createTrack("video"); - (await it.next()).value?.accept(next); + hooks.acceptRequest((await it.next()).value, next); const subscription = (await demanded)?.track("video").subscribe(); track.writeGroup(new GroupProducer(7)); expect((await subscription?.recvGroup())?.sequence).toBe(7); @@ -2376,13 +2376,13 @@ test("a request held across an epoch change asks again and never takes the old h const old = (await requests.next()).value; dynamic.updateRoute({ ...dynamic.route, epoch: NEWER }); const current = origin.consume().request(path, { epoch: NEWER }); - old?.accept(new BroadcastProducer().consume()); + hooks.acceptRequest(old, new BroadcastProducer().consume()); await settle(); expect(pinned.active.peek()).toBeUndefined(); expect(unpinned.active.peek()).toBeUndefined(); expect(current.active.peek()).toBeUndefined(); const next = (await requests.next()).value; - next?.accept(new BroadcastProducer().consume()); + hooks.acceptRequest(next, new BroadcastProducer().consume()); await settle(); expect(unpinned.active.peek()?.epoch).toBe(NEWER); expect(current.active.peek()?.epoch).toBe(NEWER); @@ -2411,7 +2411,7 @@ test.each([ const held = origin.consume().request(path); const old = new BroadcastProducer(); const oldTrack = old.createTrack("video"); - (await requests.next()).value?.accept(old); + hooks.acceptRequest((await requests.next()).value, old); await settle(); const resolved = held.active.peek(); expect(resolved?.epoch).toBe(from); @@ -2448,7 +2448,7 @@ test.each([ const demanded = wireOf(origin.consume()).demand(path); const next = new BroadcastProducer(); const nextTrack = next.createTrack("video"); - (await requests.next()).value?.accept(next); + hooks.acceptRequest((await requests.next()).value, next); await settle(); expect(fresh.active.peek()?.epoch).toBe(to); const subscription = fresh.active.peek()?.track("video").subscribe(); @@ -2489,7 +2489,7 @@ test("an epoch update does not retain an answer that already closed", async () = const requests = dynamic.requested(); const held = origin.consume().request(path); const old = new BroadcastProducer(); - (await requests.next()).value?.accept(old); + hooks.acceptRequest((await requests.next()).value, old); old.close(); // The server's closure callback has not removed the answer yet. dynamic.updateRoute({ epoch: NEWER }); @@ -2498,7 +2498,7 @@ test("an epoch update does not retain an answer that already closed", async () = expect(held.active.peek()).toBeUndefined(); const fresh = origin.consume().request(path); const next = new BroadcastProducer(); - (await requests.next()).value?.accept(next); + hooks.acceptRequest((await requests.next()).value, next); expect(fresh.active.peek()?.epoch).toBe(NEWER); fresh.close(); next.close(); @@ -2517,7 +2517,7 @@ test("an unread resolved request retains its broadcast until its last handle clo const requests = dynamic.requested(); const held = origin.consume().request(path); const old = new BroadcastProducer(); - (await requests.next()).value?.accept(old); + hooks.acceptRequest((await requests.next()).value, old); // Change epochs before the active getter or its subscription can clone the answer. dynamic.updateRoute({ epoch: NEWER }); expect(old.closed.peek()).toBeUndefined(); @@ -2540,14 +2540,14 @@ test("a session's re-request after an epoch update asks the handler again", asyn const before = wire.demand(path); const old = new BroadcastProducer(); - (await requests.next()).value?.accept(old); + hooks.acceptRequest((await requests.next()).value, old); expect(await before).toBeDefined(); dynamic.updateRoute({ ...dynamic.route, epoch: NEWER }); const after = wire.demand(path); const next = new BroadcastProducer(); const track = next.createTrack("video"); - (await requests.next()).value?.accept(next); + hooks.acceptRequest((await requests.next()).value, next); const subscription = (await after)?.track("video").subscribe(); track.writeGroup(new GroupProducer(7)); expect((await subscription?.recvGroup())?.sequence).toBe(7); @@ -2571,7 +2571,7 @@ test("an update does not override route selection: another claim at the epoch ke const held = origin.consume().request(path); const answer = new BroadcastProducer(); - (await firstRequests.next()).value?.accept(answer); + hooks.acceptRequest((await firstRequests.next()).value, answer); await settle(); expect(held.active.peek()?.epoch).toBe(EPOCH); @@ -2581,7 +2581,7 @@ test("an update does not override route selection: another claim at the epoch ke const fresh = origin.consume().request(path); const replacement = new BroadcastProducer(); const track = replacement.createTrack("video"); - (await secondRequests.next()).value?.accept(replacement); + hooks.acceptRequest((await secondRequests.next()).value, replacement); await settle(); expect(held.active.peek()?.epoch).toBe(EPOCH); expect(fresh.active.peek()?.epoch).toBe(EPOCH); diff --git a/js/net/src/origin.ts b/js/net/src/origin.ts index 4afc110529e..2c54c2cb456 100644 --- a/js/net/src/origin.ts +++ b/js/net/src/origin.ts @@ -1274,7 +1274,7 @@ export class Producer implements Table { return this.#insertRoute(prefix, Route.default, true, true); } - /** Insert an originated claim with route metadata that answers through `accept`, for tests and benches. @internal */ + /** Insert an originated claim with route metadata that answers through `hooks.acceptRequest`, for tests and benches. @internal */ dynamicRoute( prefix: Path.Valid, route: @@ -2300,10 +2300,9 @@ export class Dynamic { /** * A pending request for a broadcast to be served on demand. * - * Yielded by {@link Dynamic.requested}. {@link prepare} transfers it to a prepared - * answer; {@link accept} resolves a forwarded request with a live broadcast; - * {@link reject} refuses it. Advancing or closing the iterator without transferring - * or answering rejects the request. + * Yielded by {@link Dynamic.requested}. {@link prepare} then {@link Prepared.publish} + * resolves it with a new exact-path broadcast; {@link reject} resolves it with an error. + * Advancing the iterator or closing it without either rejects the request. * * @public */ @@ -2341,6 +2340,7 @@ export class Request { request.#withdrawn = true; void err; }; + hooks.acceptRequest = (request, source) => request.#accept(source); } /** Start answering as a new publisher instance, hidden until published. */ @@ -2359,20 +2359,12 @@ export class Request { return makePrepared(this, answer, announcer, epoch); } - /** - * Resolve a session-forwarded request with its peer's routed broadcast. - * - * The caller keeps producing into `broadcast`; repeat requests for the path share - * it for as long as it stays live. - * - * A JS app does not proxy: a broadcast a session delivered would go out labeled with - * this origin's hop. To serve upstream content, copy its tracks into a prepared answer - * and publish it. - * - * @throws Error when handed a broadcast a session delivered. The request stays open. - * @internal - */ - accept(source: broadcast.Producer | broadcast.Consumer): void { + // Resolve a session-forwarded request with its peer's routed broadcast, through + // `hooks.acceptRequest`. The caller keeps producing into `source`; repeat requests for + // the path share it for as long as it stays live. Throws when handed a broadcast a + // session delivered to an originated claim, leaving the request open: it would go out + // labeled with this origin's hop. + #accept(source: broadcast.Producer | broadcast.Consumer): void { if (this.#done) { // A withdrawn request takes no answer. Only the handed-off consumer is ours; // the application still owns its producer and any other consumers. diff --git a/js/net/src/resume.test.ts b/js/net/src/resume.test.ts index 2593bc96c61..3da0165bfab 100644 --- a/js/net/src/resume.test.ts +++ b/js/net/src/resume.test.ts @@ -40,7 +40,7 @@ async function isPending(promise: Promise): Promise { function serve(origin: Producer, source: BroadcastProducer, cost: bigint) { const handle = wireOf(origin).receive(path, Route.normalize({ epoch: EPOCH, cost })); void (async () => { - for await (const request of handle.requested()) request.accept(source.consume()); + for await (const request of handle.requested()) hooks.acceptRequest(request, source.consume()); })(); return () => handle.close(); } @@ -823,7 +823,7 @@ test("between routes a reader keeps reading the copy it has", async () => { expect(drain(sub)).toEqual([1]); const { value: pending } = await asked.next(); - pending?.accept(b.consume()); + hooks.acceptRequest(pending, b.consume()); await settle(); await settle(); tb.write(1); diff --git a/js/net/src/untimed.test.ts b/js/net/src/untimed.test.ts index 921a0284739..3946c2f11b0 100644 --- a/js/net/src/untimed.test.ts +++ b/js/net/src/untimed.test.ts @@ -4,6 +4,7 @@ import { expect, test } from "bun:test"; import { type Consumer as BroadcastConsumer, Producer as BroadcastProducer } from "./broadcast.ts"; import { accept, connect } from "./connection/index.ts"; import * as Ietf from "./ietf/index.ts"; +import { hooks } from "./internal.ts"; import * as Lite from "./lite/index.ts"; import { createMockTransportPair } from "./mock.ts"; import { Producer as OriginProducer } from "./origin.ts"; @@ -133,7 +134,7 @@ test("a forwarded untimed track stays untimed", async () => { for await (const request of dynamic.requested()) { const input = wireOf(ingest).consume(request.path); inputs.push(input); - request.accept(await republish(input)); + hooks.acceptRequest(request, await republish(input)); } })(); diff --git a/js/watch/src/broadcast.test.ts b/js/watch/src/broadcast.test.ts index 6e1d34cdf19..0deccde227c 100644 --- a/js/watch/src/broadcast.test.ts +++ b/js/watch/src/broadcast.test.ts @@ -34,6 +34,16 @@ function broadcast(name: string, paths: string[] = [name]): { source: Broadcast; return { source, owner }; } +// Answer every request `claim` receives with a new exact broadcast, returning the answers. +function serve(claim: Origin.Dynamic, epoch?: Moq.Epoch.Valid): Moq.Broadcast.Producer[] { + const answers: Moq.Broadcast.Producer[] = []; + void (async () => { + for await (const request of claim.requested()) + answers.push(request.prepare(epoch ?? Moq.Epoch.mint()).publish()); + })(); + return answers; +} + function withoutWarnings(fn: () => T): T { const warn = console.warn; console.warn = () => {}; @@ -476,13 +486,12 @@ describe("blind resolution", () => { await settle(); - // Stand in for a dynamic route answering the request. - const upstream = new Moq.Broadcast.Producer(); - const route = owner.dynamicRoute(Path.from("blind.hang")); + // Stand in for a claim answering the request. + const route = owner.dynamic(Path.from("blind.hang")); const requests = route.requested(); const next = await requests.next(); - expect(next.done).toBe(false); - next.value?.accept(upstream); + if (!next.value) throw new Error("expected a request"); + const upstream = next.value.prepare(Moq.Epoch.mint()).publish(); await settle(); const active = source.out.active.peek(); @@ -495,6 +504,7 @@ describe("blind resolution", () => { expect(upstream.closed.peek()).toBeUndefined(); source.close(); + upstream.close(); route.close(); owner.close(); await settle(); @@ -590,18 +600,14 @@ describe("restart", () => { it("moves to a more specific announcement of another instance", async () => { const owner = new Origin.Producer(); const name = Path.from("pool/job.hang"); - const pool = owner.dynamicRoute(Path.from("pool"), { epoch: Moq.Epoch.mint() }); - const requests = pool.requested(); - const upstream = new Moq.Broadcast.Producer(); - void (async () => { - for await (const request of requests) request.accept(upstream); - })(); + const pool = owner.dynamic(Path.from("pool")); + const answers = serve(pool); const source = new Broadcast({ origin: owner, name, enabled: true, announced: true, catalogFormat: "manual" }); await settle(); const old = source.out.active.peek(); expect(old).toBeDefined(); - // A newer publisher announces the exact path, which now serves the name: a restart. + // A newer publisher announces the path, which now serves the name: a restart. const exact = owner.createBroadcast(name); exact.announce({ epoch: Moq.Epoch.mint() }); await settle(); @@ -616,7 +622,7 @@ describe("restart", () => { expect(source.out.error.peek()).toBeUndefined(); exact.close(); - upstream.close(); + for (const answer of answers) answer.close(); source.close(); owner.close(); await settle(); @@ -633,22 +639,18 @@ describe("restart", () => { const old = source.out.active.peek(); expect(old).toBeDefined(); - // The exact route goes, which ends the request on it, and the prefix arrives under the same + // The exact route goes, which ends the request on it, and a prefix answers under the same // epoch before the table is read again. The player still has to ask again. exact.close(); - const pool = owner.dynamicRoute(Path.from("pool"), { epoch }); - const requests = pool.requested(); - const upstream = new Moq.Broadcast.Producer(); - void (async () => { - for await (const request of requests) request.accept(upstream); - })(); + const pool = owner.dynamic(Path.from("pool")); + const answers = serve(pool, epoch); await settle(); const resumed = source.out.active.peek(); expect(resumed).toBeDefined(); expect(resumed).not.toBe(old); expect(source.out.error.peek()).toBeUndefined(); - upstream.close(); + for (const answer of answers) answer.close(); pool.close(); source.close(); owner.close(); @@ -658,12 +660,8 @@ describe("restart", () => { it("falls back to a covering prefix when the exact route ends", async () => { const owner = new Origin.Producer(); const name = Path.from("pool/job.hang"); - const pool = owner.dynamicRoute(Path.from("pool")); - const requests = pool.requested(); - const upstream = new Moq.Broadcast.Producer(); - void (async () => { - for await (const request of requests) request.accept(upstream); - })(); + const pool = owner.dynamic(Path.from("pool")); + const answers = serve(pool); const exact = publish(owner, name); const source = new Broadcast({ origin: owner, name, enabled: true, announced: true, catalogFormat: "manual" }); await settle(); @@ -680,11 +678,11 @@ describe("restart", () => { // Nothing covers the name any more: offline, not an error. pool.close(); + for (const answer of answers) answer.close(); await settle(); expect(source.out.active.peek()).toBeUndefined(); expect(source.out.error.peek()).toBeUndefined(); - upstream.close(); source.close(); owner.close(); await settle(); @@ -695,18 +693,16 @@ describe("refusal", () => { it("reports an unroutable refusal of a restarted publisher", async () => { const owner = new Origin.Producer(); const name = Path.from("room/live.hang"); - const upstream = new Moq.Broadcast.Producer(); - const first = owner.dynamicRoute(Path.from("room"), { epoch: Moq.Epoch.mint() }); - const firstRequests = first.requested(); - void (async () => { - for await (const request of firstRequests) request.accept(upstream); - })(); + const first = owner.dynamic(Path.from("room")); + const answers = serve(first); const source = new Broadcast({ origin: owner, name, enabled: true, announced: true, catalogFormat: "manual" }); await settle(); expect(source.out.active.peek()).toBeDefined(); - // The restarted publisher refuses with the same code an ended instance reports. - const second = owner.dynamicRoute(Path.from("room"), { epoch: Moq.Epoch.mint() }); + // The publisher restarts, and refuses with the same code an ended instance reports. + first.close(); + for (const answer of answers) answer.close(); + const second = owner.dynamic(Path.from("room")); const refused = await second.requested().next(); refused.value?.reject(new Moq.Error.Stream(Moq.StreamCode.Unroutable, { message: "refused" })); await settle(); @@ -714,8 +710,6 @@ describe("refusal", () => { expect(source.out.status.peek()).toBe("error"); second.close(); - first.close(); - upstream.close(); source.close(); owner.close(); await settle(); @@ -744,7 +738,7 @@ describe("refusal", () => { for (const announced of [true, false]) { it(`reports a refusal until a fresh request (announced: ${announced})`, async () => { const owner = new Origin.Producer(); - const route = owner.dynamicRoute(Path.from("room")); + const route = owner.dynamic(Path.from("room")); const requests = route.requested(); const name = new Signal(Path.from("room/refused.hang")); const enabled = new Signal(true); @@ -1026,7 +1020,7 @@ describe("wildcard renditions", () => { await settle(); expect(videoRenditions(source)).toEqual(["source"]); - const worker = owner.dynamicRoute(Path.from(".pro/transcode")); + const worker = owner.dynamic(Path.from(".pro/transcode")); const requests = worker.requested(); await settle(); expect(videoRenditions(source)).toEqual(["source", "transcode"]); @@ -1040,25 +1034,22 @@ describe("wildcard renditions", () => { expect(request?.path).toBe(derived); expect(active).toBeUndefined(); - const produced = new Moq.Broadcast.Producer(); - produced + if (!request) throw new Error("expected a request"); + const prepared = request.prepare(Moq.Epoch.mint()); + prepared.broadcast .createTrack("video", { timescale: Moq.Time.Timescale.MILLI }) .writeFrame({ payload: new TextEncoder().encode("frame"), timestamp: Moq.Time.Timestamp.now() }); - request?.accept(produced); + const produced = prepared.publish(); await settle(); expect(active).toBeDefined(); const track = active?.track("video").subscribe().ordered(); expect(await track?.readString()).toBe("frame"); track?.close(); - // The worker announcing the path it now serves leaves the rendition where it was. - const concrete = owner.createBroadcast(derived); - concrete.announce(); - await settle(); + // The answer announcing the path it now serves leaves the rendition where it was. expect(videoRenditions(source)).toEqual(["source", "transcode"]); - expect(active).toBeDefined(); - concrete.close(); + produced.close(); worker.close(); } finally { effect.close(); @@ -1075,8 +1066,8 @@ describe("wildcard renditions", () => { try { // A catch-all (an archive) and a narrower pool both cover the derived path. - const archive = owner.dynamicRoute(Path.empty()); - const worker = owner.dynamicRoute(Path.from(".pro/transcode")); + const archive = owner.dynamic(Path.empty()); + const worker = owner.dynamic(Path.from(".pro/transcode")); await settle(); expect(videoRenditions(source)).toEqual(["source", "transcode"]); diff --git a/quest/m1/announce-takeover.md b/quest/m1/announce-takeover.md index 3ff73d8184c..53f11a0594b 100644 --- a/quest/m1/announce-takeover.md +++ b/quest/m1/announce-takeover.md @@ -79,6 +79,9 @@ takeover; conflicting epochs; equal clamped prefixes; independently ending parent/child; late snapshots; multi-hop and diamond relays; mixed versions; and held catalog/media pins. Include the actual players and composer-facing event, not only route-table checks. Measure route/readers fanout on both axes. +Remove the cold-start warm-up reads that discard a claim's first subscription +(`claimed` in moq-net `tests/integration/restart.rs` and the `drained_claim.rs` +helper) and assert that first subscription succeeds instead. Run `just check`, `just test interop --all`, and `just drafts check`; wire regressions and benchmarks into existing CI lanes. diff --git a/quest/m1/api-audit/README.md b/quest/m1/api-audit/README.md index 1c411d65c7b..fb0f80b2777 100644 --- a/quest/m1/api-audit/README.md +++ b/quest/m1/api-audit/README.md @@ -72,7 +72,6 @@ Rejected in planning: - [One frame rate name](/quest/m1/api-audit/frame-rate.md) - core `moq_video::Rate` becomes `FrameRate`, as the bindings name it - [Pinned consumers adopt the CLI flag names](/quest/m1/api-audit/cli-flags-consumers.md) - pinned downstream deployments move past the flag rename and use its spellings - [Session-only items stay private](/quest/m1/api-audit/model-private.md) - model items only the built-in sessions use are private -- [Only sessions accept a request](/quest/m1/api-audit/claim-request-type.md) - accepting an application claim's request fails to compile, and the peer-route hatches are crate-private - [A request's source enforces its validity](/quest/m1/api-audit/answer.md) - a stale source refuses new track requests by itself, with no public additions - [moq-transport stays private](/quest/m1/api-audit/ietf-private.md) - no public moq-net item exists only for moq-transport, and a client names its request target once as a URL - [One lifecycle vocabulary](/quest/m1/api-audit/lifecycle.md) - finish, abort, closed, and close mean one thing on every handle, and Rust and JS observers fire alike @@ -81,6 +80,5 @@ Rejected in planning: ## Related -- [Claim request type](/quest/m1/api-audit/claim-request-type.md) - removes `accept` from application requests; [the origin announces](/quest/m1/api-audit/announce-owner.md) narrows `Prepared` beside it - [API-diff check on release PRs](/quest/m1/api-diff.md) - the backstop that catches breaks this epic misses - [Main is no slower than release](/quest/m1/release-perf/README.md) - the other release-readiness check diff --git a/quest/m1/api-audit/announce-owner.md b/quest/m1/api-audit/announce-owner.md index 28c7f9c6f79..5c4f27dad68 100644 --- a/quest/m1/api-audit/announce-owner.md +++ b/quest/m1/api-audit/announce-owner.md @@ -25,7 +25,3 @@ error. Public API: breaking in moq-net (and `@moq/net`/moq-ffi if they mirror the trap). Wire: none. - -## Related - -- [Claim request type](/quest/m1/api-audit/claim-request-type.md) - splits `accept` off application requests in the same file; coordinate the order diff --git a/quest/m1/api-audit/claim-request-type.md b/quest/m1/api-audit/claim-request-type.md deleted file mode 100644 index 02ed19653e8..00000000000 --- a/quest/m1/api-audit/claim-request-type.md +++ /dev/null @@ -1,47 +0,0 @@ -# [S] Only sessions accept a request - -## Goal - -Answering an application claim's request with `accept` fails to compile in -Rust and to typecheck in TypeScript instead of panicking or throwing at -runtime. The peer-route hatches are crate-private. - -## Plan - -Decided 2026-10-21 (maintainer), after a downstream application's handler -called `request.accept(..)` on a `dynamic()` claim, which compiled and would -have panicked on the first request. Reshaped 2026-10-23 after the -public-model goal was dropped -([session-only items stay private](/quest/m1/api-audit/model-private.md)), -for [moq-dev/moq#6496](https://github.com/moq-dev/moq/pull/6496): - -- `origin::Producer::dynamic_route`, `origin::Dynamic::update_route`, and - `origin::Request::accept` become `pub(crate)` (JS `dynamicRoute`, - `updateRoute`, and `Request.accept` are already `@internal`). An - application answers with `prepare` then `publish`, or `reject`. -- Move the out-of-crate tests and benches that call them (moq-mux, moq-hls, - moq-cli, moq-gst, moq-c, moq-ffi, moq-net integration, js/watch) onto - `prepare`/`publish`. -- No `Forward`/`ForwardRequest` split: visibility already makes the call - unreachable. Delete the runtime assert and its test; a compile-fail - doctest and a `@ts-expect-error` line replace them. - -Kept from [moq-dev/moq#5678](https://github.com/moq-dev/moq/pull/5678): -prepare and publish, one epoch per answer, and the cold-start reset until -[announce takeover](/quest/m1/announce-takeover.md) lands. - -Parked 2026-10-24 (maintainer) until announce takeover lands: #6496's -conformance tests subscribe through a `dynamic()` claim and get `Unroutable` -from the cold-start reset (the worker's answer arrives under a fresh epoch), -not from a missing route; `dynamic()` already builds the same route as -`dynamic_route` minus `publishes: false`. Takeover removes the reset at its -source. On resuming, main has new out-of-crate callers to move: `wide_hop.rs`, -moq-hls `export/rendition.rs`, moq-cli `play/media.rs`, moq-mux -`benches/export.rs`, and moq-gst `source/imp.rs`, which used the removed -`Publisher::standalone` and needs a redesign. - -Public API: moq-net breaking (three hidden items go private). Wire: none. - -## Required - -- [Announcement takeover](/quest/m1/announce-takeover.md) - removes the cold-start reset the claim's first subscribe hits diff --git a/quest/m1/api-audit/requests.md b/quest/m1/api-audit/requests.md index b63c6d63bc8..de9cf1127bf 100644 --- a/quest/m1/api-audit/requests.md +++ b/quest/m1/api-audit/requests.md @@ -15,16 +15,11 @@ mechanism; `auth::Handle::requests() -> auth::Requests` is the precedent. `track::Request::dynamic()`) to `requests()`, and each `Dynamic` type to `Requests`. JS `dynamic()`/`Dynamic` mirror it. moq-ffi and its wrappers follow; the frozen hand-written moq-c ABI stays as it is. -- The peer-route hatches (`dynamic_route`, `update_route`) are crate-private - after [claim request type](/quest/m1/api-audit/claim-request-type.md); - rename them with the rest. +- Rename the crate-private peer-route hatches (`dynamic_route`, + `update_route`) with the rest. - Quests that say `Dynamic` (subscribe ranges, archive) switch to the new name in this PR. About 240 Rust, 30 binding, and 10 JS call sites; land it in one sweep. Public API: moq-net, `@moq/net`, moq-ffi and wrappers renamed. Wire: none. - -## Related - -- [Claim request type](/quest/m1/api-audit/claim-request-type.md) - reshapes the origin claim this renames diff --git a/rs/moq-c/src/test.rs b/rs/moq-c/src/test.rs index 0b74ccfa43e..f25e71b9899 100644 --- a/rs/moq-c/src/test.rs +++ b/rs/moq-c/src/test.rs @@ -3758,13 +3758,7 @@ fn consume_announced_delivers_a_takeover_once(dynamic_at: &str) { let dynamic = { let _runtime = crate::ffi::RUNTIME.enter(); let origin = crate::Id::try_from(origin).unwrap(); - let route = moq_net::origin::Route::default(); - State::lock() - .origin - .get(origin) - .unwrap() - .dynamic_route(dynamic_at, route) - .unwrap() + State::lock().origin.get(origin).unwrap().dynamic(dynamic_at).unwrap() }; let cb = Callback::new(); let path = b"live"; @@ -3806,16 +3800,11 @@ fn consume_announced_delivers_a_held_takeover_once() { consume_announced_delivers_a_takeover_once("live"); } -/// Serve `prefix` on `origin` dynamically through `route`. -fn serve_dynamic(origin: u32, prefix: &str, route: moq_net::origin::Route) -> moq_net::origin::Dynamic { +/// Serve `prefix` on `origin` dynamically. +fn serve_dynamic(origin: u32, prefix: &str) -> moq_net::origin::Dynamic { let _runtime = crate::ffi::RUNTIME.enter(); let origin = crate::Id::try_from(origin).unwrap(); - State::lock() - .origin - .get(origin) - .unwrap() - .dynamic_route(prefix, route) - .unwrap() + State::lock().origin.get(origin).unwrap().dynamic(prefix).unwrap() } /// Wait for the broadcast at `path` on `origin`, delivering each instance through `cb`. @@ -3847,9 +3836,9 @@ extern "C" fn gated_callback(user_data: *mut c_void, code: i32) { } } -/// An end is a gap that ended every request on the old route, even when a route above the path -/// with the same epoch carries the instance on. So the instance serving the path afterwards is -/// delivered again, though the caller still holds its handle. +/// An end is a gap that ended every request on the old route, even when a claim above the path +/// answers under the same epoch. So the instance serving the path afterwards is delivered again, +/// though the caller still holds a handle to its epoch. #[test] fn consume_announced_delivers_the_instance_again_after_an_end() { let origin = id(moq_origin_create()); @@ -3877,7 +3866,7 @@ fn consume_announced_delivers_the_instance_again_after_an_end() { // Hold the wait inside that delivery so both changes are on hand when it next looks. Seeing // the route above stand before the path's own route left would prove no gap: an update. - let dynamic = serve_dynamic(origin, "", route); + let dynamic = serve_dynamic(origin, ""); { let _runtime = crate::ffi::RUNTIME.enter(); run.unannounce(); @@ -3886,9 +3875,8 @@ fn consume_announced_delivers_the_instance_again_after_an_end() { let request = crate::ffi::RUNTIME.block_on(dynamic.requested_broadcast()).unwrap(); let served = { let _runtime = crate::ffi::RUNTIME.enter(); - let served = moq_net::broadcast::Info::new().produce(); - request.accept(served.consume()); - served + let epoch = route.epoch.clone().unwrap(); + request.prepare(epoch).unwrap().publish().unwrap() }; let again = id(recv()); @@ -3905,7 +3893,7 @@ fn consume_announced_delivers_the_instance_again_after_an_end() { #[test] fn consume_announced_delivers_a_freed_takeover_again() { let origin = id(moq_origin_create()); - let dynamic = serve_dynamic(origin, "live", moq_net::origin::Route::default()); + let dynamic = serve_dynamic(origin, "live"); let cb = Callback::new(); let task = consume_announced(origin, b"live", &cb); @@ -3928,7 +3916,7 @@ fn consume_announced_delivers_a_freed_takeover_again() { #[test] fn consume_announced_delivers_a_gap_while_resolving_once() { let origin = id(moq_origin_create()); - let dynamic = serve_dynamic(origin, "", moq_net::origin::Route::default()); + let dynamic = serve_dynamic(origin, ""); let cb = Callback::new(); let task = consume_announced(origin, b"live", &cb); diff --git a/rs/moq-cli/src/play/media.rs b/rs/moq-cli/src/play/media.rs index 4d25df38539..1ea1566605d 100644 --- a/rs/moq-cli/src/play/media.rs +++ b/rs/moq-cli/src/play/media.rs @@ -1588,6 +1588,26 @@ mod tests { (broadcast, catalog, audio) } + /// Answer the next request through `claim` with one `audio` rendition, under `route`'s epoch. + async fn answer( + claim: &moq_net::origin::Dynamic, + route: &moq_net::origin::Route, + ) -> ( + moq_net::broadcast::Producer, + catalog::Producer, + moq_mux::container::Producer, + ) { + let request = tokio::time::timeout(Duration::from_secs(5), claim.requested_broadcast()) + .await + .expect("the claim is asked") + .unwrap(); + let prepared = request.prepare(route.epoch.clone().expect("an epoch")).unwrap(); + let mut broadcast = prepared.broadcast().clone(); + let catalog = catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let audio = rendition(&broadcast, &catalog, "audio"); + (prepared.publish().unwrap(), catalog, audio) + } + /// Write `count` packets of `sample` in real time, its timeline starting at 0. async fn write(audio: &mut moq_mux::container::Producer, count: u64, sample: f32) { for index in 0..count { @@ -1657,7 +1677,7 @@ mod tests { #[tokio::test(start_paused = true)] async fn a_refused_broadcast_ends_the_player() { let origin = moq_tokio::origin::spawn(); - let handler = origin.dynamic_route("", Default::default()).unwrap(); + let handler = origin.dynamic("").unwrap(); let recorder = Recorder::default(); let player = tokio::spawn(media(&origin, Duration::from_millis(50), recorder.clone()).run()); let request = handler.requested_broadcast().await.unwrap(); @@ -1819,17 +1839,9 @@ mod tests { // The broadcast ends while its route stands, so the player goes idle. tokio::time::sleep(Duration::from_secs(60)).await; - let mut served = moq_net::broadcast::Info::new().produce(); - let served_catalog = catalog::Producer::new(&mut served, Default::default()).unwrap(); - let mut new = rendition(&served, &served_catalog, "audio"); - let pool = origin.dynamic_route("", route).unwrap(); + let pool = origin.dynamic("").unwrap(); drop(exact); - let consumer = served.consume(); - tokio::spawn(async move { - while let Ok(request) = pool.requested_broadcast().await { - request.accept(consumer.clone()); - } - }); + let (served, served_catalog, mut new) = answer(&pool, &route).await; write(&mut new, 10, NEW).await; settle(player, &recorder).await; @@ -1838,7 +1850,7 @@ mod tests { PACKET_DURATION * 10, "the idle player never played the covering route" ); - drop((served, served_catalog)); + drop((served, served_catalog, pool)); } /// The exact route goes while the old run still holds its delay, and a covering @@ -1854,27 +1866,18 @@ mod tests { let origin = moq_tokio::origin::spawn(); let route = moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()); - // One instance, served by the exact route and mirrored behind a covering one. + // One instance, served by the exact route and then answered again behind a covering one. let (exact, exact_catalog, mut old) = publish(&origin, route.clone()); - let mut mirror = moq_net::broadcast::Info::new().produce(); - let mirror_catalog = catalog::Producer::new(&mut mirror, Default::default()).unwrap(); - let mut copy = rendition(&mirror, &mirror_catalog, "audio"); let recorder = Recorder::default(); let player = tokio::spawn(media(&origin, Duration::from_millis(500), recorder.clone()).run()); for index in 0..50 { old.write(packet(index, OLD)).unwrap(); - copy.write(packet(index, OLD)).unwrap(); tokio::time::sleep(PACKET_DURATION).await; } exact.unannounce(); - let pool = origin.dynamic_route("", route).unwrap(); - let consumer = mirror.consume(); - tokio::spawn(async move { - while let Ok(request) = pool.requested_broadcast().await { - request.accept(consumer.clone()); - } - }); + let pool = origin.dynamic("").unwrap(); + let (mirror, mirror_catalog, mut copy) = answer(&pool, &route).await; for index in 50..60 { copy.write(packet(index, NEW)).unwrap(); tokio::time::sleep(PACKET_DURATION).await; @@ -1885,7 +1888,7 @@ mod tests { played(&recorder, NEW).0 > Duration::ZERO, "the player never reached the covering route" ); - drop((mirror, mirror_catalog, copy, exact, exact_catalog, old)); + drop((mirror, mirror_catalog, copy, pool, exact, exact_catalog, old)); } /// Re-pricing the same instance is an update, which playback rides through on diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index 36d9f34201a..c0bede188d1 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -5396,9 +5396,9 @@ async fn announced_broadcast_hands_out_a_held_takeover_once() { announced_broadcast_hands_out_a_takeover_once("live").await; } -/// An end is a gap that ended every request on the old route, even when a route above the path -/// with the same epoch carries the instance on. So the instance serving the path afterwards is -/// handed out again, though it is the one the caller still holds. +/// An end is a gap that ended every request on the old route, even when a claim above the path +/// answers under the same epoch. So the instance serving the path afterwards is handed out +/// again, though its epoch is the one the caller still holds. #[tokio::test] async fn announced_broadcast_hands_out_the_instance_again_after_an_end() { let origin = MoqOriginProducer::new(MoqOriginConfig::default()); @@ -5414,29 +5414,26 @@ async fn announced_broadcast_hands_out_the_instance_again_after_an_end() { .expect("timed out waiting for the broadcast") .unwrap(); - // A peer's covering route under the same epoch, which the public claim cannot express. - let dynamic = origin - .inner() - .inner - .dynamic_route("", route.clone().try_into().unwrap()) - .unwrap(); + // A covering claim answering under the same epoch. + let dynamic = origin.inner().inner.dynamic("").unwrap(); broadcast.unannounce().unwrap(); let again = tokio::spawn(async move { followed.available().await }); let request = tokio::time::timeout(TIMEOUT, dynamic.requested_broadcast()) .await .expect("the instance never asked the route above the path") .unwrap(); - let served = moq_net::broadcast::Info::new().produce(); - request.accept(served.consume()); + let epoch = route.epoch.clone().unwrap().parse().unwrap(); + let served = request.prepare(epoch).unwrap().publish().unwrap(); let again = tokio::time::timeout(TIMEOUT, again) .await .expect("the instance was never handed out again") .unwrap() .unwrap(); - assert!( - again.inner().is_clone(first.inner()), - "the route above did not carry the instance on" + assert_eq!( + again.inner().info().epoch, + first.inner().info().epoch, + "the claim above did not answer the same instance" ); drop((served, dynamic)); broadcast.close().unwrap(); diff --git a/rs/moq-gst/src/source/imp.rs b/rs/moq-gst/src/source/imp.rs index 629b8629749..876d1ddffca 100644 --- a/rs/moq-gst/src/source/imp.rs +++ b/rs/moq-gst/src/source/imp.rs @@ -3077,11 +3077,6 @@ mod session_tests { publisher } - /// A broadcast at no path, for a covering route to serve. - fn standalone(config: VideoConfig) -> Self { - Self::build(moq_net::broadcast::Info::new().produce(), config) - } - fn build(mut broadcast: moq_net::broadcast::Producer, config: VideoConfig) -> Self { let mut catalog = moq_mux::catalog::Producer::new(&mut broadcast, moq_mux::catalog::Config::default()).unwrap(); @@ -3479,8 +3474,8 @@ mod session_tests { restart(moq_net::origin::Route::default); } - /// The route serving the path can be a prefix covering it, here the root, and its restart - /// switches too. + /// The route serving the path can be a prefix covering it, here the root, and a restart of + /// its answer switches too. #[test] fn a_covering_route_restart_switches_on_the_same_pad() { let _pad_ids = pad_ids(); @@ -3488,22 +3483,24 @@ mod session_tests { let recorder = Recorder::attach(&element); let origin = origin(); - let mut old = Publisher::standalone(video_rendition()); - let serving = std::sync::Arc::new(Mutex::new(old.broadcast.consume())); - let pool = std::sync::Arc::new(origin.dynamic_route("", epoch()).unwrap()); - let (handler, answer) = (pool.clone(), serving.clone()); + let claim = origin.dynamic("").unwrap(); + let (answers, answered) = std::sync::mpsc::channel(); let handler = super::RUNTIME.spawn(async move { - while let Ok(request) = handler.requested_broadcast().await { - request.accept(answer.lock().unwrap().clone()); + while let Ok(request) = claim.requested_broadcast().await { + let prepared = request.prepare(moq_net::Epoch::mint()).unwrap(); + let answer = Publisher::build(prepared.broadcast().clone(), video_rendition()); + prepared.publish().unwrap(); + answers.send(answer).unwrap(); } }); let session = follow(&element, &origin); + let mut old = answered + .recv_timeout(Duration::from_secs(5)) + .expect("the claim answered"); let pad = recorder.written(&mut old, 1); - let mut new = Publisher::standalone(video_rendition()); - *serving.lock().unwrap() = new.broadcast.consume(); - pool.update_route(epoch()).unwrap(); + let mut new = Publisher::new(&origin, epoch(), video_rendition()); assert_eq!(recorder.written(&mut new, 2), pad, "the restart moved to another pad"); stop(session); @@ -3511,15 +3508,15 @@ mod session_tests { recorder.assert_one_pad_held(); } - /// A route that never answers must not hold the reader on a stale resolution once a - /// replacement serves the path. An answer arriving late cannot replace the chosen run. + /// A claim that never answers must not hold the reader on a stale resolution once a + /// replacement serves the path. Its late refusal cannot end the chosen run. #[test] fn a_replacement_supersedes_a_pending_broadcast_request() { let _pad_ids = pad_ids(); let element = element(); let recorder = Recorder::attach(&element); let origin = origin(); - let pool = origin.dynamic_route("", epoch()).unwrap(); + let pool = origin.dynamic("").unwrap(); let session = follow(&element, &origin); let pending = super::RUNTIME.block_on(async { tokio::time::timeout(Duration::from_secs(10), pool.requested_broadcast()) @@ -3529,12 +3526,9 @@ mod session_tests { }); let mut new = Publisher::new(&origin, epoch(), video_rendition()); recorder.written(&mut new, 2); - let mut old = Publisher::standalone(video_rendition()); - pending.accept(old.broadcast.consume()); - old.write(1); + pending.reject(moq_net::Error::Unauthorized); recorder.written(&mut new, 3); stop(session); - assert!(!recorder.seen().iter().any(|seen| matches!(seen, Seen::Buffer(_, 1)))); recorder.assert_one_pad_held(); } diff --git a/rs/moq-hls/src/export/mod.rs b/rs/moq-hls/src/export/mod.rs index 7c73d43fe84..cd77c357f38 100644 --- a/rs/moq-hls/src/export/mod.rs +++ b/rs/moq-hls/src/export/mod.rs @@ -3288,8 +3288,7 @@ mod tests { moq_net::origin::Route::default().with_hops(hops(&[first])) } - /// A standalone media publisher (not an origin leaf), so the sibling is served through a - /// routed front. + /// Write three keyframes of `payload` into `broadcast`'s `video0` track, recording each. fn write_routed_media( broadcast: &mut moq_net::broadcast::Producer, payload: &'static [u8], @@ -3312,14 +3311,6 @@ mod tests { media } - async fn accept_sibling(server: &moq_net::origin::Dynamic, broadcast: &moq_net::broadcast::Producer) { - let request = tokio::time::timeout(Duration::from_secs(5), server.requested_broadcast()) - .await - .expect("the sibling path is requested") - .expect("the handler is open"); - request.accept(broadcast); - } - async fn until_finished(rendition: &Rendition) { let deadline = tokio::time::Instant::now() + Duration::from_secs(5); loop { @@ -3348,9 +3339,8 @@ mod tests { let catalog = moq_mux::catalog::Producer::new(&mut live, moq_mux::catalog::Config::default()).unwrap(); let recorder = catalog.enroll_test("video0").unwrap(); - let mut old_media = moq_net::broadcast::Info::new().produce(); + let mut old_media = origin.create_broadcast("media").unwrap(); let _old_track = write_routed_media(&mut old_media, OLD, recorder, 0); - let old_server = origin.dynamic_route("media", sibling_route(10)).unwrap(); settle().await; let upstream = crate::resolved(&origin, "live").await; @@ -3367,7 +3357,7 @@ mod tests { let reference = Arc::new((Kind::Video, "video0".to_string(), "video0".to_string())); let watcher = tokio::spawn(watch_timeline(upstream.broadcast(), archive, reference, fanout.clone())); let rendition = renditions.get(Kind::Video, "video0").unwrap(); - accept_sibling(&old_server, &old_media).await; + old_media.announce(sibling_route(10)).unwrap(); tokio::time::timeout(Duration::from_secs(5), rendition.playable()) .await .expect("the timeline reaches the rendition"); @@ -3381,7 +3371,7 @@ mod tests { let mut early = rendition.segments(); assert_eq!(early.next().await.unwrap().unwrap().discontinuity, 0); - drop((old_server, old_media, _old_track)); + drop((old_media, _old_track)); until_finished(&rendition).await; let served = rendition .listed_segment(0) @@ -3395,24 +3385,18 @@ mod tests { } // A publisher serving the path afterwards, with rows recorded for it, is never bound. - let new_server = origin.dynamic_route("media", sibling_route(11)).unwrap(); + let mut new_media = origin.create_broadcast("media").unwrap(); + new_media.announce(sibling_route(11)).unwrap(); let _ = rendition.snapshot(); - assert!( - tokio::time::timeout(Duration::from_secs(1), new_server.requested_broadcast()) - .await - .is_err(), - "the replacement is never requested" - ); let recorder = catalog.enroll_test("video0").unwrap(); fanout.skip(None); - let mut new_media = moq_net::broadcast::Info::new().produce(); let _new_track = write_routed_media(&mut new_media, NEW, recorder, 6_000_000); tokio::time::sleep(Duration::from_secs(1)).await; let snapshot = rendition.snapshot(); assert!(snapshot.segments.is_empty(), "nothing of the replacement is listed"); assert!(snapshot.finished); watcher.abort(); - drop((new_media, _new_track, new_server, catalog, live)); + drop((new_media, _new_track, catalog, live)); } // A restart can land while a segment request waits for its media broadcast. The init @@ -3429,9 +3413,8 @@ mod tests { let catalog = moq_mux::catalog::Producer::new(&mut live, moq_mux::catalog::Config::default()).unwrap(); let recorder = catalog.enroll_test("video0").unwrap(); - let mut media = moq_net::broadcast::Info::new().produce(); + let mut media = origin.create_broadcast("media").unwrap(); let _track = write_routed_media(&mut media, KEYFRAME, recorder, 0); - let server = origin.dynamic_route("media", sibling_route(10)).unwrap(); settle().await; let upstream = crate::resolved(&origin, "live").await; @@ -3452,7 +3435,7 @@ mod tests { .await .expect("the timeline reaches the rendition"); - // The sibling request is unanswered, so the segment waits for its media broadcast. + // The sibling is unannounced, so the segment waits for its media broadcast. let mut segment = std::pin::pin!(rendition.listed_segment(0)); assert!( tokio::time::timeout(Duration::from_millis(10), segment.as_mut()) @@ -3462,7 +3445,7 @@ mod tests { fanout.set_generation(Some("run-1".into())); fanout.set_generation(Some("run-2".into())); - accept_sibling(&server, &media).await; + media.announce(sibling_route(10)).unwrap(); segment .await .unwrap() @@ -3486,9 +3469,9 @@ mod tests { let catalog = moq_mux::catalog::Producer::new(&mut live, moq_mux::catalog::Config::default()).unwrap(); let recorder = catalog.enroll_test("video0").unwrap(); - let mut media = moq_net::broadcast::Info::new().produce(); + let mut media = origin.create_broadcast("media").unwrap(); let _track = write_routed_media(&mut media, NEW, recorder, 0); - let server = origin.dynamic_route("media", sibling_route(11)).unwrap(); + media.announce(sibling_route(11)).unwrap(); settle().await; let upstream = crate::resolved(&origin, "live").await; @@ -3496,7 +3479,6 @@ mod tests { config.broadcast = Some(moq_net::path::Relative::new("media").to_owned()); let (rendition, watcher) = export(&upstream, &config); - accept_sibling(&server, &media).await; tokio::time::timeout(Duration::from_secs(5), rendition.playable()) .await .expect("the timeline reaches the rendition"); @@ -3507,7 +3489,7 @@ mod tests { .expect("the sibling present at export start is servable"); assert!(contains(&served, NEW)); watcher.abort(); - drop((media, _track, server, catalog, live)); + drop((media, _track, catalog, live)); } // The replacement is often announced after the incumbent is already gone. Nothing serves @@ -3525,9 +3507,8 @@ mod tests { let catalog = moq_mux::catalog::Producer::new(&mut live, moq_mux::catalog::Config::default()).unwrap(); let recorder = catalog.enroll_test("video0").unwrap(); - let mut old_media = moq_net::broadcast::Info::new().produce(); + let mut old_media = origin.create_broadcast("media").unwrap(); let _old_track = write_routed_media(&mut old_media, OLD, recorder, 0); - let old_server = origin.dynamic_route("media", sibling_route(10)).unwrap(); settle().await; let upstream = crate::resolved(&origin, "live").await; @@ -3535,13 +3516,13 @@ mod tests { config.broadcast = Some(moq_net::path::Relative::new("media").to_owned()); let (rendition, watcher) = export(&upstream, &config); - accept_sibling(&old_server, &old_media).await; + old_media.announce(sibling_route(10)).unwrap(); tokio::time::timeout(Duration::from_secs(5), rendition.playable()) .await .expect("the timeline reaches the rendition"); assert!(rendition.listed_segment(0).await.unwrap().is_some()); - drop((old_server, old_media, _old_track)); + drop((old_media, _old_track)); until_finished(&rendition).await; let before = rendition.snapshot().segments.len(); assert!(before > 0); @@ -3552,17 +3533,12 @@ mod tests { .expect("the old retained segment drains"); assert!(contains(&served, OLD)); - let new_server = origin.dynamic_route("media", sibling_route(11)).unwrap(); - let mut new_media = moq_net::broadcast::Info::new().produce(); + let mut new_media = origin.create_broadcast("media").unwrap(); let recorder = catalog.enroll_test("video0").unwrap(); let _new_track = write_routed_media(&mut new_media, NEW, recorder, 6_000_000); + new_media.announce(sibling_route(11)).unwrap(); let _ = rendition.snapshot(); - assert!( - tokio::time::timeout(Duration::from_secs(1), new_server.requested_broadcast()) - .await - .is_err(), - "the replacement is never requested" - ); + tokio::time::sleep(Duration::from_secs(1)).await; let snapshot = rendition.snapshot(); assert_eq!(snapshot.segments.len(), before, "only the old rows remain listed"); assert!(snapshot.finished, "the playlist ended with its sibling"); @@ -3574,7 +3550,7 @@ mod tests { assert!(contains(&served, OLD)); assert!(!contains(&served, NEW)); watcher.abort(); - drop((new_media, _new_track, new_server, catalog, live)); + drop((new_media, _new_track, catalog, live)); } // A recorder waiting at the live edge ends when the bound sibling ends, with nothing else @@ -3589,16 +3565,15 @@ mod tests { live.announce(Default::default()).expect("publish allowed"); let catalog = moq_mux::catalog::Producer::new(&mut live, moq_mux::catalog::Config::default()).unwrap(); let recorder = catalog.enroll_test("video0").unwrap(); - let mut media = moq_net::broadcast::Info::new().produce(); + let mut media = origin.create_broadcast("media").unwrap(); let track = write_routed_media(&mut media, OLD, recorder, 0); - let server = origin.dynamic_route("media", sibling_route(10)).unwrap(); settle().await; let upstream = crate::resolved(&origin, "live").await; let mut config = video_config(); config.broadcast = Some(moq_net::path::Relative::new("media").to_owned()); let (rendition, watcher) = export(&upstream, &config); - accept_sibling(&server, &media).await; + media.announce(sibling_route(10)).unwrap(); tokio::time::timeout(Duration::from_secs(5), rendition.playable()) .await .expect("the timeline reaches the rendition"); @@ -3616,7 +3591,7 @@ mod tests { } // Only the sibling broadcast goes: its recorder stays, so the timeline stays quiet. - drop((server, media)); + drop(media); let end = tokio::time::timeout(Duration::from_secs(5), cursor.next()) .await .expect("the cursor wakes when its sibling ends") @@ -3638,7 +3613,7 @@ mod tests { live.announce(Default::default()).expect("publish allowed"); let catalog = moq_mux::catalog::Producer::new(&mut live, moq_mux::catalog::Config::default()).unwrap(); let recorder = catalog.enroll_test("video0").unwrap(); - let mut media = moq_net::broadcast::Info::new().produce(); + let mut media = origin.create_broadcast("media").unwrap(); let _track = write_routed_media(&mut media, OLD, recorder, 0); settle().await; @@ -3651,9 +3626,8 @@ mod tests { .await .expect("the catalog's rows are listed before the sibling resolves"); - let server = origin.dynamic_route("media", sibling_route(10)).unwrap(); let _ = rendition.snapshot(); - accept_sibling(&server, &media).await; + media.announce(sibling_route(10)).unwrap(); let served = rendition .listed_segment(0) .await @@ -3661,7 +3635,7 @@ mod tests { .expect("the late sibling is servable"); assert!(contains(&served, OLD)); watcher.abort(); - drop((media, _track, server, catalog, live)); + drop((media, _track, catalog, live)); } // A rendition the catalog drops must end its segment cursors. The cursor holds an diff --git a/rs/moq-hls/src/export/rendition.rs b/rs/moq-hls/src/export/rendition.rs index d4fddf57491..e6630cd7ec5 100644 --- a/rs/moq-hls/src/export/rendition.rs +++ b/rs/moq-hls/src/export/rendition.rs @@ -1359,13 +1359,13 @@ mod tests { async fn pending_reads_preserve_a_later_authorization_refusal() { let origin = moq_tokio::origin::spawn(); let broadcast = origin.create_broadcast("room/live").unwrap(); - let dynamic = origin.dynamic_route("room/secret", Default::default()).unwrap(); + let dynamic = origin.dynamic("room/secret").unwrap(); let source = moq_mux::Source::new(origin.consume(), broadcast.consume()); let mut config = VideoConfig::new(hang::catalog::VideoCodec::VP8); config.broadcast = Some(moq_net::path::Relative::new("./secret").to_owned()); let rendition = Arc::new(Rendition::video("video".into(), &config, &source, Archive::default(), None).unwrap()); let request = dynamic.requested_broadcast().await.unwrap(); - let other = origin.dynamic_route("room/hidden", Default::default()).unwrap(); + let other = origin.dynamic("room/hidden").unwrap(); config.broadcast = Some(moq_net::path::Relative::new("./hidden").to_owned()); let resolving = Arc::new(Rendition::video("video".into(), &config, &source, Archive::default(), None).unwrap()); let other_request = other.requested_broadcast().await.unwrap(); diff --git a/rs/moq-mux/benches/containers.rs b/rs/moq-mux/benches/containers.rs index b81c6a2ce0b..d7137b0dde0 100644 --- a/rs/moq-mux/benches/containers.rs +++ b/rs/moq-mux/benches/containers.rs @@ -34,13 +34,11 @@ struct Fixture { _published: Box, _origin: moq_net::origin::Producer, driver: tokio::task::JoinHandle, - requests: tokio::task::JoinHandle<()>, } impl Drop for Fixture { fn drop(&mut self) { self.driver.abort(); - self.requests.abort(); } } @@ -56,8 +54,9 @@ async fn prepare(video: Video, payload: &Bytes) -> (moq_mux::Source, impl Stream frames, frames_per_group, } = video; - let mut broadcast = moq_net::broadcast::Info::new().produce(); - let consumer = broadcast.consume(); + let (origin, driver) = moq_net::origin::Producer::new(Default::default()); + let driver = tokio::spawn(moq_net::time::run(driver)); + let mut broadcast = origin.create_broadcast("bench").unwrap(); let mut catalog = moq_mux::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); let mut publishers = Vec::new(); for index in 0..tracks { @@ -85,14 +84,7 @@ async fn prepare(video: Video, payload: &Bytes) -> (moq_mux::Source, impl Stream publishers.push(publisher); } catalog.finish().unwrap(); - let (origin, driver) = moq_net::origin::Producer::new(Default::default()); - let driver = tokio::spawn(moq_net::time::run(driver)); - let dynamic = origin.dynamic_route("", Default::default()).unwrap(); - let requests = tokio::spawn(async move { - while let Ok(request) = dynamic.requested_broadcast().await { - request.accept(consumer.clone()); - } - }); + broadcast.announce(Default::default()).unwrap(); let resolved = origin.consume().request_broadcast("bench", None).await.unwrap(); let source = moq_mux::Source::new(origin.consume(), resolved); let stream = source.catalog::<()>(CatalogFormat::Hang).await.unwrap(); @@ -100,7 +92,6 @@ async fn prepare(video: Video, payload: &Bytes) -> (moq_mux::Source, impl Stream _published: Box::new((broadcast, catalog, publishers)), _origin: origin, driver, - requests, }; (source, stream, fixture) } diff --git a/rs/moq-mux/benches/export.rs b/rs/moq-mux/benches/export.rs index 58383df9172..7c6ad8d0097 100644 --- a/rs/moq-mux/benches/export.rs +++ b/rs/moq-mux/benches/export.rs @@ -14,20 +14,19 @@ struct Fixture { _published: Box, _origin: moq_net::origin::Producer, driver: tokio::task::JoinHandle, - requests: tokio::task::JoinHandle<()>, } impl Drop for Fixture { fn drop(&mut self) { self.driver.abort(); - self.requests.abort(); } } /// A finished broadcast of `tracks` audio tracks with `frames` queued frames each. async fn prepare(tracks: usize, frames: usize) -> (moq_mux::Source, moq_mux::catalog::Consumer, Fixture) { - let mut broadcast = moq_net::broadcast::Info::new().produce(); - let consumer = broadcast.consume(); + let (origin, driver) = moq_net::origin::Producer::new(Default::default()); + let driver = tokio::spawn(moq_net::time::run(driver)); + let mut broadcast = origin.create_broadcast("bench").unwrap(); let mut catalog = moq_mux::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); let stream = moq_mux::catalog::Consumer::from(catalog.consume().unwrap()); let mut publishers = Vec::new(); @@ -56,21 +55,13 @@ async fn prepare(tracks: usize, frames: usize) -> (moq_mux::Source, moq_mux::cat track.finish().unwrap(); } catalog.finish().unwrap(); - let (origin, driver) = moq_net::origin::Producer::new(Default::default()); - let driver = tokio::spawn(moq_net::time::run(driver)); - let dynamic = origin.dynamic_route("", Default::default()).unwrap(); - let requests = tokio::spawn(async move { - while let Ok(request) = dynamic.requested_broadcast().await { - request.accept(consumer.clone()); - } - }); + broadcast.announce(Default::default()).unwrap(); let resolved = origin.consume().request_broadcast("bench", None).await.unwrap(); let source = moq_mux::Source::new(origin.consume(), resolved); let fixture = Fixture { _published: Box::new((broadcast, catalog, publishers)), _origin: origin, driver, - requests, }; (source, stream, fixture) } @@ -176,8 +167,9 @@ async fn prepare_video( size: usize, pattern: &str, ) -> (moq_mux::Source, moq_mux::catalog::Consumer, Fixture) { - let mut broadcast = moq_net::broadcast::Info::new().produce(); - let consumer = broadcast.consume(); + let (origin, driver) = moq_net::origin::Producer::new(Default::default()); + let driver = tokio::spawn(moq_net::time::run(driver)); + let mut broadcast = origin.create_broadcast("bench").unwrap(); let mut catalog = moq_mux::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); let stream = moq_mux::catalog::Consumer::from(catalog.consume().unwrap()); let track = broadcast @@ -220,14 +212,7 @@ async fn prepare_video( } video.finish().unwrap(); catalog.finish().unwrap(); - let (origin, driver) = moq_net::origin::Producer::new(Default::default()); - let driver = tokio::spawn(moq_net::time::run(driver)); - let dynamic = origin.dynamic_route("", Default::default()).unwrap(); - let requests = tokio::spawn(async move { - while let Ok(request) = dynamic.requested_broadcast().await { - request.accept(consumer.clone()); - } - }); + broadcast.announce(Default::default()).unwrap(); let resolved = origin.consume().request_broadcast("bench", None).await.unwrap(); let source = moq_mux::Source::new(origin.consume(), resolved); ( @@ -237,7 +222,6 @@ async fn prepare_video( _published: Box::new((broadcast, catalog, video)), _origin: origin, driver, - requests, }, ) } diff --git a/rs/moq-mux/src/catalog/hang/consumer.rs b/rs/moq-mux/src/catalog/hang/consumer.rs index f7c9ec2f178..513abcd4b7b 100644 --- a/rs/moq-mux/src/catalog/hang/consumer.rs +++ b/rs/moq-mux/src/catalog/hang/consumer.rs @@ -158,33 +158,32 @@ mod test { publish_catalog(published) } - /// [`publish_catalog`] against a *standalone* broadcast (no origin, so its own path is - /// empty) that a dynamic handler serves at `a/pub`, read back through the requesting - /// origin cursor. That is the shape an FFI app takes when it answers an origin request - /// with a broadcast it built itself. + /// [`publish_catalog`] against the answer a dynamic handler prepares and publishes at + /// `a/pub`, read back through the requesting origin cursor. That is the shape an app + /// takes when it answers an origin request. async fn publish_catalog_served(published: Catalog<()>) -> Result> { - let broadcast = moq_net::broadcast::Info::new().produce(); - let track = broadcast - .create_track(hang::Catalog::DEFAULT_NAME, hang::Catalog::default_track_info()) - .expect("catalog track should create"); - let origin = crate::source::produce_origin(); - let dynamic = origin.dynamic_route("", Default::default()).unwrap(); + let dynamic = origin.dynamic("").unwrap(); let requesting = origin.consume().request_broadcast("a/pub", None); - let served = broadcast.consume(); // The handler hands its route back: dropping it would retract the route and end // the broadcast it just served. let handler = tokio::spawn(async move { - dynamic + let prepared = dynamic .requested_broadcast() .await .expect("handler should be asked for the path") - .accept(&served); - dynamic + .prepare(moq_net::Epoch::mint()) + .expect("the answer should prepare"); + let track = prepared + .broadcast() + .create_track(hang::Catalog::DEFAULT_NAME, hang::Catalog::default_track_info()) + .expect("catalog track should create"); + let broadcast = prepared.publish().expect("the answer should publish"); + (dynamic, broadcast, track) }); let resolved = requesting.await; - let _dynamic = handler.await.expect("handler task"); + let (_dynamic, _broadcast, track) = handler.await.expect("handler task"); let subscriber = resolved .expect("the handler served it") .track(hang::Catalog::DEFAULT_NAME) diff --git a/rs/moq-mux/src/container/consumer.rs b/rs/moq-mux/src/container/consumer.rs index 5b2827f27a4..633ba8a1b69 100644 --- a/rs/moq-mux/src/container/consumer.rs +++ b/rs/moq-mux/src/container/consumer.rs @@ -2382,21 +2382,17 @@ mod tests { let epoch = moq_net::Epoch::mint(); let origin = crate::source::produce_origin(); let hops = moq_net::Hops::try_from(vec![moq_net::Hop::new(10).unwrap()]).unwrap(); - let first_route = origin - .dynamic_route( - "live", - moq_net::origin::Route::default() - .with_epoch(epoch.clone()) - .with_hops(hops.clone()) - .with_cost(5), - ) - .unwrap(); - let pending = origin.consume().request_broadcast("live", None); - let first = moq_net::broadcast::Info::new().produce(); + let route = |cost: u64| { + moq_net::origin::Route::default() + .with_epoch(epoch.clone()) + .with_hops(hops.clone()) + .with_cost(cost) + }; + let first = origin.create_broadcast("live").unwrap(); let info = hang::container::track_info(hang::catalog::PRIORITY.video); let first_track = first.create_track("video", info.clone()).unwrap(); - first_route.requested_broadcast().await.unwrap().accept(&first); - let broadcast = pending.await.unwrap(); + first.announce(route(5)).unwrap(); + let broadcast = origin.consume().request_broadcast("live", None).await.unwrap(); let track = broadcast .track("video") .unwrap() @@ -2423,21 +2419,11 @@ mod tests { // while the first route stays up but silent. They carry the same broadcast, so each // holds every group from 1 on. The open group is given up once the track runs a // full budget past it. - let mut routes = Vec::new(); let mut sources = Vec::new(); for sequence in 1..=4 { - let route = origin - .dynamic_route( - "live", - moq_net::origin::Route::default() - .with_epoch(epoch.clone()) - .with_hops(hops.clone()) - .with_cost(5 - sequence), - ) - .unwrap(); - let source = moq_net::broadcast::Info::new().produce(); + let source = origin.create_broadcast("live").unwrap(); let mut track = source.create_track("video", info.clone()).unwrap(); - route.requested_broadcast().await.unwrap().accept(&source); + source.announce(route(5 - sequence)).unwrap(); let mut demand = track.demand(); while demand.wanted_changed().await.unwrap().is_none() {} if sequence == 1 { @@ -2446,7 +2432,6 @@ mod tests { for sequence in 1..=sequence { write_group(&mut track, sequence, &[ts(sequence * 1_000_000)]); } - routes.push(route); sources.push((source, track)); } diff --git a/rs/moq-mux/src/container/fmp4/export_test.rs b/rs/moq-mux/src/container/fmp4/export_test.rs index 58bf7593a87..896ca3e6d80 100644 --- a/rs/moq-mux/src/container/fmp4/export_test.rs +++ b/rs/moq-mux/src/container/fmp4/export_test.rs @@ -1928,12 +1928,12 @@ async fn split_h265_parameter_sets_keep_the_sync_point() { #[tokio::test(start_paused = true)] async fn an_in_band_keyframe_without_parameter_sets_fails() { let mut live = Live::avc3(); - live.track.write(annexb_frame(0, &[IDR])).unwrap(); - live.track.finish().unwrap(); - let mut exporter = crate::container::fmp4::Export::new(live.source().await, live.catalog_stream().await) .with_max_delay(RECORDING_MAX_DELAY); chunk_now(&mut exporter).await.init().expect("init"); + + live.track.write(annexb_frame(0, &[IDR])).unwrap(); + live.track.finish().unwrap(); let err = error_now(&mut exporter).await; assert!( matches!( diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index c58e9f65c05..4932fbf6dc8 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -9532,38 +9532,33 @@ async fn a_follower_built_after_its_route_went_follows_the_return() { assert_eq!(start.elapsed(), Duration::from_secs(4) + follow); } -/// The export's own instance winning the path back from a more specific route that went is no +/// The export's own instance winning the path back from a cheaper route that went is no /// replacement: its end ends the export cleanly, without stitching. #[tokio::test(start_paused = true)] async fn a_follower_takes_its_own_instance_winning_back_for_no_replacement() { let origin = crate::source::produce_origin(); - let epoch = moq_net::Epoch::mint(); - let route = |epoch| moq_net::origin::Route::default().with_epoch(epoch); + let route = |cost: u64| { + moq_net::origin::Route::default() + .with_epoch(moq_net::Epoch::mint()) + .with_cost(cost) + }; - // The export resolves through a prefix claim serving its own instance. - let mut info = moq_net::broadcast::Info::new(); - info.epoch = Some(epoch.clone()); - let mut served = info.produce(); + // The export resolves its own instance. + let mut served = origin.create_broadcast("pool/job").unwrap(); let mut catalog = crate::catalog::Producer::<()>::new(&mut served, Default::default()).unwrap(); - let pool = origin.dynamic_route("pool", route(epoch)).unwrap(); - let consumer = served.consume(); - tokio::spawn(async move { - while let Ok(request) = pool.requested_broadcast().await { - request.accept(consumer.clone()); - } - }); + served.announce(route(5)).unwrap(); let source = crate::source::resolved(&origin, "pool/job").await; let mut follower = super::Follower::new(open_export(source).await.unwrap()).unwrap(); - // A more specific route takes the path for a while, then goes. - let exact = origin.publish("pool/job", route(moq_net::Epoch::mint())).unwrap(); + // A cheaper route takes the path for a while, then goes. + let cheaper = origin.publish("pool/job", route(1)).unwrap(); assert!( tokio::time::timeout(Duration::from_secs(1), follower.next()) .await .is_err(), "still on its own instance" ); - drop(exact); + drop(cheaper); assert!( tokio::time::timeout(Duration::from_secs(1), follower.next()) .await @@ -9656,10 +9651,10 @@ async fn drain_follower(follower: &mut super::Follower) -> (Vec, Option::new(&mut served, Default::default()).unwrap(); let mut track = aac_rendition(&mut served, &mut catalog, "a.aac"); - let pool = origin.dynamic_route("pool", route).unwrap(); - let consumer = served.consume(); - tokio::spawn(async move { - while let Ok(request) = pool.requested_broadcast().await { - request.accept(consumer.clone()); - } - }); + served.announce(route).unwrap(); let resumed = start.elapsed().as_millis() as u64; for ms in (resumed..resumed + 200).step_by(20) { write_aac(&mut track, ms); @@ -9712,7 +9699,7 @@ async fn a_follower_continues_through_a_same_epoch_handoff_after_a_gap() { frames.extend(rest); assert!( matches!(end, Some(Ok(()))), - "the prefix's broadcast ends the export: {end:?}" + "the second route's broadcast ends the export: {end:?}" ); assert_eq!(pes_count(&frames), 20, "every frame of both routes went out"); @@ -9741,10 +9728,10 @@ async fn a_follower_ignores_a_covering_prefix_while_the_exact_route_serves() { let mut follower = super::Follower::new(open_export(source).await.unwrap()).unwrap(); let quiet = Duration::from_secs(1); - let pool = origin.dynamic_route("pool", route(moq_net::Epoch::mint())).unwrap(); + let pool = origin.dynamic("pool").unwrap(); assert!(tokio::time::timeout(quiet, follower.next()).await.is_err(), "a start"); drop(pool); - let pool = origin.dynamic_route("pool", route(moq_net::Epoch::mint())).unwrap(); + let pool = origin.dynamic("pool").unwrap(); assert!(tokio::time::timeout(quiet, follower.next()).await.is_err(), "a restart"); drop(pool); assert!(tokio::time::timeout(quiet, follower.next()).await.is_err(), "an end"); diff --git a/rs/moq-mux/src/source.rs b/rs/moq-mux/src/source.rs index 665c70a441e..f68afe5663a 100644 --- a/rs/moq-mux/src/source.rs +++ b/rs/moq-mux/src/source.rs @@ -444,21 +444,13 @@ pub(crate) async fn resolved(origin: &moq_net::origin::Producer, path: &str) -> Source::new(consumer, broadcast) } -/// Test helper: serve `broadcast` at `test` (and every other path) on a throwaway origin's -/// dynamic handler and return a [`Source`] built from it, so exporter tests that build a -/// local broadcast still resolve references through an origin. The origin is leaked so the -/// broadcast stays reachable for the source's lifetime (harmless in a test binary). +/// Test helper: a [`Source`] for a standalone `broadcast`, resolving its references through +/// a throwaway origin that serves nothing. The origin is leaked so its driver runs for the +/// source's lifetime (harmless in a test binary). #[cfg(test)] pub(crate) async fn announced(broadcast: &moq_net::broadcast::Consumer) -> Source { let origin = produce_origin(); - let dynamic = origin.dynamic_route("", Default::default()).unwrap(); - let served = broadcast.clone(); - tokio::spawn(async move { - while let Ok(request) = dynamic.requested_broadcast().await { - request.accept(served.clone()); - } - }); - let source = resolved(&origin, "test").await; + let source = Source::new(origin.consume(), broadcast.clone()); Box::leak(Box::new(origin)); source } @@ -483,7 +475,7 @@ mod tests { let _catalog = origin.publish("pub", Default::default()).unwrap(); settle().await; let source = resolved(&origin, "pub").await; - let dynamic = origin.dynamic_route("media", Default::default()).unwrap(); + let dynamic = origin.dynamic("media").unwrap(); let binding = source.bind(Some(&Relative::new("media")), None).unwrap(); let request = dynamic.requested_broadcast().await.unwrap(); @@ -498,9 +490,9 @@ mod tests { _ = std::future::ready(()) => {} } - let producer = moq_net::broadcast::Info::default().produce(); - let (first, second, ()) = tokio::join!(binding.broadcast(), binding.broadcast(), async { - request.accept(producer.consume()); + let prepared = request.prepare(moq_net::Epoch::mint()).unwrap(); + let (first, second, producer) = tokio::join!(binding.broadcast(), binding.broadcast(), async { + prepared.publish().unwrap() }); let first = first.unwrap(); assert!(!first.is_closed()); diff --git a/rs/moq-net/Cargo.toml b/rs/moq-net/Cargo.toml index 962e67b9aa2..0cea98dba35 100644 --- a/rs/moq-net/Cargo.toml +++ b/rs/moq-net/Cargo.toml @@ -56,6 +56,9 @@ futures = { workspace = true } # The tests' executor, with simulated time. moq-clock = { path = "../moq-clock", features = ["sim"] } moq-json = { path = "../moq-json" } +# The integration tests that land peer claims through the hidden `fuzz` module run in the +# default merge gate. +moq-net = { path = ".", features = ["fuzz"] } serde_json = { workspace = true } # The benches run on tokio. test-util (tokio::time::pause/advance) is test-only and # is NOT supported on wasm, so it must not leak into the normal dependency feature set. @@ -80,10 +83,11 @@ name = "fuzz-seeds" path = "examples/fuzz-seeds.rs" required-features = ["fuzz"] +# Lands peer claims through the hidden `fuzz` module, as a session does. [[bench]] name = "origin" harness = false -required-features = ["async"] +required-features = ["fuzz"] # Reaches the crate-private session view of an origin through the hidden `fuzz` module. [[bench]] diff --git a/rs/moq-net/MIGRATE.md b/rs/moq-net/MIGRATE.md index c1203ef085d..9dea3a51305 100644 --- a/rs/moq-net/MIGRATE.md +++ b/rs/moq-net/MIGRATE.md @@ -174,7 +174,7 @@ Treat a failed group read as the loss of that group and keep reading the track, Answer each `origin::Request` with `request.prepare(epoch)?` (a fresh `Epoch::mint()` per output run), fill `prepared.broadcast()`, then `prepared.publish()?`, which announces the broadcast at the exact path under that epoch and returns its producer. Only `publish` announces it: `prepared.broadcast().announce(..)` fails with `Error::Closed`. `Dynamic::update` is gone: drop the claim and claim again to change it, and close an answer's producer to end that output. -`Request::accept` is no longer part of the public API and panics on a request from `dynamic`. +`Request::accept`, `Producer::dynamic_route`, and `Dynamic::update_route` are crate-private (#6496): only the built-in sessions answer with a broadcast they already hold. Dropping a `Prepared` unpublished refuses the request, as dropping a `Request` does, and `publish` fails with `Error::Unroutable` once the claim went away. ### Anchors can name a time (#6074) diff --git a/rs/moq-net/benches/origin.rs b/rs/moq-net/benches/origin.rs index a293163e86d..8691745f057 100644 --- a/rs/moq-net/benches/origin.rs +++ b/rs/moq-net/benches/origin.rs @@ -29,7 +29,9 @@ //! Readers draining a swapped track are swept over their count, the groups every //! route delivered, and the routes each reader still holds a copy of. //! -//! Run with `cargo bench -p moq-net --bench origin`. +//! Peer claims land through the hidden `fuzz` module, as a session lands them. +//! +//! Run with `cargo bench -p moq-net --features fuzz --bench origin`. #![expect( clippy::disallowed_methods, @@ -165,7 +167,9 @@ fn bench_withdraw(c: &mut Criterion) { .poll(moq_net::time::Instant::now(), &kio::Waiter::noop()) .unwrap(); let routes: Vec<_> = (0..publishers) - .map(|i| producer.dynamic_route(format!("room/{i}"), peer_route(9, 1)).unwrap()) + .map(|i| { + moq_net::fuzz::dynamic_route(&producer, format!("room/{i}"), peer_route(9, 1)).unwrap() + }) .collect(); let mut cursors: Vec<_> = (0..subscribers).map(|_| producer.consume().announced()).collect(); for cursor in &mut cursors { @@ -219,10 +223,14 @@ fn bench_withdraw_wave(c: &mut Criterion) { .poll(moq_net::time::Instant::now(), &kio::Waiter::noop()) .unwrap(); let near: Vec<_> = (0..broadcasts) - .map(|i| producer.dynamic_route(format!("room/{i}"), route(&[9, 2])).unwrap()) + .map(|i| { + moq_net::fuzz::dynamic_route(&producer, format!("room/{i}"), route(&[9, 2])).unwrap() + }) .collect(); let far: Vec<_> = (0..broadcasts) - .map(|i| producer.dynamic_route(format!("room/{i}"), route(&[9, 3, 4])).unwrap()) + .map(|i| { + moq_net::fuzz::dynamic_route(&producer, format!("room/{i}"), route(&[9, 3, 4])).unwrap() + }) .collect(); let mut cursors: Vec<_> = (0..subscribers).map(|_| producer.consume().announced()).collect(); for cursor in &mut cursors { @@ -266,13 +274,13 @@ fn bench_route_churn(c: &mut Criterion) { group.bench_function(id, |b| { let (producer, _driver) = origin::Producer::new(origin::Config::default()); let _routes: Vec<_> = (0..broadcasts) - .map(|i| producer.dynamic_route(format!("room/{i}"), peer_route(1, 1)).unwrap()) + .map(|i| moq_net::fuzz::dynamic_route(&producer, format!("room/{i}"), peer_route(1, 1)).unwrap()) .collect(); let mut cursors: Vec<_> = (0..subscribers).map(|_| producer.consume().announced()).collect(); for cursor in &mut cursors { while cursor.try_next().is_some() {} } - b.iter(|| drop(producer.dynamic_route("room/0", peer_route(2, 5)).unwrap())); + b.iter(|| drop(moq_net::fuzz::dynamic_route(&producer, "room/0", peer_route(2, 5)).unwrap())); for cursor in &mut cursors { assert!(cursor.try_next().is_none(), "the winner is unchanged"); } @@ -397,9 +405,7 @@ fn bench_announce_duplicate(c: &mut Criterion) { // the entries are distinct routes rather than one re-priced in place. let _routes: Vec<_> = (1..=duplicates) .map(|peer| { - producer - .dynamic_route(PATH, peer_route(peer as u64, INCUMBENT_COST)) - .unwrap() + moq_net::fuzz::dynamic_route(&producer, PATH, peer_route(peer as u64, INCUMBENT_COST)).unwrap() }) .collect(); let mut cursors: Vec = (0..subscribers) @@ -410,9 +416,12 @@ fn bench_announce_duplicate(c: &mut Criterion) { } b.iter(|| { - let handle = producer - .dynamic_route(PATH, peer_route(duplicates as u64 + 1, INCUMBENT_COST - 1)) - .unwrap(); + let handle = moq_net::fuzz::dynamic_route( + &producer, + PATH, + peer_route(duplicates as u64 + 1, INCUMBENT_COST - 1), + ) + .unwrap(); for cursor in &mut cursors { cursor.next().now_or_never().flatten().expect("announce delivered"); } @@ -441,20 +450,18 @@ fn bench_reprice_duplicate(c: &mut Criterion) { let (producer, _driver) = origin::Producer::new(config); let _routes: Vec<_> = (1..=duplicates) .map(|peer| { - producer + let scoped = producer .scope( "", &Patterns::from(Pattern::subtree(&format!("{PATH}/p{}", (peer - 1) % 2)).unwrap()), ) - .unwrap() - .dynamic_route(PATH, peer_route(peer as u64, incumbent_cost)) - .unwrap() + .unwrap(); + moq_net::fuzz::dynamic_route(&scoped, PATH, peer_route(peer as u64, incumbent_cost)).unwrap() }) .collect(); let challenger = duplicates as u64 + 1; - let route = producer - .dynamic_route(PATH, peer_route(challenger, incumbent_cost + 1)) - .unwrap(); + let route = + moq_net::fuzz::dynamic_route(&producer, PATH, peer_route(challenger, incumbent_cost + 1)).unwrap(); let mut cursors: Vec = (0..subscribers) .map(|peer| { producer @@ -473,7 +480,7 @@ fn bench_reprice_duplicate(c: &mut Criterion) { } b.iter(|| { for cost in [incumbent_cost - 1, incumbent_cost + 1] { - route.update_route(peer_route(challenger, cost)).unwrap(); + moq_net::fuzz::update_route(&route, peer_route(challenger, cost)).unwrap(); for cursor in &mut cursors { // The winner switches between epochless routes, which is another instance. let event = cursor.next().now_or_never().flatten().expect("winner changed"); @@ -515,7 +522,7 @@ fn bench_announce_fronts(c: &mut Criterion) { let (producer, mut driver) = origin::Producer::new(origin::Config::default()); let consumer = producer.consume(); // Served but never answered: every request under it parks a front. - let _served = producer.dynamic_route("room", origin::Route::default()).unwrap(); + let _served = producer.dynamic("room").unwrap(); let _requests: Vec<_> = (0..fronts) .map(|i| consumer.request_broadcast(format!("room/{i}"), None)) .collect(); @@ -555,11 +562,7 @@ fn bench_serve_idle(c: &mut Criterion) { let (producer, _driver) = origin::Producer::new(origin::Config::default()); // One route per announced path, exactly as a session lands a peer's. let dynamics: Vec<_> = (0..routes) - .map(|i| { - producer - .dynamic_route(format!(".stats/p{}/node/edge{i}", i % 8), origin::Route::default()) - .unwrap() - }) + .map(|i| producer.dynamic(format!(".stats/p{}/node/edge{i}", i % 8)).unwrap()) .collect(); b.iter(|| { // A fresh waiter per sweep, so every poll pays a real registration: a @@ -714,8 +717,9 @@ fn bench_unroutable_churn(c: &mut Criterion) { group.bench_function(id, |b| { let (producer, mut driver) = origin::Producer::new(origin::Config::default()); let consumer = producer.consume(); - let server = producer.dynamic_route("room", origin::Route::default()).unwrap(); - let unrelated = producer.dynamic_route("other", origin::Route::default()).unwrap(); + let server = moq_net::fuzz::dynamic_route(&producer, "room", origin::Route::default()).unwrap(); + let unrelated = + moq_net::fuzz::dynamic_route(&producer, "other", origin::Route::default()).unwrap(); let mut waiting = Waiting::new( (0..requesters) .map(|i| { @@ -747,7 +751,7 @@ fn bench_unroutable_churn(c: &mut Criterion) { b.iter(|| { for _ in 0..changes { cost = 3 - cost; - changed.update_route(origin::Route::default().with_cost(cost)).unwrap(); + moq_net::fuzz::update_route(changed, origin::Route::default().with_cost(cost)).unwrap(); assert!(covering || !waiting.is_woken(), "unrelated churn wakes no request"); waiting.drain(); driver.poll(now, &waiter).unwrap(); @@ -839,7 +843,7 @@ fn pool(members: usize, paths: usize, cursors: usize) -> Pool { let consumer = producer.consume(); let cursors = (0..cursors).map(|_| consumer.announced()).collect(); let members: Vec<_> = (0..members) - .map(|i| producer.dynamic_route("pool", pool_route(POOL_HOP + i as u64)).unwrap()) + .map(|i| moq_net::fuzz::dynamic_route(&producer, "pool", pool_route(POOL_HOP + i as u64)).unwrap()) .collect(); let waiter = kio::Waiter::noop(); @@ -851,7 +855,7 @@ fn pool(members: usize, paths: usize, cursors: usize) -> Pool { for member in &members { while let Poll::Ready(request) = member.poll_requested_broadcast(&waiter) { let broadcast = broadcast::Info::new().produce(); - request.unwrap().accept(&broadcast); + moq_net::fuzz::accept(request.unwrap(), &broadcast); producers.push(broadcast); } } @@ -898,7 +902,7 @@ fn bench_pool_churn(c: &mut Criterion) { let waiter = kio::Waiter::noop(); let hop = POOL_HOP + members as u64; b.iter(|| { - let joined = pool.producer.dynamic_route("pool", pool_route(hop)).unwrap(); + let joined = moq_net::fuzz::dynamic_route(&pool.producer, "pool", pool_route(hop)).unwrap(); pool.driver.poll(moq_net::time::Instant::now(), &waiter).unwrap(); drop(joined); pool.driver.poll(moq_net::time::Instant::now(), &waiter).unwrap(); @@ -926,7 +930,7 @@ fn bench_pool_cursors(c: &mut Criterion) { let waiter = kio::Waiter::noop(); let hop = POOL_HOP + members as u64; b.iter(|| { - let joined = pool.producer.dynamic_route("pool", pool_route(hop)).unwrap(); + let joined = moq_net::fuzz::dynamic_route(&pool.producer, "pool", pool_route(hop)).unwrap(); pool.driver.poll(moq_net::time::Instant::now(), &waiter).unwrap(); drop(joined); pool.driver.poll(moq_net::time::Instant::now(), &waiter).unwrap(); diff --git a/rs/moq-net/src/fuzz.rs b/rs/moq-net/src/fuzz.rs index 8ab9c80d086..e7ca9a21ea8 100644 --- a/rs/moq-net/src/fuzz.rs +++ b/rs/moq-net/src/fuzz.rs @@ -298,6 +298,25 @@ pub fn excluding(origin: crate::origin::Consumer, peer: crate::Hop) -> crate::or origin.excluding(peer) } +/// Land a claim with a peer's wire route, as a session does, for the origin bench. +pub fn dynamic_route( + origin: &crate::origin::Producer, + prefix: impl crate::AsPath, + route: crate::origin::Route, +) -> Result { + origin.dynamic_route(prefix, route) +} + +/// Replace a peer claim's route in place, as a session does on a route update. +pub fn update_route(dynamic: &crate::origin::Dynamic, route: crate::origin::Route) -> Result<(), crate::Error> { + dynamic.update_route(route) +} + +/// Answer a peer claim's request with the broadcast its session already holds. +pub fn accept(request: crate::origin::Request, broadcast: impl crate::Consume) { + request.accept(broadcast) +} + /// A session's snapshot of its peer's initial announcements, for the synced bench. /// Drop it to land the snapshot. pub struct Snapshot(#[allow(dead_code, reason = "held for its drop, which lands the snapshot")] crate::model::Snapshot); diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 8a9dab3a04f..cddf23defdb 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -2357,8 +2357,7 @@ impl Producer { } /// Land a peer's claim with its wire route metadata. - #[doc(hidden)] - pub fn dynamic_route(&self, prefix: impl AsPath, route: Route) -> Result { + pub(crate) fn dynamic_route(&self, prefix: impl AsPath, route: Route) -> Result { self.dynamic_inner(prefix, route, false) } @@ -5400,8 +5399,7 @@ impl Dynamic { /// The prefix is fixed at announce time: to move a route, drop this and call /// [`Producer::dynamic`] again. Fails with [`Error::Closed`] once the origin's /// [`Driver`] has been dropped. - #[doc(hidden)] - pub fn update_route(&self, route: Route) -> Result<(), Error> { + pub(crate) fn update_route(&self, route: Route) -> Result<(), Error> { self.announcement.update(route) } @@ -5541,6 +5539,14 @@ impl ServeState { /// [`Consumer::request_broadcast`]; [`prepare`](Self::prepare) then [`Prepared::publish`] /// resolves it with a new exact-path broadcast, and [`reject`](Self::reject) resolves it /// with an error. Dropping the request without either rejects it. +/// +/// Only the built-in sessions answer with a broadcast they already hold: +/// +/// ```compile_fail,E0624 +/// fn answer(request: moq_net::origin::Request, broadcast: moq_net::broadcast::Consumer) { +/// request.accept(broadcast); +/// } +/// ``` pub struct Request { // Absolute path that was requested. path: PathOwned, @@ -5577,16 +5583,7 @@ impl Request { } /// Resolve a session-forwarded claim with its peer's already-routed broadcast. - /// - /// # Panics - /// - /// On a request from [`Producer::dynamic`], which answers only through [`Self::prepare`]. - #[doc(hidden)] - pub fn accept(self, broadcast: impl Consume) { - assert!( - self.publisher.is_none(), - "an application claim answers through prepare and publish" - ); + pub(crate) fn accept(self, broadcast: impl Consume) { let broadcast = broadcast.consume(); let _ = ServeState::resolve(&self.home, &self.path, &self.producer, Ok(broadcast)); // `self.producer` drops here, closing the channel; the value is still observable. @@ -5850,8 +5847,7 @@ impl kio::Task for Routing { /// Derive a read view from a handle. /// /// Lets APIs accept either a producer or a consumer (e.g. -/// `Client::with_publisher`, -/// [`Request::accept`]). The blanket `&T` impl means you can +/// `Client::with_publisher`). The blanket `&T` impl means you can /// pass by value (`foo(x)`) to hand off ownership, or by reference (`foo(&x)`) /// to keep it, without spelling out `.consume()`. pub trait Consume { @@ -10612,16 +10608,6 @@ mod tests { assert!(matches!(stale.await, Err(Error::Unroutable))); } - #[moq_clock::test] - #[cfg(feature = "async")] - #[should_panic(expected = "prepare and publish")] - async fn an_application_claim_cannot_accept_an_arbitrary_broadcast() { - let producer = origin(1).produce(); - let claim = producer.dynamic("room").unwrap(); - let _pending = producer.consume().request_broadcast("room/alice", None); - queued(&claim).await.accept(broadcast::Info::new().produce()); - } - /// A refusal from the longest prefix is the answer: neither a costlier advertiser of /// that prefix nor a catch-all is asked instead. #[moq_clock::test] diff --git a/rs/moq-net/tests/announce_allocs.rs b/rs/moq-net/tests/announce_allocs.rs index b41579ff697..1b6e3eb84c4 100644 --- a/rs/moq-net/tests/announce_allocs.rs +++ b/rs/moq-net/tests/announce_allocs.rs @@ -57,7 +57,7 @@ fn peer_route(peer: u64, cost: u64) -> origin::Route { /// again, with `cursors` cursors rooted at `root` watching it. fn churn_allocs(root: &str, cursors: usize) -> usize { let (producer, _driver) = origin::Producer::new(origin::Config::default()); - let _winner = producer.dynamic_route("room/a", peer_route(1, 1)).unwrap(); + let _winner = producer.publish("room/a", peer_route(1, 1)).unwrap(); let all = Patterns::from(Pattern::all()); let mut cursors: Vec<_> = (0..cursors) .map(|_| producer.consume().scope(root, &all).unwrap().announced()) @@ -66,7 +66,7 @@ fn churn_allocs(root: &str, cursors: usize) -> usize { while cursor.try_next().is_some() {} } - let cycle = || drop(producer.dynamic_route("room/a", peer_route(2, 5)).unwrap()); + let cycle = || drop(producer.publish("room/a", peer_route(2, 5)).unwrap()); // Reach the steady state first, with every table grown to what it settles at. for _ in 0..4 { cycle(); diff --git a/rs/moq-net/tests/integration/drained_claim.rs b/rs/moq-net/tests/integration/drained_claim.rs index 84ac5f0ad17..de3ba3d33ee 100644 --- a/rs/moq-net/tests/integration/drained_claim.rs +++ b/rs/moq-net/tests/integration/drained_claim.rs @@ -4,16 +4,16 @@ //! //! Two workers each claim the `pool` prefix with `origin::Producer::dynamic` and //! announce it to relay `R` on their own sessions, the way a transcode pool does. A -//! viewer on `R` reads `pool/p` from worker `A`, which then closes that output and -//! drains its claim. Nothing tells `R` the output closed, so only the front ending -//! once unread lets the next viewer reach worker `B`, which wins the path now, rather -//! than the drained worker. +//! viewer on `R` reads `pool/p` from worker `A`, over the cheaper link, which then +//! closes that output and drains its session. Nothing tells `R` the output closed, so +//! only the front ending once unread lets the next viewer reach worker `B`, which wins +//! the path now, rather than the drained worker. use crate::support; use std::{cell::RefCell, rc::Rc, time::Duration}; -use moq_net::{Hop, Timestamp, Version, broadcast, origin, track}; +use moq_net::{Epoch, Hop, Timestamp, Version, broadcast, goaway::Goaway, origin, track}; use support::harness::{MockConnectOptions, MockPair, connect_mock}; /// A millisecond track: the frames written here carry timestamps. @@ -29,11 +29,12 @@ fn produce_origin(hop: u64) -> origin::Producer { producer } -/// Have `to` pull everything `from` publishes. -async fn link(version: Version, from: &origin::Producer, to: &origin::Producer) -> MockPair { +/// Have `to` pull everything `from` publishes, over a link priced at `cost`. +async fn link(version: Version, from: &origin::Producer, to: &origin::Producer, cost: u64) -> MockPair { let mut options = MockConnectOptions::new(version); options.server_publish = Some(from.consume()); options.client_subscribe = Some(to.clone()); + options.cost = Some(cost); connect_mock(options).await } @@ -45,34 +46,33 @@ async fn settle() { /// A worker claiming `pool`, answering each request with an output whose `video` /// track carries the worker's name. struct Worker { - claim: Rc, + _claim: Rc, /// Each output with its track, which would end once its producer drops. outputs: Rc>>, origin: origin::Producer, } impl Worker { - fn new(hop: u64, name: &'static str, cost: u64) -> Self { + fn new(hop: u64, name: &'static str) -> Self { let origin = produce_origin(hop); - let claim = Rc::new( - origin - .dynamic_route("pool", origin::Route::default().with_cost(cost)) - .unwrap(), - ); + let claim = Rc::new(origin.dynamic("pool").unwrap()); let outputs = Rc::new(RefCell::new(Vec::new())); let (handler, served) = (claim.clone(), outputs.clone()); drop(moq_clock::sim::spawn(async move { while let Ok(request) = handler.requested_broadcast().await { - let output = broadcast::Info::new().produce(); - let track = output.create_track("video", timed()).unwrap(); + let prepared = request.prepare(Epoch::mint()).unwrap(); + let track = prepared.broadcast().create_track("video", timed()).unwrap(); let mut group = track.append_group().unwrap(); group.write_frame(Timestamp::ZERO, name.as_bytes().to_vec()).unwrap(); group.finish().unwrap(); - request.accept(&output); - served.borrow_mut().push((output, track)); + served.borrow_mut().push((prepared.publish().unwrap(), track)); } })); - Self { claim, outputs, origin } + Self { + _claim: claim, + outputs, + origin, + } } /// How many outputs the worker was asked for. @@ -81,8 +81,13 @@ impl Worker { } } -/// Read the first frame of `video` at `pool/p` through `relay`. +/// Read the first frame of `video` at `pool/p` through `relay`. The first subscription +/// through a claim is reset once the answer's exact route arrives, the cold-start reset +/// accepted until announce takeover lands, so the read retries once. async fn watch(relay: &origin::Producer) -> String { + let cold = relay.consume().request_broadcast("pool/p", None).await.unwrap(); + drop(cold.track("video").unwrap().subscribe(None).await); + settle().await; let broadcast = relay.consume().request_broadcast("pool/p", None).await.unwrap(); let mut subscription = broadcast.track("video").unwrap().subscribe(None).await.unwrap(); let mut group = subscription.recv_group().await.unwrap().expect("a group"); @@ -93,19 +98,17 @@ async fn watch(relay: &origin::Producer) -> String { async fn drained_claim(version: &str) { let version: Version = version.parse().unwrap(); let relay = produce_origin(3); - let a = Worker::new(1, "a", 0); - let b = Worker::new(2, "b", 10); - let _a = link(version, &a.origin, &relay).await; - let _b = link(version, &b.origin, &relay).await; + let a = Worker::new(1, "a"); + let b = Worker::new(2, "b"); + let a_link = link(version, &a.origin, &relay, 1).await; + let _b = link(version, &b.origin, &relay, 10).await; settle().await; assert_eq!(watch(&relay).await, "a", "{version}"); - // A closes its output and drains its claim, so B wins the path from here on. + // A closes its output and drains its session, so B wins the path from here on. a.outputs.borrow().iter().for_each(|(output, _)| output.close()); - a.claim - .update_route(a.claim.route().with_cost(origin::Cost::DRAIN)) - .unwrap(); + a_link.server.drain().send(Goaway::new()).unwrap(); // Past the linger, which is as long as the cache window, nothing reads the relay's // front for the path. moq_clock::sleep(moq_net::cache::DEFAULT_EXPIRY * 2).await; diff --git a/rs/moq-net/tests/integration/ietf_conformance.rs b/rs/moq-net/tests/integration/ietf_conformance.rs index 163ae649fe2..07c18ad865e 100644 --- a/rs/moq-net/tests/integration/ietf_conformance.rs +++ b/rs/moq-net/tests/integration/ietf_conformance.rs @@ -1,6 +1,9 @@ #![cfg(feature = "async")] //! Moxygen announces one prefix, then serves its sixteen-field test namespaces. +//! +//! It answers each namespace without announcing its exact path, as only a session's peer +//! claim does, so the claim lands through the hidden `fuzz` module. use crate::support::harness::{MockConnectOptions, connect_mock, spawn}; use moq_net::{Hop, Version, broadcast, origin}; @@ -21,7 +24,7 @@ async fn routing(version: &str, cluster: bool) { let publisher = origin(1); let relay = origin(2); let reader = origin(4); - let prefix = Arc::new(publisher.dynamic_route(PREFIX, Default::default()).unwrap()); + let prefix = Arc::new(moq_net::fuzz::dynamic_route(&publisher, PREFIX, Default::default()).unwrap()); let broadcast = broadcast::Info::new().produce(); let track = broadcast.create_track("test", moq_net::track::Info::default()).unwrap(); let serving = prefix.clone(); @@ -29,7 +32,7 @@ async fn routing(version: &str, cluster: bool) { moq_clock::sim::spawn(async move { let request = serving.requested_broadcast().await.unwrap(); assert_eq!(request.path().as_str(), NAMESPACE); - request.accept(answer); + moq_net::fuzz::accept(request, answer); }); let mut options = MockConnectOptions::new(version); diff --git a/rs/moq-net/tests/integration/mesh_withdraw.rs b/rs/moq-net/tests/integration/mesh_withdraw.rs index 0c2e6605c21..c27e0cfb016 100644 --- a/rs/moq-net/tests/integration/mesh_withdraw.rs +++ b/rs/moq-net/tests/integration/mesh_withdraw.rs @@ -657,9 +657,9 @@ async fn hop_limit_bounds_receive_and_send() { .unwrap(), ) }; - let _long = source.dynamic_route("long", route(2)).unwrap(); - let _boundary = source.dynamic_route("boundary", route(1)).unwrap(); - let short = source.dynamic_route("short", route(0)).unwrap(); + let _long = source.publish("long", route(2)).unwrap(); + let _boundary = source.publish("boundary", route(1)).unwrap(); + let short = source.publish("short", route(0)).unwrap(); moq_clock::sim::idle().await; let received = drain(&mut received); assert!(!received.contains_key("long"), "{name}: accepted over-limit path"); @@ -678,12 +678,12 @@ async fn hop_limit_bounds_receive_and_send() { Some(&Kind::End), "{name}" ); - let _replacement = source.dynamic_route("short", route(1)).unwrap(); + let _replacement = source.publish("short", route(1)).unwrap(); moq_clock::sim::idle().await; assert!(drain(&mut sent).is_empty(), "{name}: forwarded over-limit replacement"); continue; } - short.update_route(route(1)).unwrap(); + short.announce(route(1)).unwrap(); moq_clock::sim::idle().await; let ended = drain(&mut sent); assert_eq!( @@ -738,9 +738,9 @@ async fn hop_limit_falls_back_to_a_forwardable_route() { .with_cost(origin::Cost::new(cost)) }; // Three hops at the relay once the sender is appended: in the cap, but full. - let _cheap = far.dynamic_route("room", route(100, 2, 1)).unwrap(); + let _cheap = far.publish("room", route(100, 2, 1)).unwrap(); // Two hops at the relay, so it has room for the relay's own. - let pricey = near.dynamic_route("room", route(200, 1, 50)).unwrap(); + let pricey = near.publish("room", route(200, 1, 50)).unwrap(); moq_clock::sim::idle().await; let started = drain(&mut sent); assert_eq!( diff --git a/rs/moq-net/tests/integration/restart.rs b/rs/moq-net/tests/integration/restart.rs index c5b24d1d1a2..cec0496f1a9 100644 --- a/rs/moq-net/tests/integration/restart.rs +++ b/rs/moq-net/tests/integration/restart.rs @@ -172,7 +172,8 @@ relay_chain_tests! { } /// A worker claiming `pool` without an epoch, answering each request with an output -/// whose `video` track carries the worker's name. +/// whose `video` track carries the worker's name. The claim is a peer's, landed through +/// the hidden `fuzz` module as a session lands it, so its answers carry no exact route. struct Worker { origin: origin::Producer, _claim: Rc, @@ -182,7 +183,7 @@ struct Worker { impl Worker { fn new(hop: u64) -> Self { let origin = produce_origin(hop); - let claim = Rc::new(origin.dynamic_route("pool", origin::Route::default()).unwrap()); + let claim = Rc::new(moq_net::fuzz::dynamic_route(&origin, "pool", origin::Route::default()).unwrap()); let outputs = Rc::new(RefCell::new(Vec::new())); let (handler, served) = (claim.clone(), outputs.clone()); drop(moq_clock::sim::spawn(async move { @@ -190,7 +191,7 @@ impl Worker { let output = broadcast::Info::new().produce(); let track = output.create_track("video", timed()).unwrap(); write(&track, &format!("{hop}")); - request.accept(&output); + moq_net::fuzz::accept(request, &output); served.borrow_mut().push((output, track)); } })); @@ -255,7 +256,9 @@ async fn prefix_pool_lite07() { } /// A claim at `live` whose handler answers each request with a fresh output, kept so -/// the test can write into it: `outputs[n]` answered the `n`th request. +/// the test can write into it: `outputs[n]` answered the `n`th request. Only a peer's +/// claim carries an epoch or re-prices, so it lands through the hidden `fuzz` module as +/// a session lands it. struct Claim { dynamic: Rc, outputs: Rc>>, @@ -265,14 +268,14 @@ impl Claim { fn new(origin: &origin::Producer, epoch: Option) -> Self { let mut route = origin::Route::default(); route.epoch = epoch; - let dynamic = Rc::new(origin.dynamic_route("live", route).unwrap()); + let dynamic = Rc::new(moq_net::fuzz::dynamic_route(origin, "live", route).unwrap()); let outputs = Rc::new(RefCell::new(Vec::new())); let (handler, served) = (dynamic.clone(), outputs.clone()); drop(moq_clock::sim::spawn(async move { while let Ok(request) = handler.requested_broadcast().await { let output = broadcast::Info::new().produce(); let track = output.create_track("video", timed()).unwrap(); - request.accept(&output); + moq_net::fuzz::accept(request, &output); served.borrow_mut().push((output, track)); } })); @@ -292,7 +295,11 @@ impl Claim { fn update_epoch(&self, epoch: Option) { let mut route = self.dynamic.route(); route.epoch = epoch; - self.dynamic.update_route(route).unwrap(); + self.update_route(route); + } + + fn update_route(&self, route: origin::Route) { + moq_net::fuzz::update_route(&self.dynamic, route).unwrap(); } } @@ -320,7 +327,7 @@ async fn dynamic_epoch(version: &str, from: Option, to: Option) { assert_eq!(read(&mut sticky).await, "old:0", "{version}"); // A re-price at the same epoch is an update at most: every handle survives. - claim.dynamic.update_route(claim.dynamic.route().with_cost(5)).unwrap(); + claim.update_route(claim.dynamic.route().with_cost(5)); while let Ok(event) = moq_clock::timeout(Duration::from_secs(1), announced.next()).await { assert!( matches!(event, Some(announce::Event::Update(_))), @@ -401,12 +408,9 @@ async fn dynamic_epoch_second_route_keeps_winning() { let (a, _) = epochs(); let origin = produce_origin(1); let first = Claim::new(&origin, Some(a.clone())); - first.dynamic.update_route(first.dynamic.route().with_cost(1)).unwrap(); + first.update_route(first.dynamic.route().with_cost(1)); let second = Claim::new(&origin, Some(a.clone())); - second - .dynamic - .update_route(second.dynamic.route().with_cost(5)) - .unwrap(); + second.update_route(second.dynamic.route().with_cost(5)); let mut announced = origin.consume().announced(); match next_event(&mut announced).await { announce::Event::Start(announce) => assert_eq!(announce.route.epoch.as_ref(), Some(&a)), diff --git a/rs/moq-net/tests/integration/wide_hop.rs b/rs/moq-net/tests/integration/wide_hop.rs index c1d2872dea2..ccb7e4ac848 100644 --- a/rs/moq-net/tests/integration/wide_hop.rs +++ b/rs/moq-net/tests/integration/wide_hop.rs @@ -28,7 +28,7 @@ async fn widest_hop_crosses_each_varint_form() { let upstream: Version = "moq-transport-18".parse().unwrap(); let _upstream = peer(upstream, &source, &relay).await; let _downstream = peer(downstream.parse().unwrap(), &relay, &sink).await; - let _route = source.dynamic_route("room", origin::Route::default()).unwrap(); + let _route = source.dynamic("room").unwrap(); moq_clock::sim::idle().await; let ids = |route: origin::Route| route.hops.iter().map(|hop| hop.id()).collect::>();