diff --git a/cpp/moq/include/moq/moq.hpp b/cpp/moq/include/moq/moq.hpp index 58dc5805260..7ca45691861 100644 --- a/cpp/moq/include/moq/moq.hpp +++ b/cpp/moq/include/moq/moq.hpp @@ -4,7 +4,6 @@ #include -#include #include #include #include @@ -62,18 +61,6 @@ inline void shutdown() noexcept { ::uniffi::shutdown_async_dispatcher(); } -// The wall-clock time an epoch encodes, to the millisecond. Errors unless it is a UUIDv7. -// Millisecond ticks hold every 48-bit UUIDv7 time, where system_clock's nanoseconds overflow in 2262. -inline expected> -epoch_time(const std::string &epoch) { - auto ms = epoch_time_ms(epoch); - if (!ms) { - return unexpected(std::move(ms).error()); - } - return std::chrono::time_point( - std::chrono::milliseconds(static_cast(*ms))); -} - #ifdef MOQ_COROUTINES namespace detail { diff --git a/cpp/moq/test/probe.cpp b/cpp/moq/test/probe.cpp index 7999228f3cb..aedf066fe98 100644 --- a/cpp/moq/test/probe.cpp +++ b/cpp/moq/test/probe.cpp @@ -191,13 +191,6 @@ int main() { // It prints the message Rust's Display gives. CHECK(fingerprints.error().to_string() == "bind: not listening; call listen() first"); - // An epoch's time keeps every 48-bit millisecond, past where system_clock's nanoseconds wrap. - CHECK(ok(moq::epoch_time("018bcfe5-6800-7000-8000-000000000000"), "epoch_time").time_since_epoch() == - std::chrono::milliseconds(1700000000000)); - CHECK(ok(moq::epoch_time("ffffffff-ffff-7000-8000-000000000000"), "epoch_time").time_since_epoch() == - std::chrono::milliseconds(0xffffffffffff)); - CHECK(!moq::epoch_time("not-an-epoch")); - // Publisher: a server whose origin serves one broadcast with one track. auto origin = moq::OriginProducer::init({}); auto broadcast = ok(origin->create_broadcast("probe"), "create_broadcast"); diff --git a/dart/moq/lib/src/epoch.dart b/dart/moq/lib/src/epoch.dart index 7479a523e99..c0db608e414 100644 --- a/dart/moq/lib/src/epoch.dart +++ b/dart/moq/lib/src/epoch.dart @@ -5,10 +5,3 @@ import 'package:moq_ffi/moq_ffi.dart'; /// `AnnounceRoute.epoch`, so viewers see a restart as a new broadcast instead of a /// stalled one. String mintEpoch() => moqMintEpoch(); - -/// The UTC wall-clock time [epoch] encodes, to the millisecond. Throws unless -/// it is a lowercase hyphenated UUIDv7. -DateTime epochTime(String epoch) => DateTime.fromMillisecondsSinceEpoch( - moqEpochTimeMs(epoch: epoch), - isUtc: true, -); diff --git a/dart/moq/test/moq_test.dart b/dart/moq/test/moq_test.dart index c58a9481f52..c2aef3256e8 100644 --- a/dart/moq/test/moq_test.dart +++ b/dart/moq/test/moq_test.dart @@ -461,10 +461,6 @@ void main() { final broadcast = origin.createBroadcast(path: 'epoch'); final first = mintEpoch(); - expect( - epochTime('018bcfe5-6800-7000-8000-000000000000'), - DateTime.utc(2023, 11, 14, 22, 13, 20), - ); broadcast.announce(route: AnnounceRoute(epoch: first)); final start = await nextRoute(announced) as AnnounceEventStart; expect(start.announce.route.epoch, first); @@ -474,7 +470,10 @@ void main() { final restart = await nextRoute(announced) as AnnounceEventRestart; expect(restart.announce.route.epoch, second); - expect(() => epochTime('not-an-epoch'), throwsA(isA())); + expect( + () => broadcast.announce(route: AnnounceRoute(epoch: 'not-an-epoch')), + throwsA(isA()), + ); announced.cancel(); announced.dispose(); }); diff --git a/dart/moq_ffi/lib/src/moq.dart b/dart/moq_ffi/lib/src/moq.dart index c8a13a36090..4962c2744e8 100644 --- a/dart/moq_ffi/lib/src/moq.dart +++ b/dart/moq_ffi/lib/src/moq.dart @@ -8,7 +8,9 @@ import "dart:ffi"; import "dart:io" show Platform, File, Directory; import "dart:isolate"; import "dart:typed_data"; + import "package:ffi/ffi.dart"; + import "uniffi_runtime.dart"; export "uniffi_runtime.dart"; @@ -16421,17 +16423,6 @@ void moqLogLevel({required String level}) { }, moqExceptionErrorHandler); } -int moqEpochTimeMs({required String epoch}) { - return rustCallWithLifter( - (status) => uniffi_moq_ffi_fn_func_moq_epoch_time_ms( - FfiConverterString.lower(epoch), - status, - ), - FfiConverterUInt64.lift, - moqExceptionErrorHandler, - ); -} - String moqMintEpoch() { return rustCallWithLifter( (status) => uniffi_moq_ffi_fn_func_moq_mint_epoch(status), @@ -19115,14 +19106,6 @@ external void uniffi_moq_ffi_fn_func_moq_log_level( Pointer uniffiStatus, ); -@Native)>( - assetId: _uniffiAssetId, -) -external int uniffi_moq_ffi_fn_func_moq_epoch_time_ms( - RustBuffer epoch, - Pointer uniffiStatus, -); - @Native)>(assetId: _uniffiAssetId) external RustBuffer uniffi_moq_ffi_fn_func_moq_mint_epoch( Pointer uniffiStatus, @@ -19470,9 +19453,6 @@ external void ffi_moq_ffi_rust_future_complete_void( @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_func_moq_log_level(); -@Native(assetId: _uniffiAssetId) -external int uniffi_moq_ffi_checksum_func_moq_epoch_time_ms(); - @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_func_moq_mint_epoch(); @@ -20179,9 +20159,6 @@ void _checkApiChecksums() { if (uniffi_moq_ffi_checksum_func_moq_log_level() != 24625) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } - if (uniffi_moq_ffi_checksum_func_moq_epoch_time_ms() != 13125) { - throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); - } if (uniffi_moq_ffi_checksum_func_moq_mint_epoch() != 34550) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } diff --git a/go/wrapper/moq_test.go b/go/wrapper/moq_test.go index b7410b44a3d..1b39e8a8e52 100644 --- a/go/wrapper/moq_test.go +++ b/go/wrapper/moq_test.go @@ -1541,9 +1541,6 @@ func TestAnnouncedEpochRestarts(t *testing.T) { t.Fatal(err) } first := moq.MintEpoch() - if at, err := moq.EpochTime("018bcfe5-6800-7000-8000-000000000000"); err != nil || !at.Equal(time.UnixMilli(1_700_000_000_000)) { - t.Fatalf("EpochTime = %v, %v; want 2023-11-14T22:13:20Z", at, err) - } if err := broadcast.Announce(moq.AnnounceRoute{Epoch: &first}); err != nil { t.Fatal(err) } @@ -1560,8 +1557,9 @@ func TestAnnouncedEpochRestarts(t *testing.T) { t.Fatalf("restart = %+v, want epoch %s", restart, second) } - if _, err := moq.EpochTime("not-an-epoch"); !errors.Is(err, moq.ErrInvalidEpoch) { - t.Fatalf("EpochTime(bad) = %v, want ErrInvalidEpoch", err) + bad := "not-an-epoch" + if err := broadcast.Announce(moq.AnnounceRoute{Epoch: &bad}); !errors.Is(err, moq.ErrInvalidEpoch) { + t.Fatalf("Announce(bad epoch) = %v, want ErrInvalidEpoch", err) } } diff --git a/go/wrapper/types.go b/go/wrapper/types.go index a96dabf05d2..a8d36371723 100644 --- a/go/wrapper/types.go +++ b/go/wrapper/types.go @@ -1,8 +1,6 @@ package moq import ( - "time" - ffi "moq.dev/moq-ffi/moq" ) @@ -206,16 +204,6 @@ func MintEpoch() string { return ffi.MoqMintEpoch() } -// EpochTime is the wall-clock time an epoch encodes, to the millisecond. It -// returns ErrInvalidEpoch unless epoch is a lowercase hyphenated UUIDv7. -func EpochTime(epoch string) (time.Time, error) { - ms, err := ffi.MoqEpochTimeMs(epoch) - if err != nil { - return time.Time{}, err - } - return time.UnixMilli(int64(ms)), nil -} - // Protocol registries and recognized kinds carried by ProtocolError. const ( // ErrorScopeSession identifies a session close code. diff --git a/js/net/MIGRATE.md b/js/net/MIGRATE.md index 2bdc084a66e..15ff507eb2d 100644 --- a/js/net/MIGRATE.md +++ b/js/net/MIGRATE.md @@ -4,6 +4,10 @@ How to update your code for each breaking change, newest first. ## Unreleased +### `Varint.sizeLeadingOnes` is gone + +Take the length of `Varint.encodeLeadingOnes(v)`. + ### Published names refuse navigation segments Broadcast creation and route publication now refuse literal `.` and `..` segments, including segments in the producer's root. diff --git a/js/net/api/index.api.md b/js/net/api/index.api.md index f3c73cf771b..1afd5d29b42 100644 --- a/js/net/api/index.api.md +++ b/js/net/api/index.api.md @@ -21,7 +21,6 @@ function accept(input: AcceptProps): Promise; interface AcceptProps { consume?: Producer_5; - extensions?: Extensions.Extensions; implementation?: string; publish?: Consumer_5; transport: WebTransport; @@ -36,15 +35,6 @@ class Allocator implements Handle_2 { static unlimited(): Allocator; } -const ALPN: { - readonly DRAFT_14: "moq-00"; - readonly DRAFT_15: "moqt-15"; - readonly DRAFT_16: "moqt-16"; - readonly DRAFT_17: "moqt-17"; - readonly DRAFT_18: "moqt-18"; - readonly DRAFT_22: "moqt-22"; -}; - declare namespace Announce { export { Announce_2 as Announce, @@ -86,12 +76,6 @@ interface Auth_2 { } declare namespace Auth_3 { - export { - - } -} - -declare namespace Auth_4 { export { Duplicate, Handle, @@ -142,31 +126,13 @@ interface CertificateHash { function certificateHash(cert: string | BufferSource): Promise>; -class ClientSetup { - constructor(input: { - versions: number[]; - parameters?: SetupOptions; - }); - static decode(r: Reader, version: IetfVersion): Promise; - encode(w: Writer, version: IetfVersion): Promise; - static id: number; - parameters: SetupOptions; - versions: number[]; -} - -declare namespace Cluster { - export { - - } -} - function connect(input: ConnectProps): Promise; export class Connection { constructor(props?: ConnectionProps); static readonly accept: typeof accept; announced(scope?: Path_2.Pattern, options?: Announce_3.Options): Announce_3.Consumer; - readonly auth: Getter; + readonly auth: Getter; readonly bandwidth: Getter; static readonly certificateHash: typeof certificateHash; close(abort?: Error): void; @@ -190,7 +156,7 @@ export namespace Connection { export type CertificateHash = CertificateHash; export type ConnectProps = ConnectProps; export type Established = Established; - export type Extensions = Extensions_2; + export type Extensions = Extensions; export type Goaway = GoawayProps; export type Probe = Probe; export type Props = ConnectionProps; @@ -219,9 +185,8 @@ interface ConnectionProps { } interface ConnectProps { - auth?: Auth_4.Handle; + auth?: Auth_3.Handle; consume?: Producer_5; - extensions?: Extensions.Extensions; implementation?: string; publish?: Consumer_5; signal?: AbortSignal; @@ -299,22 +264,6 @@ const Cost: { zero: bigint; }; -class Cursor { - constructor(buffer: Uint8Array, version: StreamVersion); - bool(): boolean; - exact(size: number, decode: (c: Cursor) => T): T; - get offset(): number; - read(size: number): Uint8Array; - get remaining(): number; - string(): string; - u16(): number; - u53(): number; - u62(): bigint; - u8(): number; - varint(): U64; - readonly version: StreamVersion; -} - interface Datagram { payload: Uint8Array; sequence: number; @@ -335,13 +284,6 @@ class Demand { get wanted(): Getter; } -class Demand_2 { - get closed(): GetPromise; - readonly sequence: number; - unused(): Promise; - get used(): Getter; -} - class Duplicate extends Error { constructor(message?: string); } @@ -369,7 +311,6 @@ declare namespace Epoch { export { parse, mint, - time, Valid_2 as Valid } } @@ -378,17 +319,10 @@ declare namespace Epoch_2 { export { parse, mint, - time, Valid_2 as Valid } } -declare namespace Epoch_3 { - export { - - } -} - declare namespace Error_2 { export { ConnectTimeout, @@ -409,7 +343,7 @@ declare namespace Error_2 { interface Established { abort(): void; announced(scope?: Path_2.Pattern, options?: Announce_3.Options): Announce_3.Consumer; - readonly auth: Auth_4.Handle; + readonly auth: Auth_3.Handle; close(): Promise; closed: Promise; readonly peerImplementation: Promise; @@ -424,16 +358,7 @@ type Event_2 = { kind: "start" | "update" | "restart" | "end"; } & Announce_2; -declare namespace Extensions { - export { - offered, - intoSetup, - Extensions_2 as Extensions, - Offered - } -} - -interface Extensions_2 { +interface Extensions { auth?: boolean; epoch?: boolean; solicit?: boolean; @@ -476,7 +401,6 @@ declare namespace Group { Frame, ReadOptions, Info_2 as Info, - Demand_2 as Demand, Producer_2 as Producer, Consumer_2 as Consumer } @@ -510,27 +434,6 @@ type Hop = z.infer; const HopSchema: z.core.$ZodBranded, "Hop", "out">; -declare namespace Ietf { - export { - Auth_3 as Auth, - Cluster, - Epoch_3 as Epoch, - Setup, - ClientSetup, - ServerSetup, - initialMaxRequestId, - SetupOption, - REQUEST_WINDOW, - SetupOptions, - versionName, - Version, - ALPN, - IetfVersion - } -} - -type IetfVersion = typeof Version.DRAFT_14 | typeof Version.DRAFT_15 | typeof Version.DRAFT_16 | typeof Version.DRAFT_17 | typeof Version.DRAFT_18 | typeof Version.DRAFT_22; - export const IMPLEMENTATION: string; interface Info { @@ -547,10 +450,6 @@ interface Info_2 { function infoDefaults(info?: Partial): Info; -function initialMaxRequestId(peerClient: boolean, window?: bigint): bigint; - -function intoSetup(params: Ietf.SetupOptions, extensions: Offered, version: Ietf.IetfVersion): void; - function isAnonymous(route: Route): boolean; function isWebTransportSupported(): boolean; @@ -585,17 +484,6 @@ class NotFound extends Stream { }); } -type Offered = Required; - -function offered(extensions?: Extensions_2): Offered; - -interface OpenOptions { - sendOrder?: number; - timeout?: number | null; - version: StreamVersion; - waitUntilAvailable?: boolean; -} - interface Options { hidden?: boolean; } @@ -758,7 +646,6 @@ class Producer_2 { close(abort?: Error): void; get closed(): GetPromise; consume(): Consumer_2; - demand(): Demand_2; get isClosed(): boolean; readonly sequence: number; writeFrame(frame: Frame): void; @@ -810,29 +697,6 @@ class ProtocolViolation extends Error { readonly code: SessionCode; } -class Reader { - constructor(stream: ReadableStream, buffer: Uint8Array | undefined, version: StreamVersion); - constructor(stream: undefined, buffer: Uint8Array, version: StreamVersion); - bool(): Promise; - get closed(): Promise; - decode(decode: (c: Cursor) => T): Promise; - decodeMaybe(decode: (c: Cursor) => T): Promise; - discard(): Promise; - done(): Promise; - peekU62(): Promise; - read(size: number): Promise; - readAll(): Promise; - stop(reason: unknown): void; - string(): Promise; - tryDecode>(decode: (c: Cursor) => T): T | undefined; - u16(): Promise; - u53(): Promise; - u62(): Promise; - u8(): Promise; - varint(): Promise; - version: StreamVersion; -} - interface ReadOptions { from?: number; } @@ -875,8 +739,6 @@ class Request_3 { reject(err: Error): void; } -const REQUEST_WINDOW = 240000n; - class Requesting { readonly active: Getter; close(): void; @@ -916,18 +778,6 @@ const Route: { }): Route; }; -class ServerSetup { - constructor(input: { - version: number; - parameters?: SetupOptions; - }); - static decode(r: Reader, version: IetfVersion): Promise; - encode(w: Writer, version: IetfVersion): Promise; - static id: number; - parameters: SetupOptions; - version: number; -} - class Session extends Error { constructor(code: SessionCode, options?: { cause?: unknown; @@ -963,54 +813,10 @@ export const SessionCode: ((code: number) => SessionCode) & { readonly TooManyRequestUpdates: SessionCode; }; -class Setup { - constructor(input?: { - parameters?: SetupOptions; - }); - static decode(r: Reader, version: IetfVersion): Promise; - encode(w: Writer, version: IetfVersion): Promise; - static id: number; - parameters: SetupOptions; -} - -const SetupOption: { - readonly Path: 1n; - readonly MaxRequestId: 2n; - readonly AuthorizationToken: 3n; - readonly MaxAuthTokenCacheSize: 4n; - readonly Authority: 5n; - readonly Implementation: 7n; - readonly HopId: 0x40b54n; - readonly RelayCost: 0x40b56n; - readonly Solicit: 0x40b5an; - readonly Hidden: 0x40b5cn; - readonly MaxAge: 0x915c4n; - readonly Auth: 0x40b70n; - readonly ActiveCount: 0x40b64n; - readonly Epoch: 0x40b68n; -}; - -class SetupOptions { - constructor(); - bytes: Map; - static decode(r: Reader, version: IetfVersion): Promise; - encode(w: Writer, version: IetfVersion): Promise; - getBytes(id: bigint): Uint8Array | undefined; - getVarint(id: bigint): bigint | undefined; - removeBytes(id: bigint): boolean; - removeVarint(id: bigint): boolean; - setBytes(id: bigint, value: Uint8Array): void; - setVarint(id: bigint, value: bigint): void; - get size(): number; - vars: Map; -} - export { Signals } function size(v: number): number; -function sizeLeadingOnes(v: number | bigint): number; - class Staged { [Symbol.dispose](): void; commit(): Producer; @@ -1064,8 +870,6 @@ interface StreamOptions { reason?: string; } -type StreamVersion = IetfVersion | Version_2; - function stripPrefix(prefix: Valid, path: Valid): Valid | null; class Subscriber { @@ -1108,8 +912,6 @@ declare namespace Time { } } -function time(epoch: Valid_2): Date; - export interface Timed { at?: Timestamp; value: T; @@ -1215,28 +1017,8 @@ type TracksEvent = { type Transport = "webtransport" | "websocket" | "webrtc"; -interface TryOpenOptions extends OpenOptions { - cancel: Promise; -} - function tryResolve(base: Valid, rel: Relative): Valid | undefined; -class U64 { - constructor(hi: number, lo: number); - add(delta: number): U64; - compare(other: U64): number; - equals(other: U64): boolean; - static fromBigInt(v: bigint): U64; - static fromNumber(v: number): U64; - readonly hi: number; - readonly lo: number; - static readonly MAX: U64; - toBigInt(): bigint; - toNumber(): number; - toString(): string; - static readonly ZERO: U64; -} - class Unsupported extends Error { constructor(message?: string); readonly code: SessionCode; @@ -1252,7 +1034,6 @@ type Valid_2 = string & { declare namespace Varint { export { - sizeLeadingOnes, encodeLeadingOnesTo, encodeLeadingOnes, decodeLeadingOnes, @@ -1268,32 +1049,6 @@ declare namespace Varint { } } -const Version: { - readonly DRAFT_07: 4278190087; - readonly DRAFT_14: 4278190094; - readonly DRAFT_15: 4278190095; - readonly DRAFT_16: 4278190096; - readonly DRAFT_17: 4278190097; - readonly DRAFT_18: 4278190098; - readonly DRAFT_22: 4278190102; -}; - -type Version = (typeof Version)[keyof typeof Version]; - -const Version_2: { - readonly DRAFT_01: 4279086337; - readonly DRAFT_02: 4279086338; - readonly DRAFT_03: 4279086339; - readonly DRAFT_04: 4279086340; - readonly DRAFT_05: 4279086341; - readonly DRAFT_06: 4279086342; - readonly DRAFT_07: 4279086343; -}; - -type Version_2 = (typeof Version_2)[keyof typeof Version_2]; - -function versionName(v: Version): string; - interface WebSocketProps { delay?: Milli; enabled?: boolean; @@ -1305,26 +1060,5 @@ interface WebTransportProps extends Omit, version: StreamVersion); - bool(v: boolean): Promise; - close(): void; - get closed(): Promise; - static collect(version: StreamVersion, f: (w: Writer) => Promise): Promise; - i32(v: number): Promise; - static open(quic: WebTransport, options: OpenOptions): Promise; - reset(reason: unknown): void; - setPriority(sendOrder: number): void; - string(str: string): Promise; - static tryOpen(quic: WebTransport, options: TryOpenOptions): Promise; - u16(v: number): Promise; - u53(v: number): Promise; - u62(v: bigint): Promise; - u8(v: number): Promise; - varint(v: U64): Promise; - version: StreamVersion; - write(v: Uint8Array): Promise; -} - ``` diff --git a/js/net/api/zod.api.md b/js/net/api/zod.api.md index 953b970143b..fbc1c617171 100644 --- a/js/net/api/zod.api.md +++ b/js/net/api/zod.api.md @@ -57,13 +57,6 @@ class Demand { get wanted(): Getter; } -class Demand_2 { - get closed(): GetPromise; - readonly sequence: number; - unused(): Promise; - get used(): Getter; -} - interface FetchGroupOptions { maxDelay?: Milli; priority?: number; @@ -82,7 +75,6 @@ declare namespace group { Frame, ReadOptions, Info_2 as Info, - Demand_2 as Demand, Producer_2 as Producer, Consumer } @@ -167,7 +159,6 @@ class Producer_2 { close(abort?: Error): void; get closed(): GetPromise; consume(): Consumer; - demand(): Demand_2; get isClosed(): boolean; readonly sequence: number; writeFrame(frame: Frame): void; diff --git a/js/net/src/connection/accept.ts b/js/net/src/connection/accept.ts index 67e1c554ea7..f93fb497182 100644 --- a/js/net/src/connection/accept.ts +++ b/js/net/src/connection/accept.ts @@ -31,7 +31,11 @@ export interface AcceptProps { */ consume?: OriginProducer; - /** The moq-transport extensions to offer, each on unless set to `false`. */ + /** + * The moq-transport extensions to offer, each on unless set to `false`. + * + * @internal Tests only; nothing outside the package consumes it. + */ extensions?: Extensions.Extensions; /** The implementation this side advertises in SETUP; see the `implementation` connect option. */ diff --git a/js/net/src/connection/connect.ts b/js/net/src/connection/connect.ts index 9d16f8b33cb..4da8769cd11 100644 --- a/js/net/src/connection/connect.ts +++ b/js/net/src/connection/connect.ts @@ -215,7 +215,11 @@ export interface ConnectProps { */ consume?: OriginProducer; - /** The moq-transport extensions to offer, each on unless set to `false`. */ + /** + * The moq-transport extensions to offer, each on unless set to `false`. + * + * @internal Tests only; nothing outside the package consumes it. + */ extensions?: Extensions.Extensions; /** diff --git a/js/net/src/epoch.test.ts b/js/net/src/epoch.test.ts index c6b002642a6..b4ab9804375 100644 --- a/js/net/src/epoch.test.ts +++ b/js/net/src/epoch.test.ts @@ -1,28 +1,23 @@ import { expect, test } from "bun:test"; import * as Epoch from "./epoch.ts"; -test("mint uses UUIDv7 from the wall clock and orders successive identities", () => { - // No mocked clock: UUIDv7 never moves backwards, so any epoch minted earlier in the - // process would pin a mocked past time to the newest one seen. - const before = Date.now(); +test("mint parses as canonical and orders successive identities", () => { const epoch = Epoch.mint(); expect(Epoch.parse(epoch)).toBe(epoch); - expect(Epoch.time(epoch).getTime()).toBeGreaterThanOrEqual(before); expect(Epoch.mint() > epoch).toBe(true); }); // Shared with rs/moq-net's `epoch_vectors`, so both parse and order alike. const vectors = (await Bun.file(new URL("../../../rs/moq-net/src/epoch.json", import.meta.url)).json()) as { - valid: Array<{ text: string; unix_ms: number }>; + valid: string[]; invalid: string[]; ordered: string[]; }; -test("shared epoch parse, reject, order and time vectors", () => { - for (const row of vectors.valid) { - const epoch = Epoch.parse(row.text); - expect(epoch).toBe(row.text as Epoch.Valid); - expect(Epoch.time(epoch).getTime()).toBe(row.unix_ms); +test("shared epoch parse, reject, and order vectors", () => { + for (const text of vectors.valid) { + const epoch = Epoch.parse(text); + expect(epoch).toBe(text as Epoch.Valid); expect(Epoch.fromBytes(Epoch.toBytes(epoch))).toBe(epoch); } for (const text of vectors.invalid) expect(() => Epoch.parse(text)).toThrow(RangeError); diff --git a/js/net/src/epoch.ts b/js/net/src/epoch.ts index b95f8a89db5..78f1881be97 100644 --- a/js/net/src/epoch.ts +++ b/js/net/src/epoch.ts @@ -19,13 +19,6 @@ export function mint(): Valid { return parse(v7()); } -/** Read the wall-clock time encoded in the UUID, with millisecond precision. */ -// biome-ignore lint/style/noRestrictedGlobals: decodes a timestamp, not a clock read. -export function time(epoch: Valid): Date { - // biome-ignore lint/style/noRestrictedGlobals: decodes a timestamp, not a clock read. - return new Date(Number.parseInt(epoch.slice(0, 8) + epoch.slice(9, 13), 16)); -} - /** The UUID's 16 bytes, as the wire carries them. @internal */ export function toBytes(epoch: Valid): Uint8Array { const hex = epoch.replaceAll("-", ""); diff --git a/js/net/src/group.ts b/js/net/src/group.ts index 61edb18538a..08b525a37c8 100644 --- a/js/net/src/group.ts +++ b/js/net/src/group.ts @@ -116,7 +116,11 @@ function appendFrame(state: GroupState, frame: Frame) { let makeDemand: (sequence: number, used: Getter, state: GroupState) => Demand; -/** A watch-only view of a group's mirror readers, including coalesced fetches. */ +/** + * A watch-only view of a group's mirror readers, including coalesced fetches. + * + * @internal The sessions' own fetch coalescing is its only consumer. + */ export class Demand { /** The group sequence this handle watches. */ readonly sequence: number; @@ -313,7 +317,11 @@ export class Producer { return makeConsumer(dst); } - /** Watch the mirror readers attached to this group. */ + /** + * Watch the mirror readers attached to this group. + * + * @internal The sessions' own fetch coalescing is its only consumer. + */ demand(): Demand { return this.#demand; } diff --git a/js/net/src/ietf/parameters.ts b/js/net/src/ietf/parameters.ts index 28915815bf7..abfe20880d1 100644 --- a/js/net/src/ietf/parameters.ts +++ b/js/net/src/ietf/parameters.ts @@ -366,14 +366,6 @@ export class Parameters { this.vars.set(MSG_PARAM_EXPIRES, v); } - get deliveryTimeout(): bigint | undefined { - return this.vars.get(MSG_PARAM_DELIVERY_TIMEOUT); - } - - set deliveryTimeout(v: bigint) { - this.vars.set(MSG_PARAM_DELIVERY_TIMEOUT, v); - } - get maxCacheDuration(): bigint | undefined { return this.vars.get(MSG_PARAM_MAX_CACHE_DURATION); } diff --git a/js/net/src/ietf/setup_options.ts b/js/net/src/ietf/setup_options.ts index 6a60538e57d..3ba20dc11d1 100644 --- a/js/net/src/ietf/setup_options.ts +++ b/js/net/src/ietf/setup_options.ts @@ -97,20 +97,6 @@ export class SetupOptions { return this.vars.get(id); } - removeBytes(id: bigint): boolean { - if (id % 2n !== 1n) { - throw new Error(`invalid parameter id: ${id.toString()}, must be odd`); - } - return this.bytes.delete(id); - } - - removeVarint(id: bigint): boolean { - if (id % 2n !== 0n) { - throw new Error(`invalid parameter id: ${id.toString()}, must be even`); - } - return this.vars.delete(id); - } - async encode(w: Writer, version: IetfVersion) { if (version !== Version.DRAFT_14 && version !== Version.DRAFT_15) { // d17+: no count prefix; d16: count prefix diff --git a/js/net/src/varint.ts b/js/net/src/varint.ts index 1d592475586..e420a59b700 100644 --- a/js/net/src/varint.ts +++ b/js/net/src/varint.ts @@ -36,12 +36,6 @@ function sizeOf(buf: Uint8Array, size: (first: number) => number): number { return n; } -/** Number of bytes needed to encode a value in the leading-ones varint format. */ -export function sizeLeadingOnes(v: number | bigint): number { - const lo = split(v); - return lengthLeadingOnes(parts.hi, lo); -} - /** Encode a value in leading-ones varint format into the provided buffer, returning the written subarray. */ export function encodeLeadingOnesTo(dst: ArrayBuffer, v: number | bigint): Uint8Array { const lo = split(v); diff --git a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Epoch.kt b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Epoch.kt index 6417bdbe744..4d7789de23c 100644 --- a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Epoch.kt +++ b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Epoch.kt @@ -1,17 +1,8 @@ package dev.moq -import kotlin.time.Instant - /** * Mint a fresh publisher epoch from the wall clock and secure randomness, ordered * newest last. Mint one per publisher run and announce it in [AnnounceRoute.epoch], so * viewers see a restart as a new broadcast instead of a stalled one. */ fun mintEpoch(): String = uniffi.moq.moqMintEpoch() - -/** - * The wall-clock time [epoch] encodes, to the millisecond. Throws unless it is a - * lowercase hyphenated UUIDv7. - */ -fun epochTime(epoch: String): Instant = - Instant.fromEpochMilliseconds(uniffi.moq.moqEpochTimeMs(epoch).toLong()) diff --git a/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt b/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt index 98a86a2ff47..ff5c5a574d5 100644 --- a/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt +++ b/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt @@ -40,7 +40,6 @@ import kotlin.test.assertNotNull import kotlin.test.assertNull import kotlin.test.assertTrue import kotlin.time.Duration.Companion.milliseconds -import kotlin.time.Instant @Serializable private data class Status(val state: String) @@ -571,7 +570,6 @@ class SmokeTest { val announced = origin.consume().announced(AnnounceConfig()) origin.createBroadcast("epoch").use { broadcast -> val first = mintEpoch() - assertEquals(Instant.fromEpochMilliseconds(1_700_000_000_000), epochTime("018bcfe5-6800-7000-8000-000000000000")) broadcast.announce(AnnounceRoute(epoch = first)) val start = assertIs(announced.nextRoute()) assertEquals(first, start.announce.route.epoch) @@ -581,7 +579,7 @@ class SmokeTest { val restart = assertIs(announced.nextRoute()) assertEquals(second, restart.announce.route.epoch) - assertFailsWith { epochTime("not-an-epoch") } + assertFailsWith { broadcast.announce(AnnounceRoute(epoch = "not-an-epoch")) } } } } diff --git a/py/moq-rs/docs/index.md b/py/moq-rs/docs/index.md index 2e05799eede..4b1f49a51b3 100644 --- a/py/moq-rs/docs/index.md +++ b/py/moq-rs/docs/index.md @@ -245,7 +245,6 @@ Rust side ([`moq-ffi`](https://crates.io/crates/moq-ffi)). protocol_error log_level mint_epoch - epoch_time ``` ```{toctree} diff --git a/py/moq-rs/moq/__init__.py b/py/moq-rs/moq/__init__.py index 4121392f893..444da326c93 100644 --- a/py/moq-rs/moq/__init__.py +++ b/py/moq-rs/moq/__init__.py @@ -12,7 +12,7 @@ from . import json as json from . import video as video from .client import Client, CongestionControl, connect, qlog_supported, versions -from .epoch import epoch_time, mint_epoch +from .epoch import mint_epoch from .errors import is_auth, is_shutdown, protocol_error from .log import log_level from .origin import ( @@ -147,7 +147,6 @@ "VideoPixelFormat", "WebSocketConfig", "connect", - "epoch_time", "is_auth", "is_shutdown", "log_level", diff --git a/py/moq-rs/moq/epoch.py b/py/moq-rs/moq/epoch.py index 745b7e6d69e..4293fb2f9e1 100644 --- a/py/moq-rs/moq/epoch.py +++ b/py/moq-rs/moq/epoch.py @@ -2,9 +2,7 @@ from __future__ import annotations -from datetime import datetime, timedelta, timezone - -from moq_ffi import moq_epoch_time_ms, moq_mint_epoch +from moq_ffi import moq_mint_epoch def mint_epoch() -> str: @@ -14,12 +12,3 @@ def mint_epoch() -> str: restart reads as a new broadcast to viewers instead of a stalled one. """ return moq_mint_epoch() - - -def epoch_time(epoch: str) -> datetime: - """The UTC wall-clock time an epoch encodes, to the millisecond. - - Raises `moq.Error` unless ``epoch`` is a lowercase hyphenated UUIDv7, and - ``OverflowError`` for a time past year 9999, which ``datetime`` cannot hold. - """ - return datetime(1970, 1, 1, tzinfo=timezone.utc) + timedelta(milliseconds=moq_epoch_time_ms(epoch)) diff --git a/py/moq-rs/tests/test_local.py b/py/moq-rs/tests/test_local.py index 5378c8446d3..b7b17e0cd60 100644 --- a/py/moq-rs/tests/test_local.py +++ b/py/moq-rs/tests/test_local.py @@ -4,7 +4,7 @@ import json import struct from dataclasses import replace -from datetime import datetime, timedelta, timezone +from datetime import timedelta from typing import Any, cast import moq @@ -384,8 +384,6 @@ async def test_announced_epoch_restarts(): broadcast = origin.create_broadcast("epoch") first = moq.mint_epoch() - fixed = datetime(2023, 11, 14, 22, 13, 20, tzinfo=timezone.utc) - assert moq.epoch_time("018bcfe5-6800-7000-8000-000000000000") == fixed broadcast.announce(moq.AnnounceRoute(epoch=first)) start = await next_route(announced) assert isinstance(start, moq.AnnounceEventStart) @@ -398,9 +396,7 @@ async def test_announced_epoch_restarts(): assert restart.announce.route.epoch == second with pytest.raises(moq.Error.InvalidEpoch): - moq.epoch_time(first.upper()) - with pytest.raises(OverflowError): - moq.epoch_time("ffffffff-ffff-7000-8000-000000000000") + broadcast.announce(moq.AnnounceRoute(epoch=first.upper())) with pytest.raises(moq.Error.InvalidEpoch): broadcast.announce(moq.AnnounceRoute(epoch="not-an-epoch")) broadcast.close() diff --git a/quest/m1/api-audit/README.md b/quest/m1/api-audit/README.md index f16910977f4..955517b3214 100644 --- a/quest/m1/api-audit/README.md +++ b/quest/m1/api-audit/README.md @@ -79,7 +79,6 @@ Rejected in planning: - [Two broadcast resolutions](/quest/m1/api-audit/resolve.md) - Rust, JS, and the bindings resolve a broadcast with `request(path, epoch)` or `follow(path, epoch)`, and every racy two-step caller is gone - [JS origin publications](/quest/m1/api-audit/js-publication.md) - `@moq/net` origins hand out a `Publication` that announces, so `Prepared` and standalone broadcasts lose methods that always throw - [Binding publications](/quest/m1/api-audit/binding-publication.md) - moq-ffi, the C ABI, and every wrapper hand out a publication handle that announces, replacing the private adapters -- [moq-net exports with a consumer](/quest/m1/api-audit/net-exports.md) - the dead and test-only moq-net, `@moq/net`, and moq-ffi items the follow-up audit found go private or away - [Bindings read gaps and ends](/quest/m1/api-audit/decode-read-bindings.md) - moq-ffi and every wrapper surface `Gap` and `End` instead of discarding them - [Session-only items stay private](/quest/m1/api-audit/model-private.md) - model items only the built-in sessions use are private - [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 diff --git a/quest/m1/api-audit/net-exports.md b/quest/m1/api-audit/net-exports.md deleted file mode 100644 index 695c612833b..00000000000 --- a/quest/m1/api-audit/net-exports.md +++ /dev/null @@ -1,48 +0,0 @@ -# [M] moq-net exports with a consumer - -## Goal - -The moq-net, `@moq/net`, and moq-ffi items the 2026-10-23 audit follow-up -found without a consumer go private, test-only, or away. The first pass, -which trimmed the model exports added since release and renamed -`Track.Location` to `Position`, landed in -[moq-dev/moq#6470](https://github.com/moq-dev/moq/pull/6470). - -## Plan - -Decided 2026-10-23 (API audit follow-up, maintainer): - -- `Epoch::time` and FFI `moq_epoch_time_ms` with its wrappers: only binding - tests read them. `moq_mint_epoch` stays. -- `announce::Consumer::is_synced` (tests only) goes test-only, and - `announce::Consumer::synced` goes; `poll_synced` stays for `Follow`. -- `varint::MAX_QUIC`: `pub(crate)`. -- JS `Group.Demand` and `ConnectProps`/`AcceptProps.extensions`: `@internal`. -- Dead code, found by marking every pub item `#[deprecated]` and building - the workspace. Delete what nothing references or constructs: - `Error::TooManyParameters`, `DecodeError::{ExpectedData, Missing, - Unsupported}`, `frame::Buffer::{is_full, filled_mut}`, - `frame::Consumer::poll_read_all`, `track::Ordered::poll_finished`, - `Path::next_part`, `stats::Registry::exclude`, - `stats::Traffic::{is_announced, total_bytes}`, `stats::Tier::is_default`, - `transport::poll::Buffered::get_mut`, `setup::Token::CAT`; JS - `Varint.sizeLeadingOnes`, `SetupOptions.{removeBytes, removeVarint}`, and - the `deliveryTimeout` accessors. -- Test-only plumbing, deleted with its tests moved to what remains: the whole - `broadcast::Demand` with both `demand()` getters (JS `Broadcast.Demand` is - already internal), `goaway::Consumer::{peek, recv}`, - `bandwidth::Producer::{abort, closed, used, unused}`, - `Version::{from_alpn, is_ietf, names}`, `Server::with_origin`, - `stats::Handle::{parent, tier}`, and group - `poll_unused`/`abort_unused` (their `allow(dead_code)` waited on an - answering API that is now dropped). Test-only reader and writer - conveniences (`group::Consumer::read_frames`, `frame::Consumer::read_all`, - and the like) stay: applications are their consumer. - -Kept, so the next audit skips them: `transport::memory` stays public and -ungated, since an in-process transport is real API that other crates' tests -use. `Route::{auth, with_auth}` stay public, since `doc/concept/standard.md` -documents them and a shared `auth::Handle` keeps them across a reconnect -(decided 2026-10-22, maintainer). - -Public API: removals and renames in moq-net and `@moq/net`. Wire: none. diff --git a/quest/m1/api-audit/net-names.md b/quest/m1/api-audit/net-names.md index 9a8c7d58826..c50b5328a95 100644 --- a/quest/m1/api-audit/net-names.md +++ b/quest/m1/api-audit/net-names.md @@ -13,8 +13,7 @@ Decided 2026-10-23 (API audit follow-up, maintainer): - **Demand says `wanted`.** `group::Demand` drops `is_used`/`used`/`unused`/`poll_used`/`poll_unused` for `track::Demand`'s `wanted()` snapshot plus `wanted_changed()` (and its poll). - `broadcast::Demand` is deleted instead - ([net-exports](/quest/m1/api-audit/net-exports.md)). + `broadcast::Demand` is already deleted. - **Allocator.** `Session::bandwidth()` returns the send allocator beside `send_bandwidth()`/`recv_bandwidth()` estimates: rename it `allocator()`, and `Client::with_bandwidth` `with_allocator`. diff --git a/rs/AGENTS.md b/rs/AGENTS.md index 39786d01644..12d36ff1cff 100644 --- a/rs/AGENTS.md +++ b/rs/AGENTS.md @@ -12,7 +12,7 @@ Component crates are named `moq-` (`hang` is the exception); infrastr # Producer / Consumer -Most crates use a split-handle pattern: a `Producer` writes, any number of `Consumer`s read, state is shared via `kio`. Handles are refcounted: clones share state, the last drop closes cleanly, `abort(err)` closes with an error. A parked clone that keeps something alive is a bug in the holder, and a consumer never keeps its producer or parent alive. A track, group, or frame consumer seeing `Error::Dropped` is a bug too: its producer went away unfinished without calling `abort` with a specific error, so find that path and abort with the real cause. Lifecycle watchers like `broadcast::Demand` return `Dropped` after `close()` by design. +Most crates use a split-handle pattern: a `Producer` writes, any number of `Consumer`s read, state is shared via `kio`. Handles are refcounted: clones share state, the last drop closes cleanly, `abort(err)` closes with an error. A parked clone that keeps something alive is a bug in the holder, and a consumer never keeps its producer or parent alive. A track, group, or frame consumer seeing `Error::Dropped` is a bug too: its producer went away unfinished without calling `abort` with a specific error, so find that path and abort with the real cause. Lifecycle watchers like `track::Demand` return `Dropped` after `close()` by design. # Async / poll diff --git a/rs/moq-cli/src/args.rs b/rs/moq-cli/src/args.rs index bd141019e16..b850ad2c92f 100644 --- a/rs/moq-cli/src/args.rs +++ b/rs/moq-cli/src/args.rs @@ -3061,7 +3061,7 @@ mod tests { } } - /// Every `*-version` flag offers exactly the versions [`Version::names`] parses. + /// Every `*-version` flag offers exactly the versions moq-net parses. /// /// The lists are `choices(...)` literals because Usage reads them at expansion /// time, so they are copies. This is what keeps a new protocol draft from @@ -3083,9 +3083,13 @@ mod tests { } } - // As a set: `names()` is preference-ordered (newest first) while a choice - // list reads ascending, and that ordering is a presentation call. - let mut expected: Vec<&str> = hang::moq_net::Version::names().collect(); + // As a set: a choice list reads ascending, and that ordering is a presentation call. + // Every wire code moq-net parses, including the opt-in `-wip` drafts. + let codes = (0xff0dad00..=0xff0dadff).chain(0xff000000..=0xff0000ff); + let mut expected: Vec<&str> = codes + .filter_map(hang::moq_net::Version::from_code) + .map(|version| version.as_str()) + .collect(); expected.sort_unstable(); let mut found = Vec::new(); walk(Cli::spec().root, &mut found); @@ -3093,7 +3097,7 @@ mod tests { assert!(!found.is_empty(), "no version flag carried a choice list"); for (long, choices) in &mut found { choices.sort_unstable(); - assert_eq!(choices, &expected, "--{long} is out of step with Version::names()"); + assert_eq!(choices, &expected, "--{long} is out of step with moq-net's versions"); } } } diff --git a/rs/moq-cli/tests/integration/listen.rs b/rs/moq-cli/tests/integration/listen.rs index 9bf6e859734..67ca7b6c10b 100644 --- a/rs/moq-cli/tests/integration/listen.rs +++ b/rs/moq-cli/tests/integration/listen.rs @@ -120,7 +120,7 @@ async fn drains(stage: &str) { let (connection, _origin) = connects(&listening, stage).await; let draining = connection.draining().unwrap(); listening.signal(); - let goaway = tokio::time::timeout(TIMEOUT, draining.recv()) + let goaway = tokio::time::timeout(TIMEOUT, moq_net::kio::wait(|waiter| draining.poll(waiter))) .await .expect("no GOAWAY after SIGTERM") .expect("CLI closed the session without GOAWAY"); @@ -178,7 +178,10 @@ async fn an_export_stage_ending_during_drain_waits_for_every_session() { let draining = publisher.draining().unwrap(); let signalled = std::time::Instant::now(); listening.signal(); - tokio::time::timeout(TIMEOUT, draining.recv()).await.unwrap().unwrap(); + tokio::time::timeout(TIMEOUT, moq_net::kio::wait(|waiter| draining.poll(waiter))) + .await + .unwrap() + .unwrap(); publisher.closed().await.unwrap(); tokio::time::timeout(TIMEOUT, legacy.closed()) .await diff --git a/rs/moq-ffi/src/bandwidth.rs b/rs/moq-ffi/src/bandwidth.rs index 74f8d079768..a46db8781fb 100644 --- a/rs/moq-ffi/src/bandwidth.rs +++ b/rs/moq-ffi/src/bandwidth.rs @@ -347,7 +347,7 @@ mod consumer_tests { tokio::time::pause(); let producer = moq_net::bandwidth::Producer::new(); let consumer = MoqBandwidthConsumer::new(producer.consume()); - producer.abort(moq_net::Error::Cancel).unwrap(); - assert!(matches!(consumer.rate_changed().await, Err(MoqError::Cancelled))); + drop(producer); + assert!(consumer.rate_changed().await.is_err()); } } diff --git a/rs/moq-ffi/src/origin.rs b/rs/moq-ffi/src/origin.rs index 959e35ee82d..ffffbf609fc 100644 --- a/rs/moq-ffi/src/origin.rs +++ b/rs/moq-ffi/src/origin.rs @@ -99,18 +99,6 @@ pub fn moq_mint_epoch() -> String { moq_net::Epoch::mint().to_string() } -/// The wall-clock time an epoch encodes, in milliseconds since the Unix epoch. -/// -/// Errors with `InvalidEpoch` unless the text is a lowercase hyphenated UUIDv7. -#[uniffi::export] -pub fn moq_epoch_time_ms(epoch: String) -> Result { - let epoch: moq_net::Epoch = epoch.parse()?; - // This numeric API retains UUIDv7's full 48-bit range, beyond Wall's year 2554. - // Parsing above guarantees these fields are canonical hexadecimal Unix milliseconds. - let text = epoch.as_str(); - Ok(u64::from_str_radix(&text[..8], 16).unwrap() << 16 | u64::from_str_radix(&text[9..13], 16).unwrap()) -} - /// An origin in which broadcasts can be published. #[derive(uniffi::Object)] pub struct MoqOriginProducer { diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index e5acb1baa0e..e6ff01c7342 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -423,18 +423,8 @@ async fn announced_route_keeps_static_cost_on_reannounce() { #[test] fn epoch_helpers_mint_parse_and_refuse() { - assert_eq!( - moq_epoch_time_ms("ffffffff-ffff-7fff-bfff-ffffffffffff".into()).unwrap(), - (1_u64 << 48) - 1 - ); let epoch = moq_mint_epoch(); let parsed: moq_net::Epoch = epoch.parse().unwrap(); - let unix_ms = parsed - .time() - .unwrap() - .saturating_duration_since(moq_clock::Wall::UNIX_EPOCH) - .as_millis(); - assert_eq!(u128::from(moq_epoch_time_ms(epoch.clone()).unwrap()), unix_ms); assert!(moq_mint_epoch() > epoch, "a later mint orders newer"); // The text the route carries round-trips, and anything but a UUIDv7 is refused. @@ -448,10 +438,6 @@ fn epoch_helpers_mint_parse_and_refuse() { let uuid_v4 = "6ba7b810-9dad-41d1-80b4-00c04fd430c8".to_string(); for bad in [uuid_v4, epoch.to_uppercase(), String::new()] { - assert!(matches!( - moq_epoch_time_ms(bad.clone()), - Err(MoqError::InvalidEpoch { .. }) - )); let route = MoqAnnounceRoute { epoch: Some(bad), ..Default::default() @@ -7339,7 +7325,8 @@ async fn memory_shutdown_pair() -> (Arc, Arc) { }; let accept = async { let (session, driver) = moq_net::Server::new() - .with_origin(b.inner.clone()) + .with_publisher(&b.inner) + .with_subscriber(b.inner.clone()) .with_versions(version.into()) .accept(moq_clock::Instant::now(), server) .await diff --git a/rs/moq-net/AGENTS.md b/rs/moq-net/AGENTS.md index 1b00fcc5dbc..9f66f70687b 100644 --- a/rs/moq-net/AGENTS.md +++ b/rs/moq-net/AGENTS.md @@ -7,7 +7,7 @@ The wire layer. Generic over the transport and media-agnostic: the relay never i - A route change is never fatal on its own. A subscription stays on the instance it started on while that instance still has a route, even after a newer route wins the path. Once none is left (e.g. the publisher's session closed), it ends with an error, never a clean end, and the app requests the path again. Never stitch it onto another instance. A clean end means only a deliberate publisher finish or the local session closing. Players start, stop, and restart on announcements, never on request errors. See "Publisher epochs" in `doc/concept/moq-lite.md`. - The cache is a performance optimization. A relay behaves the same warm or cold: report upstream state (e.g. the largest group), not what the cache happens to hold. - Sessions are caller-driven: `connect`/`accept` return `(Session, Driver)`. The `Driver` holds no `Session` clone, so spawning it never keeps the session alive. -- Version matching: outside the crate branch on `is_lite()` / `is_ietf()` and `alpn()`; inside it match the draft enums. List old drafts explicitly and let the newest behavior be the fallthrough so future drafts fall forward. +- Version matching: outside the crate branch on `is_lite()` and `alpn()`; inside it match the draft enums. List old drafts explicitly and let the newest behavior be the fallthrough so future drafts fall forward. - Interop rule of thumb: respond to every feature a peer may use, but don't request new ones. Never emit or accept partial groups; a group starts at frame 0. Exception: moq-transport passthrough forwards a mid-group start as the publisher sent it. - `Timestamp` is an instant with a scale, not a scalar. `checked_add`/`checked_sub` error on mismatched scales; `.convert()` first. `ZERO` is second-scale, so seed a max with `Option`, not `ZERO`. A track's timestamps never rewind. A local write refuses one, but some received tracks skip that check: a peer's rewind must never panic or cost unbounded work, yet results on that track may be wrong, so spend nothing on staying correct after one. diff --git a/rs/moq-net/MIGRATE.md b/rs/moq-net/MIGRATE.md index f57d1574730..b0d41ad16ac 100644 --- a/rs/moq-net/MIGRATE.md +++ b/rs/moq-net/MIGRATE.md @@ -4,6 +4,22 @@ How to update your code for each breaking change, newest first. ## Unreleased +### Exports without a consumer are gone + +These had no caller outside tests; replace them as follows. + +- `broadcast::Demand` and both `broadcast::{Producer, Consumer}::demand()`: watch each track's `track::Demand` instead. +- `goaway::Consumer::{peek, recv}`: poll it, `goaway.poll(&kio::Waiter::noop())` to peek or `kio::wait(|waiter| goaway.poll(waiter)).await` to wait. +- `bandwidth::Producer::{abort, closed}`: drop the last producer to close it. +- `Version::{names, from_alpn, is_ietf}`: use `Versions::all().iter()` or `Version::from_code`, match `alpn()` yourself, and `!is_lite()`. +- `Server::with_origin(origin)`: call `with_publisher(&origin).with_subscriber(origin)`. +- `stats::Handle::{parent, tier}`, `stats::Registry::exclude`, `stats::Traffic::{is_announced, total_bytes}`, `stats::Tier::is_default`: read the public fields, such as `announces_started > announces_ended` or `tier.label().is_empty()`. +- `frame::Consumer::poll_read_all`: use `read_all`, or poll `poll_read_chunk` until the frame ends. +- `frame::Buffer::{is_full, filled_mut}`: compare `len()` with `capacity()`, and read `filled()`. +- `Path::next_part`: split `as_str()` on the first `/`. +- `setup::Token::CAT`: compare `kind` with `1`. +- `Error::TooManyParameters` and `DecodeError::{ExpectedData, Missing, Unsupported}` were never produced; drop their match arms. + ### `Client::with_url` replaces `with_path` and `with_authority` Pass the URL you dial: `Client::new().with_url(url)` instead of `.with_path(path).with_authority(authority)`. @@ -473,7 +489,7 @@ A publisher that restarts its group numbering must publish under a new broadcast ### No `VarInt` Varints are plain `u64`s. -Use `moq_net::varint::decode_quic(buf)` and `moq_net::varint::encode_quic(v, buf)`, which fails past `varint::MAX_QUIC`. +Use `moq_net::varint::decode_quic(buf)` and `moq_net::varint::encode_quic(v, buf)`, which fails past `2^62 - 1`. ## 0.2 -> 0.3 diff --git a/rs/moq-net/benches/session.rs b/rs/moq-net/benches/session.rs index 440508101c8..c6ed587d981 100644 --- a/rs/moq-net/benches/session.rs +++ b/rs/moq-net/benches/session.rs @@ -419,7 +419,7 @@ impl Cluster { options.transport.stream_buffer = self.window; // Draft 22, the one IETF version here, keeps its SETUP stream open beside the groups. options.transport.max_concurrent_uni_streams = - self.credit.map(|credit| credit + usize::from(self.version.is_ietf())); + self.credit.map(|credit| credit + usize::from(!self.version.is_lite())); // Hidden included, as the dash's `.stats/` scope sees its subtree. options.server_publish = Some(self.relays[relay].consume().with_hidden(true)); options.client_subscribe = Some(origin.clone()); diff --git a/rs/moq-net/benches/synced.rs b/rs/moq-net/benches/synced.rs index a648e25131e..1716a716c37 100644 --- a/rs/moq-net/benches/synced.rs +++ b/rs/moq-net/benches/synced.rs @@ -60,9 +60,9 @@ fn bench_land(c: &mut Criterion) { let watcher = scoped(&table.producer, "room/incoming").announced(); b.iter(|| { let snapshot = fuzz::snapshot(&table.producer, "room/incoming"); - assert!(!watcher.is_synced()); + assert!(!watcher.poll_synced(&moq_net::kio::Waiter::noop()).is_ready()); drop(snapshot); - assert!(watcher.is_synced()); + assert!(watcher.poll_synced(&moq_net::kio::Waiter::noop()).is_ready()); }); }); } @@ -79,7 +79,7 @@ fn bench_register(c: &mut Criterion) { let _held = fuzz::snapshot(&table.producer, "room/incoming"); b.iter(|| { let watcher = incoming.announced(); - assert!(!watcher.is_synced()); + assert!(!watcher.poll_synced(&moq_net::kio::Waiter::noop()).is_ready()); }); }); } diff --git a/rs/moq-net/src/auth.rs b/rs/moq-net/src/auth.rs index bf182107ff2..b6ad83f0fe8 100644 --- a/rs/moq-net/src/auth.rs +++ b/rs/moq-net/src/auth.rs @@ -51,7 +51,6 @@ pub(crate) fn protocol_error(err: Error) -> Error { Error::Decode(_) | Error::BoundsExceeded(_) | Error::WrongSize - | Error::TooManyParameters | Error::UnexpectedMessage | Error::UnexpectedStream => Error::ProtocolViolation, err => err, @@ -3049,7 +3048,7 @@ mod request_token_tests { async fn a_typed_request_token_is_unsupported() { let handle = Link::new(true); let mut requests = handle.handle().requests(); - let verdict = handle.verify_request(token(b"cat", crate::setup::Token::CAT), None); + let verdict = handle.verify_request(token(b"cat", 1), None); let err = verdict.grant().await.expect_err("a typed token"); assert!(matches!(err.error, Error::Unsupported), "{err:?}"); assert!( diff --git a/rs/moq-net/src/coding/decode.rs b/rs/moq-net/src/coding/decode.rs index 2d050ee3b72..8d266df1017 100644 --- a/rs/moq-net/src/coding/decode.rs +++ b/rs/moq-net/src/coding/decode.rs @@ -71,22 +71,10 @@ pub enum DecodeError { max: usize, }, - /// The stream ended where a payload was required. - #[error("expected data")] - ExpectedData, - /// A parameter or field appeared more than once. #[error("duplicate")] Duplicate, - /// A required parameter or field was absent. - #[error("missing")] - Missing, - - /// The value is well-formed but this implementation does not handle it. - #[error("unsupported")] - Unsupported, - /// Bytes remained after the value was fully decoded. #[error("trailing bytes")] TrailingBytes, diff --git a/rs/moq-net/src/coding/varint.rs b/rs/moq-net/src/coding/varint.rs index 33e4ec6945d..b7c2b5739e6 100644 --- a/rs/moq-net/src/coding/varint.rs +++ b/rs/moq-net/src/coding/varint.rs @@ -15,7 +15,7 @@ use crate::{Version, ietf, lite}; pub struct BoundsExceeded; /// The largest value the QUIC form can carry: `2^62 - 1`. -pub const MAX_QUIC: u64 = (1 << 62) - 1; +pub(crate) const MAX_QUIC: u64 = (1 << 62) - 1; /// How a protocol version lays out a varint on the wire. #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -265,7 +265,7 @@ pub fn decode_quic(r: &mut R) -> Result { /// Encode `value` as a QUIC varint (two-bit length tag). /// -/// Fails with [`EncodeError::BoundsExceeded`] past [`MAX_QUIC`], writing nothing. +/// Fails with [`EncodeError::BoundsExceeded`] past `2^62 - 1`, writing nothing. pub fn encode_quic(value: u64, w: &mut W) -> Result<(), EncodeError> { let len = size(value, Form::Quic)?; if w.remaining_mut() < len { diff --git a/rs/moq-net/src/epoch.json b/rs/moq-net/src/epoch.json index baa8065bf82..505e6b8be4d 100644 --- a/rs/moq-net/src/epoch.json +++ b/rs/moq-net/src/epoch.json @@ -1,25 +1,10 @@ { "valid": [ - { - "text": "00000000-0000-7000-8000-000000000000", - "unix_ms": 0 - }, - { - "text": "0199b7f4-3c2a-7d1e-9f0b-2b6c1a9d8e7f", - "unix_ms": 1759727860778 - }, - { - "text": "0199b7f4-3c2a-7d1e-af0b-2b6c1a9d8e7f", - "unix_ms": 1759727860778 - }, - { - "text": "0199b7f4-9e10-7f42-8a3c-5d7e2b1c0f9a", - "unix_ms": 1759727885840 - }, - { - "text": "ffffffff-ffff-7fff-bfff-ffffffffffff", - "unix_ms": 281474976710655 - } + "00000000-0000-7000-8000-000000000000", + "0199b7f4-3c2a-7d1e-9f0b-2b6c1a9d8e7f", + "0199b7f4-3c2a-7d1e-af0b-2b6c1a9d8e7f", + "0199b7f4-9e10-7f42-8a3c-5d7e2b1c0f9a", + "ffffffff-ffff-7fff-bfff-ffffffffffff" ], "invalid": [ "", diff --git a/rs/moq-net/src/epoch.rs b/rs/moq-net/src/epoch.rs index f398e5d5c91..7e1796a5cfb 100644 --- a/rs/moq-net/src/epoch.rs +++ b/rs/moq-net/src/epoch.rs @@ -1,8 +1,6 @@ -use moq_clock::{OutOfRange, Wall}; use std::fmt; use std::str::FromStr; use std::sync::Arc; -use std::time::Duration; /// One publisher instance's identity, the lowercase hyphenated text of a UUIDv7. #[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] @@ -24,16 +22,6 @@ impl Epoch { pub fn as_str(&self) -> &str { &self.0 } - - /// The UUID's wall time in milliseconds, or an error outside Wall's range. - pub fn time(&self) -> Result { - // Parsing enforces the canonical UUID form, whose first 48 bits are Unix milliseconds. - let millis = - u64::from_str_radix(&self.0[..8], 16).unwrap() << 16 | u64::from_str_radix(&self.0[9..13], 16).unwrap(); - Wall::UNIX_EPOCH - .checked_add(Duration::from_millis(millis)) - .ok_or(OutOfRange) - } } impl Epoch { @@ -101,14 +89,9 @@ mod tests { fn epoch_vectors() { let vectors: serde_json::Value = serde_json::from_str(include_str!("epoch.json")).unwrap(); for row in vectors["valid"].as_array().unwrap() { - let text = row["text"].as_str().unwrap(); + let text = row.as_str().unwrap(); let epoch: Epoch = text.parse().unwrap(); assert_eq!(epoch.as_str(), text); - let millis = row["unix_ms"].as_u64().unwrap(); - match Wall::UNIX_EPOCH.checked_add(Duration::from_millis(millis)) { - Some(expected) => assert_eq!(epoch.time().unwrap(), expected), - None => assert_eq!(epoch.time(), Err(OutOfRange)), - } } for row in vectors["invalid"].as_array().unwrap() { assert!(row.as_str().unwrap().parse::().is_err(), "{row}"); diff --git a/rs/moq-net/src/error.rs b/rs/moq-net/src/error.rs index 7cde78255cc..a3a7ca7a81a 100644 --- a/rs/moq-net/src/error.rs +++ b/rs/moq-net/src/error.rs @@ -456,10 +456,6 @@ pub enum Error { #[error(transparent)] Encode(#[from] coding::EncodeError), - /// A message carried more parameters than this endpoint accepts. - #[error("too many parameters")] - TooManyParameters, - /// The peer went past this session's `crate::session::Limits`: too many announcements, or /// too many requests on moq-transport drafts 14 to 16. #[error("too many requests")] @@ -638,7 +634,6 @@ impl From<&Error> for SessionError { Error::Expired => Self::Expired, Error::Unsupported => Self::Unsupported, Error::Version | Error::UnknownAlpn(_) => Self::Version, - Error::TooManyParameters => Self::KeyValueFormatting, Error::TooManyRequests => Self::TooManyRequests, Error::GoawayTimeout => Self::GoawayTimeout, Error::Timeout => Self::Timeout, @@ -727,7 +722,6 @@ impl From<&Error> for StreamError { Error::UnexpectedStream | Error::NoStreamCredit => Self::Internal, Error::Version | Error::UnknownAlpn(_) - | Error::TooManyParameters | Error::TooManyRequests | Error::Duplicate | Error::GoawayTimeout diff --git a/rs/moq-net/src/fuzz.rs b/rs/moq-net/src/fuzz.rs index bacfd781d51..2b0f83e5d27 100644 --- a/rs/moq-net/src/fuzz.rs +++ b/rs/moq-net/src/fuzz.rs @@ -669,7 +669,7 @@ where /// ones, and the two disagree about which byte sequences are even legal. /// /// The leading-ones form spans the full `u64`, while the QUIC form stops at -/// [`crate::coding::varint::MAX_QUIC`]; a value always re-encodes in the form it was read in. +/// `2^62 - 1`; a value always re-encodes in the form it was read in. pub fn varint(data: &[u8]) -> bool { let Some((&selector, rest)) = data.split_first() else { return false; diff --git a/rs/moq-net/src/goaway.rs b/rs/moq-net/src/goaway.rs index 2404e9f10cb..6fa14f4c988 100644 --- a/rs/moq-net/src/goaway.rs +++ b/rs/moq-net/src/goaway.rs @@ -144,14 +144,6 @@ pub struct Consumer { } impl Consumer { - /// The peer's GOAWAY, or `None` if it has not sent one. - /// - /// The synchronous read behind [`poll`](Self::poll), for callers that only - /// want to know whether the session is draining. - pub fn peek(&self) -> Option { - self.state.read().clone() - } - /// Poll for the peer's GOAWAY. /// /// `Ready(Err)` once the session closes without one, so a caller can stop @@ -166,12 +158,6 @@ impl Consumer { Poll::Pending => Poll::Pending, } } - - /// Wait for the peer's GOAWAY, or `None` if the session closes without one. - #[cfg(feature = "async")] - pub async fn recv(&self) -> Option { - kio::wait(|waiter| self.poll(waiter)).await.ok() - } } /// The receive-side GOAWAY signal handed to a subscriber. @@ -494,13 +480,12 @@ mod tests { assert!(matches!(err, Error::ProtocolViolation)); } - /// The consumer reports the recorded GOAWAY both synchronously and by polling. - #[moq_clock::test] - #[cfg(feature = "async")] - async fn consumer_observes_the_recorded_goaway() { + /// The consumer reports the recorded GOAWAY once it arrives. + #[test] + fn consumer_observes_the_recorded_goaway() { let (handle, protocol) = Handle::new(true); let consumer = handle.consumer(); - assert_eq!(consumer.peek(), None); + assert!(consumer.poll(&kio::Waiter::noop()).is_pending()); let goaway = Goaway { uri: "https://elsewhere.example/".to_string(), @@ -508,18 +493,22 @@ mod tests { }; protocol.record(goaway.clone()).unwrap(); - assert_eq!(consumer.peek(), Some(goaway.clone())); - assert_eq!(consumer.recv().await, Some(goaway)); + let Poll::Ready(Ok(received)) = consumer.poll(&kio::Waiter::noop()) else { + panic!("the recorded GOAWAY"); + }; + assert_eq!(received, goaway); } - /// A session that closes without a GOAWAY resolves `recv` rather than parking + /// A session that closes without a GOAWAY resolves the poll rather than parking /// forever, so a caller watching for one can stop. - #[moq_clock::test] - #[cfg(feature = "async")] - async fn recv_resolves_when_the_session_closes() { + #[test] + fn poll_resolves_when_the_session_closes() { let (handle, protocol) = Handle::new(true); let consumer = handle.consumer(); drop(protocol); - assert_eq!(consumer.recv().await, None); + assert!(matches!( + consumer.poll(&kio::Waiter::noop()), + Poll::Ready(Err(Error::Closed)) + )); } } diff --git a/rs/moq-net/src/ietf/session.rs b/rs/moq-net/src/ietf/session.rs index 2875c771bc5..104314e4109 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -3818,7 +3818,10 @@ mod tests { .expect_err("a second GOAWAY is fatal"); assert_eq!(SessionError::from(&err), SessionError::ProtocolViolation, "{version:?}"); assert!( - handle.consumer().peek().is_none(), + !matches!( + handle.consumer().poll(&kio::Waiter::noop()), + std::task::Poll::Ready(Ok(_)) + ), "{version:?}: the first GOAWAY surfaced" ); } diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 42dd0314e49..ab8944ba0ea 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -140,7 +140,6 @@ pub(super) fn is_protocol_violation(err: &Error) -> bool { Error::Decode(_) | Error::BoundsExceeded(_) | Error::WrongSize - | Error::TooManyParameters | Error::ProtocolViolation | Error::UnexpectedMessage | Error::UnexpectedStream diff --git a/rs/moq-net/src/model/bandwidth.rs b/rs/moq-net/src/model/bandwidth.rs index 467eb242f5a..54fcf1db2b0 100644 --- a/rs/moq-net/src/model/bandwidth.rs +++ b/rs/moq-net/src/model/bandwidth.rs @@ -74,7 +74,6 @@ impl Rate { #[derive(Default)] struct State { bitrate: Option, - abort: Option, } /// Produces bandwidth estimates, notifying consumers when the value changes. @@ -93,7 +92,7 @@ impl Producer { /// Set the current bandwidth estimate, or `None` while the backend has none. pub fn set(&self, bitrate: Option) -> Result<()> { - let mut state = self.modify()?; + let mut state = self.state.write().map_err(|_| Error::Dropped)?; if state.bitrate != bitrate { state.bitrate = bitrate; } @@ -114,21 +113,6 @@ impl Producer { Weak(self.state.weak()) } - /// Close the producer with an error, notifying all consumers. - pub fn abort(&self, err: Error) -> Result<()> { - let mut state = self.modify()?; - state.abort = Some(err); - state.close(); - Ok(()) - } - - /// Block until the channel is closed, returning the cause. - #[cfg(feature = "async")] - pub async fn closed(&self) -> Error { - self.state.closed().await; - self.close_error() - } - /// Block until there are no active consumers. #[cfg(feature = "async")] pub async fn unused(&self) -> Result<()> { @@ -137,7 +121,7 @@ impl Producer { /// Poll until there are no active consumers. Errors if the channel closes first. pub fn poll_unused(&self, waiter: &kio::Waiter) -> Poll> { - self.state.poll_unused(waiter).map_err(|_| self.close_error()) + self.state.poll_unused(waiter).map_err(|_| Error::Dropped) } /// Whether at least one active consumer exists right now. @@ -153,18 +137,7 @@ impl Producer { /// Poll until at least one active consumer exists. Errors if the channel closes first. pub fn poll_used(&self, waiter: &kio::Waiter) -> Poll> { - self.state.poll_used(waiter).map_err(|_| self.close_error()) - } - - fn modify(&self) -> Result> { - self.state - .write() - .map_err(|r| r.abort.clone().unwrap_or(Error::Dropped)) - } - - /// The close error, once the channel is closed. - fn close_error(&self) -> Error { - self.state.read().abort.clone().unwrap_or(Error::Dropped) + self.state.poll_used(waiter).map_err(|_| Error::Dropped) } } @@ -717,7 +690,7 @@ impl Consumer { // terminal. Collapsing this into `Ok(None)` would be indistinguishable // from a live-but-unavailable estimate, and since a closed channel is // always immediately ready, a `select!` over it would spin forever. - Poll::Ready(Err(state)) => return Poll::Ready(Err(state.abort.clone().unwrap_or(Error::Dropped))), + Poll::Ready(Err(_)) => return Poll::Ready(Err(Error::Dropped)), Poll::Pending => return Poll::Pending, }, // A share recomputes its slice on every wakeup, so it filters the @@ -1039,7 +1012,7 @@ mod tests { estimate.set(None).unwrap(); assert_eq!(share.changed().await.unwrap(), None); - estimate.abort(Error::Cancel).unwrap(); + drop(estimate); assert!(share.changed().await.is_err()); assert!(share.changed().await.is_err()); } @@ -1225,7 +1198,7 @@ mod tests { assert_eq!(consumer.changed().await.unwrap(), None); // Gone for good. - producer.abort(Error::Cancel).unwrap(); + drop(producer); assert!(consumer.changed().await.is_err()); // And it stays terminal rather than flapping back to a value. assert!(consumer.changed().await.is_err()); diff --git a/rs/moq-net/src/model/broadcast.rs b/rs/moq-net/src/model/broadcast.rs index 9f4f8a17950..12474e48c46 100644 --- a/rs/moq-net/src/model/broadcast.rs +++ b/rs/moq-net/src/model/broadcast.rs @@ -167,37 +167,6 @@ impl BroadcastState { } self.requests.get(name).map(|pending| pending.peek()) } - - /// Live demand: a pending request or a consumed track. See [`Demand`]. - fn is_used(&self) -> bool { - !self.requests.is_empty() || self.tracks.iter().any(|track| track.is_used()) - } - - /// Whether [`Self::is_used`] is `want`, parking `waiter` on the tracks in the way - /// otherwise. The consumer counts live on the per-track channels, whose flips don't - /// write this state, so each track's own answer decides: a re-read after the polls - /// would see a reader that came or went since, with nothing parked to wake on its - /// return. - fn poll_demand(&self, waiter: &kio::Waiter, want: bool) -> Poll<()> { - if want { - // A closed track is never read again, so only an open, read one is demand. - let used = !self.requests.is_empty() - || self - .tracks - .iter() - .any(|track| matches!(track.poll_used(waiter), Poll::Ready(Ok(())))); - return if used { Poll::Ready(()) } else { Poll::Pending }; - } - if !self.requests.is_empty() { - return Poll::Pending; - } - // Every other track is unused already, so the first one still read is what to - // wait on: its last reader leaving wakes us to look again. - match self.tracks.iter().all(|track| track.poll_unused(waiter).is_ready()) { - true => Poll::Ready(()), - false => Poll::Pending, - } - } } /// Manages tracks within a broadcast. @@ -342,14 +311,6 @@ impl Producer { &self.info } - /// A watch-only handle to the broadcast's demand. See [`Demand`]. - pub fn demand(&self) -> Demand { - Demand { - alive: self.alive.token.consume().weak(), - state: self.state.clone(), - } - } - /// Produce a new track and insert it into the broadcast. /// /// Pass a name and an optional [`track::Info`], so a bare name works: @@ -480,7 +441,7 @@ impl Producer { self.alive.close(); } - /// Whether any [`Consumer`] handle exists, unlike [`Self::demand`], which counts tracks. + /// Whether any [`Consumer`] handle exists, which is not itself demand for a track. pub(super) fn is_held(&self) -> bool { self.alive.token.is_used() } @@ -1031,21 +992,6 @@ impl Consumer { state.peek(name).or_else(|| ended.filter(|_| state.closing).cloned()) } - /// A watch-only handle to the broadcast's demand. See [`Demand`]. - /// - /// The consumer-side sibling of [`Producer::demand`], for a holder that has - /// only a read handle. Holding this handle, or the [`Consumer`] it came from, - /// is not itself demand. - /// - /// Demand going away is [`Demand::poll_unused`] resolving. The broadcast going - /// away is [`Error::Dropped`], which here means the upstream producer ended. - pub fn demand(&self) -> Demand { - Demand { - alive: self.alive.weak(), - state: self.state.clone(), - } - } - /// Block until the broadcast ends, by [`Producer::close`] or every producer dropping. #[cfg(feature = "async")] pub async fn closed(&self) { @@ -1126,69 +1072,6 @@ impl super::WeakEntry for WeakConsumer { } } -/// A cloneable, watch-only handle to a broadcast's subscriber demand. -/// -/// Obtained from [`Producer::demand`] or [`Consumer::demand`]; the broadcast-level sibling of -/// [`track::Demand`](crate::track::Demand). Demand means live interest in the -/// broadcast's content: a pending track request or a consumed track. A publisher -/// uses it to run expensive work only while someone is watching. -/// -/// It's a weak handle: it neither keeps the broadcast alive nor counts as -/// demand itself. Once every producer is gone, [`poll_used`](Self::poll_used) / -/// [`poll_unused`](Self::poll_unused) return [`Error::Dropped`]. -#[derive(Clone)] -pub struct Demand { - alive: kio::ConsumerWeak<()>, - state: kio::Shared, -} - -impl Demand { - /// Whether the broadcast has live demand right now. - /// - /// A point-in-time snapshot with no registration; use [`Self::poll_used`] / - /// [`Self::poll_unused`] (or their `poll_*` forms) to wait for the edge. - pub fn is_used(&self) -> bool { - self.state.read().is_used() - } - - /// Block until the broadcast has demand. Resolves immediately if it already - /// does; returns [`Error::Dropped`] once every producer is gone. - #[cfg(feature = "async")] - pub async fn used(&self) -> Result<(), Error> { - kio::wait(|waiter| self.poll_used(waiter)).await - } - - /// Block until the broadcast has no demand. Resolves immediately if it has - /// none; returns [`Error::Dropped`] once every producer is gone. - #[cfg(feature = "async")] - pub async fn unused(&self) -> Result<(), Error> { - kio::wait(|waiter| self.poll_unused(waiter)).await - } - - /// Poll until a consumer is using this resource. - pub fn poll_used(&self, waiter: &kio::Waiter) -> Poll> { - self.poll_demand(waiter, true) - } - - /// Poll until no consumer is using this resource. - pub fn poll_unused(&self, waiter: &kio::Waiter) -> Poll> { - self.poll_demand(waiter, false) - } - - fn poll_demand(&self, waiter: &kio::Waiter, want: bool) -> Poll> { - // Closure is checked first, matching `track::Demand`: a dead broadcast - // reports Dropped rather than pretending to answer. - if self.alive.poll_closed(waiter).is_ready() { - return Poll::Ready(Err(Error::Dropped)); - } - let ready = self.state.poll(waiter, |state| state.poll_demand(waiter, want)); - match ready { - Poll::Ready(_) => Poll::Ready(Ok(())), - Poll::Pending => Poll::Pending, - } - } -} - #[cfg(test)] #[expect(missing_docs, reason = "test-only assertion helpers")] impl Consumer { @@ -1221,8 +1104,6 @@ impl Consumer { #[cfg(test)] mod test { use super::*; - #[cfg(feature = "async")] - use std::time::Duration; #[test] fn a_subscription_is_registered_before_its_request_is_visible() { @@ -1423,95 +1304,6 @@ mod test { assert_eq!(producer.unique_name(""), "12"); } - /// Await with a timeout so a missed demand wake fails the test instead of - /// hanging it (time is paused, so the timeout fires instantly when idle). - #[cfg(feature = "async")] - async fn expect(fut: impl Future) -> T { - moq_clock::timeout(Duration::from_secs(1), fut) - .await - .expect("timed out waiting for a demand edge") - } - - /// Demand on an ordinary broadcast tracks subscriber interest, not - /// production: a live track producer alone is unused, and so is a bare track - /// handle, a subscribed track is used, and both edges wake parked waiters. - #[moq_clock::test] - #[cfg(feature = "async")] - async fn demand_ordinary() { - let producer = Info::new().produce(); - let consumer = producer.consume(); - let demand = producer.demand(); - - // No demand yet; `unused` resolves immediately. - assert!(!demand.is_used()); - demand.unused().await.unwrap(); - - // Producing alone is not demand. - let _track = producer.create_track("a", crate::track::Info::timed()).unwrap(); - assert!(!demand.is_used()); - - // A bare handle is not demand. - let track = consumer.track("a").unwrap(); - assert!(!demand.is_used()); - - // A subscription appearing wakes a parked `used`. - let mut handle = None; - let (used, ()) = futures::join!(expect(demand.used()), async { handle = Some(track.subscribe(None)) }); - used.unwrap(); - assert!(demand.is_used()); - - // The last consumer dropping wakes a parked `unused`. - let (unused, ()) = futures::join!(expect(demand.unused()), async { drop(handle) }); - unused.unwrap(); - assert!(!demand.is_used()); - - // Every producer gone: both edges report the closure. - producer.close(); - assert!(matches!(demand.used().await, Err(Error::Dropped))); - assert!(matches!(demand.unused().await, Err(Error::Dropped))); - } - - /// One read track keeps the broadcast in demand however many others are unread, and - /// its last reader leaving wakes a parked `unused`. - #[moq_clock::test] - #[cfg(feature = "async")] - async fn demand_unused_waits_for_every_track() { - let producer = Info::new().produce(); - let consumer = producer.consume(); - let demand = producer.demand(); - let _video = producer.create_track("video", None).unwrap(); - let _audio = producer.create_track("audio", None).unwrap(); - - let video = consumer.track("video").unwrap().subscribe(None); - assert!( - demand.poll_unused(&kio::Waiter::noop()).is_pending(), - "a read track is demand" - ); - - let (unused, ()) = futures::join!(expect(demand.unused()), async { drop(video) }); - unused.unwrap(); - } - - /// A consumer demand handle distinguishes lost demand from a dropped producer. - #[moq_clock::test] - #[cfg(feature = "async")] - async fn consumer_demand_reports_dropped_producer() { - let producer = Info::new().produce(); - let consumer = producer.consume(); - let watched = consumer.demand(); - let _video = producer.create_track("video", crate::track::Info::timed()).unwrap(); - - let track = consumer.track("video").unwrap().subscribe(None); - assert!(watched.is_used()); - - let (unused, ()) = futures::join!(expect(watched.unused()), async { drop(track) }); - unused.unwrap(); - - drop(producer); - assert!(matches!(watched.used().await, Err(Error::Dropped))); - assert!(matches!(watched.unused().await, Err(Error::Dropped))); - } - /// Subscribe and assert the result hasn't resolved yet (it stays pending until /// a publisher accepts). Returns the pending subscription to resolve after accepting. macro_rules! subscribe_pending { diff --git a/rs/moq-net/src/model/frame.rs b/rs/moq-net/src/model/frame.rs index eb6947aeb24..0f08d87d5e2 100644 --- a/rs/moq-net/src/model/frame.rs +++ b/rs/moq-net/src/model/frame.rs @@ -157,7 +157,7 @@ impl Buffer { } /// Whether the buffer is at capacity, so [`Self::push`] would refuse. - pub fn is_full(&self) -> bool { + pub(crate) fn is_full(&self) -> bool { self.0.is_full() } @@ -166,8 +166,9 @@ impl Buffer { &self.0 } - /// The frames from the most recent fill, mutably (to take payloads out, say). - pub fn filled_mut(&mut self) -> &mut [Frame] { + /// The frames from the most recent fill, mutably, for the batch a read hands back. + #[cfg(feature = "async")] + pub(crate) fn filled_mut(&mut self) -> &mut [Frame] { &mut self.0 } @@ -935,14 +936,7 @@ impl Consumer { kio::wait(|waiter| self.poll_read_chunk(waiter)).await } - /// Poll for all remaining bytes, resolving once the frame is finished. - pub fn poll_read_all(&mut self, waiter: &kio::Waiter) -> Poll> { - if self.recover.is_none() { - return self.poll_read_all_once(waiter); - } - self.poll_resumed(waiter, Self::poll_read_all_once) - } - + #[cfg(feature = "async")] fn poll_read_all_once(&mut self, waiter: &kio::Waiter) -> Poll> { if self.expired { return Poll::Ready(Err(Error::Old)); @@ -982,7 +976,13 @@ impl Consumer { /// Return all remaining bytes, blocking until the frame is finished. #[cfg(feature = "async")] pub async fn read_all(&mut self) -> Result { - kio::wait(|waiter| self.poll_read_all(waiter)).await + kio::wait(|waiter| { + if self.recover.is_none() { + return self.poll_read_all_once(waiter); + } + self.poll_resumed(waiter, Self::poll_read_all_once) + }) + .await } } diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index c0aafa374c2..92cd0891e36 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -1674,17 +1674,6 @@ impl Producer { self.handle.frame_count() } - /// Abort with `err` only while nothing consumes the group, returning whether it is - /// closed. Consumer creation and the check share a lock, so a reader arriving first - /// keeps the group alive instead of reading the abort. - #[allow( - dead_code, - reason = "only tests call it; the dead-code audit decides whether it stays" - )] - pub(crate) fn abort_unused(&self, err: Error) -> bool { - self.handle.abort_unused(err) - } - /// Whether the group has been aborted (including pool eviction). The track's /// read paths treat an aborted cached group as absent. /// @@ -3577,12 +3566,12 @@ mod test { let consumer = producer.handle.try_consume().expect("open"); assert!( - !producer.abort_unused(crate::Error::Cancel), + !producer.handle.abort_unused(crate::Error::Cancel), "a reader declines the abort" ); drop(consumer); - assert!(producer.abort_unused(crate::Error::Cancel)); + assert!(producer.handle.abort_unused(crate::Error::Cancel)); assert!( producer.handle.try_consume().is_none(), "an aborted group mints no reader" diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index cb205d915f2..211cfe3ca59 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -6938,7 +6938,8 @@ impl AnnounceConsumer { } /// Whether this consumer has synced; see [`Self::poll_synced`]. - pub fn is_synced(&self) -> bool { + #[cfg(test)] + pub(crate) fn is_synced(&self) -> bool { let state = self.state.read(); state.is_closed() || state.synced() } @@ -6967,12 +6968,6 @@ impl AnnounceConsumer { .map(|_| ()) } - /// Wait until this consumer has synced; see [`Self::poll_synced`]. - #[cfg(feature = "async")] - pub async fn synced(&self) { - kio::wait(|waiter| self.poll_synced(waiter)).await - } - // The wire sender uses this only to bound its initial replay. #[cfg(feature = "async")] pub(crate) fn pending_len(&self) -> usize { diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index b79c1c41422..be6b9cff0ff 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -5046,7 +5046,7 @@ struct Drift { /// Keeps one handed-out group tied to the subscription whose cursor selected it. struct GroupExpiry { /// Weak so a handed-out group never joins the track's consumer count, which is what - /// [`crate::broadcast::Demand`] reads to decide the track still has readers. A group + /// [`Demand`] reads to decide the track still has readers. A group /// outlives the cursor that produced it (a relay serves one for the life of its /// stream), and pinning demand on that would hold an upstream subscription open past /// the last real subscriber. @@ -7169,18 +7169,13 @@ impl Ordered { self.inner.update(subscription) } - /// Poll for the track's declared exclusive end, without blocking. - pub fn poll_finished(&mut self, waiter: &kio::Waiter) -> Poll> { - self.inner.poll_finished(waiter) - } - /// Block until the track declares its end, returning the exclusive end position. /// /// See [`Subscriber::finished`]: this reports the declared end, not that every group /// has arrived. #[cfg(feature = "async")] pub async fn finished(&mut self) -> Result { - kio::wait(|waiter| self.poll_finished(waiter)).await + kio::wait(|waiter| self.inner.poll_finished(waiter)).await } /// Poll for the largest location within [`Self::with_range`]; see [`Subscriber::poll_largest`]. @@ -14407,7 +14402,10 @@ mod test { drop(first.await.unwrap()); let hit = consumer.fetch_group(5, None); - assert!(!group.abort_unused(Error::Cancel), "the pending hit wants the group"); + assert!( + !group.handle().abort_unused(Error::Cancel), + "the pending hit wants the group" + ); let mut fetched = hit.await.unwrap(); assert_eq!(&fetched.read_frame().await.unwrap().unwrap().payload[..], b"head"); } @@ -14463,7 +14461,7 @@ mod test { .unwrap(); let group = req.accept(Info::timed()).unwrap(); drop(first.await.unwrap()); - assert!(group.abort_unused(Error::Cancel)); + assert!(group.handle().abort_unused(Error::Cancel)); let retry = consumer.fetch_group(5, None); let req = dynamic diff --git a/rs/moq-net/src/path/mod.rs b/rs/moq-net/src/path/mod.rs index 540909b2875..7280b916ffe 100644 --- a/rs/moq-net/src/path/mod.rs +++ b/rs/moq-net/src/path/mod.rs @@ -62,7 +62,7 @@ impl<'a> AsPath for &'a String { /// A borrowed slice of the path, or a suffix of a shared reference-counted buffer. /// /// The `Shared` variant is what makes owned paths cheap: cloning bumps a refcount and -/// suffix operations (strip_prefix, next_part) only advance `start`, so one allocation +/// suffix operations (strip_prefix) only advance `start`, so one allocation /// serves every copy of a path as it fans out to consumers. #[derive(Clone)] enum Repr<'a> { @@ -250,20 +250,6 @@ impl<'a> Path<'a> { self.as_str().split('/').filter(|part| !part.is_empty()) } - /// Strip the directory component of the path, if any, and return the rest of the path. - pub fn next_part(&'a self) -> Option<(&'a str, Path<'a>)> { - let s = self.as_str(); - if s.is_empty() { - return None; - } - - if let Some(i) = s.find('/') { - Some((&s[..i], self.slice_from(i + 1))) - } else { - Some((s, Path::empty())) - } - } - /// Whether a segment starts with `.`, which hides the path from announce /// discovery below the requested prefix. pub(crate) fn is_hidden(&self) -> bool { @@ -1235,12 +1221,6 @@ mod tests { assert_eq!(stripped.as_str(), "room/broadcast"); assert_eq!(stripped.as_str().as_ptr(), path.as_str()["customer/".len()..].as_ptr()); - // next_part shares the rest as well. - let (dir, rest) = path.next_part().unwrap(); - assert_eq!(dir, "customer"); - let rest = rest.to_owned(); - assert_eq!(rest.as_str().as_ptr(), stripped.as_str().as_ptr()); - // join produces an owned path whose clones share. let joined = path.join("alice"); let joined2 = joined.clone(); diff --git a/rs/moq-net/src/server.rs b/rs/moq-net/src/server.rs index 2e45e751d2e..a13de9735dc 100644 --- a/rs/moq-net/src/server.rs +++ b/rs/moq-net/src/server.rs @@ -106,11 +106,6 @@ impl Server { self } - /// Set both publish and subscribe from one shared [`origin::Producer`]. - pub fn with_origin(self, origin: origin::Producer) -> Self { - self.with_publisher(&origin).with_subscriber(origin) - } - /// Cap what each client can make its session hold. Defaults to [`session::Limits::default`](crate::session::Limits::default). pub fn with_limits(mut self, limits: crate::session::Limits) -> Self { self.limits = limits; diff --git a/rs/moq-net/src/session.rs b/rs/moq-net/src/session.rs index 4a0bafec220..47f5fee479b 100644 --- a/rs/moq-net/src/session.rs +++ b/rs/moq-net/src/session.rs @@ -351,8 +351,8 @@ impl Session { /// Observe a GOAWAY from the peer, telling us to migrate elsewhere. /// - /// [`peek`](goaway::Consumer::peek) is the cheap synchronous check; - /// [`recv`](goaway::Consumer::recv) waits for one. Once a GOAWAY arrives, this + /// [`poll`](goaway::Consumer::poll) it with a no-op waiter for a synchronous check, or + /// wait on it for one. Once a GOAWAY arrives, this /// session's routes cost [`Cost::DRAIN`](crate::origin::Cost::DRAIN): requests /// keep opening on it until a replacement session's route outranks it, and /// existing subscriptions keep flowing until the session closes. diff --git a/rs/moq-net/src/setup.rs b/rs/moq-net/src/setup.rs index 100461caed3..6a1a4945143 100644 --- a/rs/moq-net/src/setup.rs +++ b/rs/moq-net/src/setup.rs @@ -118,8 +118,6 @@ pub struct Token { impl Token { /// Token Type 0: a format the endpoints agreed on out of band, such as a JWT. pub const OUT_OF_BAND: u64 = 0x0; - /// Token Type 1: a Common Access Token (draft-ietf-moq-c4m). - pub const CAT: u64 = 0x1; } /// Draft-17+ unified SETUP message, with the same encoding for both client and server. diff --git a/rs/moq-net/src/stats.rs b/rs/moq-net/src/stats.rs index f67b94e537d..62d3a248043 100644 --- a/rs/moq-net/src/stats.rs +++ b/rs/moq-net/src/stats.rs @@ -641,11 +641,6 @@ impl Traffic { self.stale.add(other.stale); } - /// True while the broadcast is announced (an announce guard is open). - pub fn is_announced(&self) -> bool { - self.announces_started > self.announces_ended - } - /// Distinct sessions currently subscribed (viewers on the egress side). pub fn active_broadcasts(&self) -> u64 { self.broadcasts_started.saturating_sub(self.broadcasts_ended) @@ -656,13 +651,6 @@ impl Traffic { self.subscriptions_started.saturating_sub(self.subscriptions_ended) } - /// All bytes attributable to this slice: payload plus announce overhead. - /// Both inputs are monotonic, so the sum regresses only when the entry was - /// garbage collected and re-created. - pub fn total_bytes(&self) -> u64 { - self.bytes.saturating_add(self.announced_bytes) - } - /// True once every started counter equals its ended counterpart: no guard is /// held, so no more traffic can flow until a new start. pub fn is_idle(&self) -> bool { @@ -840,11 +828,6 @@ impl Tier { &self.0 } - /// True for the default (unprefixed) tier. - pub fn is_default(&self) -> bool { - self.0.is_empty() - } - /// The tier label as used in metrics: empty (`""`) for the default tier, /// otherwise the label (e.g. `"region/sjc"`), as stats frames key it. pub fn as_str(&self) -> &str { @@ -1181,11 +1164,6 @@ impl Registry { self.shared.is_some() } - /// The excluded patterns. See [`Config::exclude`]. - pub fn exclude(&self) -> &Patterns { - &self.exclude - } - /// The shared state, panicking for a disabled registry. Tests build enabled /// registries so this is always present. #[cfg(test)] @@ -1406,16 +1384,6 @@ impl Handle { } } - /// The registry this handle is tied to. - pub fn parent(&self) -> &Registry { - &self.stats - } - - /// The tier this handle bumps into. - pub fn tier(&self) -> &Tier { - &self.tier - } - /// Record a connected session under `label` on this tier. Hold /// the returned guard for the session's lifetime; dropping it bumps /// `sessions_ended`. Counts presence regardless of any data flow, so a diff --git a/rs/moq-net/src/version.rs b/rs/moq-net/src/version.rs index feaa63dda49..4f80533c721 100644 --- a/rs/moq-net/src/version.rs +++ b/rs/moq-net/src/version.rs @@ -84,11 +84,6 @@ impl Version { } } - /// Iterate the names accepted by [`FromStr`]. - pub fn names() -> impl Iterator { - ALL.iter().map(Self::as_str) - } - /// Handshake / config name, e.g. `moq-lite-05` or `moq-transport-22`. pub fn as_str(&self) -> &'static str { match self { @@ -147,28 +142,6 @@ impl Version { } } - /// Parse from ALPN string. - /// - /// Returns `None` for `ALPN_LITE` since multiple versions share - /// that ALPN, requiring SETUP negotiation to determine the version. - pub fn from_alpn(alpn: &str) -> Option { - match alpn { - ALPN_LITE => None, // Multiple versions share this ALPN, need SETUP negotiation - ALPN_LITE_03 => Some(Self::Lite(lite::Version::Lite03)), - ALPN_LITE_04 => Some(Self::Lite(lite::Version::Lite04)), - ALPN_LITE_05 => Some(Self::Lite(lite::Version::Lite05)), - ALPN_LITE_06 => Some(Self::Lite(lite::Version::Lite06)), - ALPN_LITE_07_WIP => Some(Self::Lite(lite::Version::Lite07)), - ALPN_14 => Some(Self::Ietf(ietf::Version::Draft14)), - ALPN_15 => Some(Self::Ietf(ietf::Version::Draft15)), - ALPN_16 => Some(Self::Ietf(ietf::Version::Draft16)), - ALPN_17 => Some(Self::Ietf(ietf::Version::Draft17)), - ALPN_18 => Some(Self::Ietf(ietf::Version::Draft18)), - ALPN_22 => Some(Self::Ietf(ietf::Version::Draft22)), - _ => None, - } - } - /// Returns the ALPN string for this version. pub fn alpn(&self) -> &'static str { match self { @@ -194,14 +167,6 @@ impl Version { Self::Ietf(_) => false, } } - - /// Whether this is an IETF protocol version. - pub fn is_ietf(&self) -> bool { - match self { - Self::Ietf(_) => true, - Self::Lite(_) => false, - } - } } impl fmt::Display for Version { @@ -363,7 +328,6 @@ mod tests { fn default_versions_prefer_lite_06() { let newest = Version::Lite(lite::Version::Lite06); assert_eq!(newest.alpn(), "moq-lite-06"); - assert_eq!(Version::from_alpn("moq-lite-06-wip"), None); assert!("moq-lite-06-wip".parse::().is_err()); assert_eq!(Versions::all().iter().next(), Some(&newest)); assert_eq!(Versions::all().alpns().first(), Some(&newest.alpn())); @@ -376,9 +340,7 @@ mod tests { assert_eq!(wip.alpn(), "moq-lite-07-wip"); assert_eq!(wip.to_string(), "moq-lite-07-wip"); assert_eq!("moq-lite-07-wip".parse::(), Ok(wip)); - assert_eq!(Version::from_alpn("moq-lite-07-wip"), Some(wip)); assert!("moq-lite-07".parse::().is_err()); - assert_eq!(Version::from_alpn("moq-lite-07"), None); assert!(!Versions::all().contains(&wip)); assert!(!ALPNS.contains(&wip.alpn())); diff --git a/rs/moq-net/tests/integration/auth.rs b/rs/moq-net/tests/integration/auth.rs index 584efd62d19..5d3db13f0fe 100644 --- a/rs/moq-net/tests/integration/auth.rs +++ b/rs/moq-net/tests/integration/auth.rs @@ -1941,7 +1941,9 @@ async fn a_handle_backs_one_active_session(version: &'static str) { assert!(matches!(concurrent, Err(Error::Duplicate)), "{:?}", concurrent.err()); first.server.drain().send(moq_net::goaway::Goaway::new()).unwrap(); - first.client.draining().recv().await.expect("the GOAWAY"); + support::Draining::recv(&first.client.draining()) + .await + .expect("the GOAWAY"); let mut second = connect(options()).await; let mut second_tokens = serve(second.requests.take().unwrap(), |_| Some(Grant::all())); let _second_issued = presented(&mut second_tokens).await; diff --git a/rs/moq-net/tests/integration/goaway.rs b/rs/moq-net/tests/integration/goaway.rs index fabd92ec0f5..1cc87bd46a0 100644 --- a/rs/moq-net/tests/integration/goaway.rs +++ b/rs/moq-net/tests/integration/goaway.rs @@ -11,6 +11,7 @@ use crate::support; use std::time::Duration; use moq_net::{Hop, Version, goaway::Goaway}; +use support::Draining; use support::harness::{MockConnectOptions, MockPair, connect_mock}; use support::mock::create_mock_session_pair; @@ -82,8 +83,11 @@ async fn a_stalled_lite_goaway_is_still_closed_at_the_deadline() { #[moq_clock::test] async fn a_stalled_ietf_goaway_is_still_closed_at_the_deadline() { - for name in Version::names().filter(|name| name.starts_with("moq-transport-")) { - closes_stalled_goaway(name.parse().unwrap(), true).await; + for version in support::every_version() + .into_iter() + .filter(|version| !version.is_lite()) + { + closes_stalled_goaway(version, true).await; } } diff --git a/rs/moq-net/tests/integration/lazy_discovery.rs b/rs/moq-net/tests/integration/lazy_discovery.rs index defb57c2b74..476469ea66d 100644 --- a/rs/moq-net/tests/integration/lazy_discovery.rs +++ b/rs/moq-net/tests/integration/lazy_discovery.rs @@ -98,7 +98,7 @@ impl Lazy { /// Read until the stream syncs, returning the prefixes it then holds. async fn synced(announced: &mut announce::Consumer) -> Vec { let mut seen = Vec::new(); - while !announced.is_synced() { + while !announced.poll_synced(&moq_net::kio::Waiter::noop()).is_ready() { let event = moq_clock::timeout(SETTLE, announced.next()) .await .expect("synced in time") diff --git a/rs/moq-net/tests/integration/rejoin.rs b/rs/moq-net/tests/integration/rejoin.rs index 11bdfee5e44..3efe02589fa 100644 --- a/rs/moq-net/tests/integration/rejoin.rs +++ b/rs/moq-net/tests/integration/rejoin.rs @@ -340,7 +340,7 @@ async fn rejoin_goes_live_without_the_join_head() { /// copy idle, or it takes the cached group as the live edge. #[moq_clock::test] async fn rejoin_during_the_cancel_skips_the_cache() { - for version in Version::names() { + for version in support::every_version() { moq_clock::timeout(TEST_TIMEOUT, async { let publisher = produce_origin(1); let relay = produce_origin(2); @@ -349,7 +349,7 @@ async fn rejoin_during_the_cancel_skips_the_cache() { let track = broadcast.create_track("video", timed()).unwrap(); broadcast.announce(Default::default()).unwrap(); - let mut options = MockConnectOptions::new(version.parse::().unwrap()); + let mut options = MockConnectOptions::new(version); options.server_publish = Some(publisher.consume()); options.client_subscribe = Some(relay.clone()); options.latency = Duration::from_millis(50); @@ -401,7 +401,10 @@ async fn rejoin_waits_for_the_answers_first_frame() { // Lite01/02 carry no max age, so the publisher serves the relay's cached head again and // that answer is what makes the cache current. let unbudgeted = ["moq-lite-01", "moq-lite-02"]; - for version in Version::names().filter(|version| !unbudgeted.contains(version)) { + for version in support::every_version() + .into_iter() + .filter(|version| !unbudgeted.contains(&version.as_str())) + { moq_clock::timeout(TEST_TIMEOUT, async { let publisher = produce_origin(1); let relay = produce_origin(2); @@ -410,7 +413,7 @@ async fn rejoin_waits_for_the_answers_first_frame() { let track = broadcast.create_track("video", timed()).unwrap(); broadcast.announce(Default::default()).unwrap(); - let mut options = MockConnectOptions::new(version.parse::().unwrap()); + let mut options = MockConnectOptions::new(version); options.server_publish = Some(publisher.consume()); options.client_subscribe = Some(relay.clone()); let pair = connect_mock(options).await; diff --git a/rs/moq-net/tests/integration/stream_credit.rs b/rs/moq-net/tests/integration/stream_credit.rs index 2fac6daaf9a..af927a68852 100644 --- a/rs/moq-net/tests/integration/stream_credit.rs +++ b/rs/moq-net/tests/integration/stream_credit.rs @@ -29,7 +29,7 @@ async fn one_stream_credit_serves_every_backlogged_group() { let mut options = MockConnectOptions::new(version); // Draft 22 keeps its SETUP stream open, so it needs one more for groups. - options.transport.max_concurrent_uni_streams = Some(1 + usize::from(version.is_ietf())); + options.transport.max_concurrent_uni_streams = Some(1 + usize::from(!version.is_lite())); options.server_publish = Some(publisher.consume()); options.client_subscribe = Some(viewer.clone()); let _pair = connect_mock(options).await; diff --git a/rs/moq-net/tests/integration/support/mod.rs b/rs/moq-net/tests/integration/support/mod.rs index dee03d04c5e..8ca825b8d21 100644 --- a/rs/moq-net/tests/integration/support/mod.rs +++ b/rs/moq-net/tests/integration/support/mod.rs @@ -5,7 +5,7 @@ pub mod harness; pub mod mock; -use moq_net::track; +use moq_net::{goaway, track}; /// Wait until anyone wants the track, returning the group they want. #[allow(dead_code, reason = "shared by several test binaries, each using only part of it")] @@ -23,3 +23,36 @@ pub async fn unwanted(mut demand: track::Demand) -> moq_net::Result<()> { while demand.wanted_changed().await?.is_some() {} Ok(()) } + +/// Reads of the peer's GOAWAY over [`goaway::Consumer::poll`]. +#[allow(dead_code, reason = "shared by several test binaries, each using only part of it")] +pub trait Draining { + /// The peer's GOAWAY, or `None` if it has not sent one. + fn peek(&self) -> Option; + + /// Wait for the peer's GOAWAY, or `None` if the session closes without one. + async fn recv(&self) -> Option; +} + +impl Draining for goaway::Consumer { + fn peek(&self) -> Option { + match self.poll(&moq_net::kio::Waiter::noop()) { + std::task::Poll::Ready(Ok(goaway)) => Some(goaway), + _ => None, + } + } + + async fn recv(&self) -> Option { + moq_net::kio::wait(|waiter| self.poll(waiter)).await.ok() + } +} + +/// Every version this crate speaks, found by wire code, including the opt-in `-wip` +/// drafts that [`moq_net::Versions::all`] leaves out. +#[allow(dead_code, reason = "shared by several test binaries, each using only part of it")] +pub fn every_version() -> Vec { + (0xff0dad00..=0xff0dadff) + .chain(0xff000000..=0xff0000ff) + .filter_map(moq_net::Version::from_code) + .collect() +} diff --git a/rs/moq-net/tests/integration/synced.rs b/rs/moq-net/tests/integration/synced.rs index fe90b215aa3..b6696e3e197 100644 --- a/rs/moq-net/tests/integration/synced.rs +++ b/rs/moq-net/tests/integration/synced.rs @@ -24,7 +24,7 @@ fn produce_origin(hop: u64) -> origin::Producer { /// Read events until the consumer syncs, returning the prefixes then announced. async fn read_until_synced(announced: &mut announce::Consumer) -> BTreeSet { let mut seen = BTreeSet::new(); - while !announced.is_synced() { + while !announced.poll_synced(&moq_net::kio::Waiter::noop()).is_ready() { let event = moq_clock::timeout(SETTLE, announced.next()) .await .expect("synced in time") @@ -141,5 +141,5 @@ async fn a_cold_relay_syncs_then_forwards_upstream_routes_as_live() { let _upstream = connect_mock(options).await; let event = announced.next().await.unwrap(); assert!(matches!(event, announce::Event::Start(route) if route.prefix.as_str() == "room/cam")); - assert!(announced.is_synced()); + assert!(announced.poll_synced(&moq_net::kio::Waiter::noop()).is_ready()); } diff --git a/rs/moq-net/tests/integration/track_stream_demand.rs b/rs/moq-net/tests/integration/track_stream_demand.rs index 21d941b643d..9ef71922418 100644 --- a/rs/moq-net/tests/integration/track_stream_demand.rs +++ b/rs/moq-net/tests/integration/track_stream_demand.rs @@ -44,25 +44,6 @@ async fn link(version: Version, publisher: &origin::Producer, subscriber: &origi connect_mock(options).await } -/// Every demand edge a `broadcast::Demand` reports from here on, `true` for used. -macro_rules! record { - ($demand:expr) => {{ - let edges = Rc::new(RefCell::new(Vec::new())); - let log = edges.clone(); - let demand = $demand; - drop(moq_clock::sim::spawn(async move { - while demand.used().await.is_ok() { - log.borrow_mut().push(true); - if demand.unused().await.is_err() { - break; - } - log.borrow_mut().push(false); - } - })); - edges - }}; -} - /// Every demand edge a `track::Demand` reports from here on, `true` for wanted. macro_rules! record_track { ($demand:expr) => {{ @@ -83,8 +64,8 @@ macro_rules! record_track { } /// A reader subscribes through `relays` relays and stays: the publisher, which starts -/// the track on demand, is asked for it once, and its broadcast and track each see exactly -/// one `used` edge, then `unused` once the reader leaves. +/// the track on demand, is asked for it once, and its track sees exactly one `used` edge, +/// then `unused` once the reader leaves. /// /// With `reorder`, the reader's SUBSCRIBE stream reaches the first hop well after its /// TRACK stream's data, as QUIC may deliver them, so a hop that let go of the TRACK @@ -103,8 +84,6 @@ async fn one_used_edge(version: &str, relays: usize, reorder: bool) { let broadcast = nodes[0].create_broadcast("room").unwrap(); let mut dynamic = broadcast.dynamic(); broadcast.announce(Default::default()).unwrap(); - let broadcast_edges = record!(broadcast.demand()); - // Serve every request for the track, keeping what it started. let requests = Rc::new(RefCell::new(Vec::new())); let track_edges = Rc::new(RefCell::new(None)); @@ -156,7 +135,6 @@ async fn one_used_edge(version: &str, relays: usize, reorder: bool) { moq_clock::sleep(Duration::from_secs(2)).await; assert_eq!(requests.borrow().len(), 1, "{label}: asked for the track again"); let track_edges = track_edges.borrow().clone().expect("the track was requested"); - assert_eq!(*broadcast_edges.borrow(), [true], "{label}: broadcast demand"); assert_eq!(*track_edges.borrow(), [true], "{label}: track demand"); // Holding the TRACK stream must not outlive the reader. @@ -168,7 +146,6 @@ async fn one_used_edge(version: &str, relays: usize, reorder: bool) { .unwrap_or_else(|_| panic!("{label}: demand outlived the reader")) .unwrap(); moq_clock::sleep(Duration::from_millis(100)).await; - assert_eq!(*broadcast_edges.borrow(), [true, false], "{label}: broadcast demand"); assert_eq!(*track_edges.borrow(), [true, false], "{label}: track demand"); publisher.abort(); } @@ -191,8 +168,7 @@ async fn through_a_relay() { /// A bare track handle is no demand, through `relays` relays: holding one asks the /// publisher for nothing, a query on it does until it is dropped, and once a subscription -/// made from it ends, the publisher and the reader's own broadcast both read unused while -/// the handle is still held. +/// made from it ends, the publisher reads unused while the handle is still held. async fn bare_handle_is_lazy(version: &str, relays: usize) { let version: Version = version.parse().unwrap(); let label = format!("{version} over {relays} relays"); @@ -206,8 +182,6 @@ async fn bare_handle_is_lazy(version: &str, relays: usize) { let broadcast = nodes[0].create_broadcast("room").unwrap(); let mut dynamic = broadcast.dynamic(); broadcast.announce(Default::default()).unwrap(); - let broadcast_edges = record!(broadcast.demand()); - let requests = Rc::new(RefCell::new(Vec::new())); let served = requests.clone(); let publisher = moq_clock::sim::spawn(async move { @@ -232,7 +206,6 @@ async fn bare_handle_is_lazy(version: &str, relays: usize) { requests.borrow().is_empty(), "{label}: a bare handle asked for the track" ); - assert!(broadcast_edges.borrow().is_empty(), "{label}: a bare handle is demand"); // A held query reaches the publisher, and holds the track there until dropped. let mut query = handle.query(); @@ -273,10 +246,6 @@ async fn bare_handle_is_lazy(version: &str, relays: usize) { .await .unwrap_or_else(|_| panic!("{label}: the publisher still reads the track as used")) .unwrap(); - assert!( - !remote.demand().is_used(), - "{label}: the reader's broadcast is still used" - ); drop(handle); publisher.abort(); } diff --git a/rs/moq-net/tests/integration/unannounce_release.rs b/rs/moq-net/tests/integration/unannounce_release.rs index 69cb03f9e2e..551289eb499 100644 --- a/rs/moq-net/tests/integration/unannounce_release.rs +++ b/rs/moq-net/tests/integration/unannounce_release.rs @@ -46,7 +46,7 @@ async fn release(version: Version) { #[moq_clock::test] async fn unannounced_remote_track_releases_demand() { - for name in Version::names() { - release(name.parse().unwrap()).await; + for version in support::every_version() { + release(version).await; } } diff --git a/rs/moq-net/tests/loom.rs b/rs/moq-net/tests/loom.rs index ba0807b9ebf..fdb438069bd 100644 --- a/rs/moq-net/tests/loom.rs +++ b/rs/moq-net/tests/loom.rs @@ -206,50 +206,6 @@ fn subscriber_wakes_parked_demand() { }); } -/// A publisher parked on `broadcast::Demand::used` while one reader hands over to the -/// next: the last reader leaves and a new one arrives straight from the track producer, -/// which never touches the broadcast's own state. Whatever the publisher's poll saw, it -/// must be woken for the newcomer. -#[test] -fn broadcast_demand_sees_a_reader_return() { - model(|| { - let broadcast = broadcast::Info::new().produce(); - let consumer = broadcast.consume(); - let video = broadcast.create_track("video", None).expect("create track"); - let demand = broadcast.demand(); - let first = consumer.track("video").expect("track").subscribe(None); - - let reader = thread::spawn(move || { - drop(first); - let next = video.subscribe(None); - // Returned, so neither the reader nor the track goes before the join. - (video, next) - }); - - block_on(demand.used()).expect("the returning reader was missed"); - drop(reader.join().unwrap()); - }); -} - -/// A publisher parked on `broadcast::Demand::unused` while a reader comes and goes -/// straight through the track producer: its leaving must wake the publisher. -#[test] -fn broadcast_demand_sees_a_reader_leave() { - model(|| { - let broadcast = broadcast::Info::new().produce(); - let video = broadcast.create_track("video", None).expect("create track"); - let demand = broadcast.demand(); - - let reader = thread::spawn(move || { - drop(video.subscribe(None)); - video - }); - - block_on(demand.unused()).expect("the reader leaving was missed"); - drop(reader.join().unwrap()); - }); -} - /// A reader taking and dropping a track around the front's demand poll. The front must /// see it go: one that missed the edge keeps its source subscribed for nobody, which /// shows up here as a deadlock waiting for the source to go unread. diff --git a/rs/moq-relay/src/shutdown.rs b/rs/moq-relay/src/shutdown.rs index 8e25d9c9968..cec15abfa13 100644 --- a/rs/moq-relay/src/shutdown.rs +++ b/rs/moq-relay/src/shutdown.rs @@ -569,7 +569,8 @@ mod tests { let (_lease, consumer) = moq_auth::lease::Producer::new(moq_auth::Grant::new(all(), all())); admission.grant(consumer); let connection = connecting.await.unwrap().unwrap(); - let goaway = connection.draining().unwrap().recv().await.unwrap(); + let draining = connection.draining().unwrap(); + let goaway = moq_net::kio::wait(|waiter| draining.poll(waiter)).await.unwrap(); assert_eq!(goaway.uri(), ""); connection.closed().await.unwrap(); running.await.unwrap().unwrap(); diff --git a/rs/moq-relay/src/web.rs b/rs/moq-relay/src/web.rs index 56a9ff01047..5b51d48d851 100644 --- a/rs/moq-relay/src/web.rs +++ b/rs/moq-relay/src/web.rs @@ -2479,7 +2479,10 @@ mod tests { } None => socket_step(upgrade_while_draining(web.stream, true)).await, }; - let goaway = socket_step(session.draining().recv()).await.unwrap(); + let draining = session.draining(); + let goaway = socket_step(moq_net::kio::wait(|waiter| draining.poll(waiter))) + .await + .unwrap(); assert_eq!(goaway.uri(), ""); session.abort(moq_net::Error::Cancel); socket_step(web.running).await.unwrap().unwrap(); @@ -2583,7 +2586,9 @@ mod tests { web.trigger.start(); let session = socket_step(upgrade_while_draining(web.stream, false)).await; drop(pending); - assert_eq!(socket_step(session.draining().recv()).await.unwrap().uri(), ""); + let draining = session.draining(); + let goaway = socket_step(moq_net::kio::wait(|waiter| draining.poll(waiter))).await; + assert_eq!(goaway.unwrap().uri(), ""); session.abort(moq_net::Error::Cancel); socket_step(web.running).await.unwrap().unwrap(); } diff --git a/rs/moq-relay/src/websocket.rs b/rs/moq-relay/src/websocket.rs index 447bec62e57..7b0b409ce3d 100644 --- a/rs/moq-relay/src/websocket.rs +++ b/rs/moq-relay/src/websocket.rs @@ -383,13 +383,11 @@ mod tests { fn supported_subprotocols_lists_full_matrix() { // Guard the literals: they must stay the IETF draft-18-and-newer ALPNs // (wire 0xff000012 and 0xff000016). - assert_eq!( - QMUX01_ONLY_ALPNS - .iter() - .map(|&a| moq_net::Version::from_alpn(a).map(|v| v.code())) - .collect::>(), - vec![Some(0xff000012), Some(0xff000016)] - ); + let drafts: Vec<_> = [0xff000012, 0xff000016] + .into_iter() + .map(|code| moq_net::Version::from_code(code).unwrap().alpn()) + .collect(); + assert_eq!(QMUX01_ONLY_ALPNS, drafts); let list = supported_subprotocols(moq_net::ALPNS); diff --git a/rs/moq-relay/tests/integration/auth_lifetime.rs b/rs/moq-relay/tests/integration/auth_lifetime.rs index 44e67f7958a..77b49f53efd 100644 --- a/rs/moq-relay/tests/integration/auth_lifetime.rs +++ b/rs/moq-relay/tests/integration/auth_lifetime.rs @@ -1074,7 +1074,7 @@ async fn the_drain_waits_for_a_closed_session_to_report_its_end() { // Keep the deadline from racing the client receiving GOAWAY under load. tokio::time::pause(); trigger.start(); - tokio::time::timeout(TIMEOUT, client.draining().recv()) + tokio::time::timeout(TIMEOUT, moq_net::kio::wait(|waiter| client.draining().poll(waiter))) .await .expect("GOAWAY timeout") .expect("a GOAWAY"); diff --git a/rs/moq-relay/tests/integration/goaway_cluster.rs b/rs/moq-relay/tests/integration/goaway_cluster.rs index cdf9b7328c7..f5fc86e0b3e 100644 --- a/rs/moq-relay/tests/integration/goaway_cluster.rs +++ b/rs/moq-relay/tests/integration/goaway_cluster.rs @@ -688,14 +688,19 @@ async fn cluster_diamond_goaway_seamless_failover_inner() { // ── no GOAWAY cascade to the downstream subscriber ─────────────────── assert!( - sub_connection.draining().expect("connected").peek().is_none(), + sub_connection + .draining() + .expect("connected") + .poll(&moq_net::kio::Waiter::noop()) + .is_pending(), "BOTTOM must not propagate the upstream GOAWAY downstream" ); // The async observer must stay pending too (bounded probe: everything // above already synchronized, so 500ms of silence is decisive). + let draining = sub_connection.draining().expect("connected"); let leaked = tokio::time::timeout( Duration::from_millis(500), - sub_connection.draining().expect("connected").recv(), + moq_net::kio::wait(|waiter| draining.poll(waiter)), ) .await; assert!( diff --git a/rs/moq-relay/tests/integration/runtime_uring.rs b/rs/moq-relay/tests/integration/runtime_uring.rs index 499e1fc18dd..3e98f82740b 100644 --- a/rs/moq-relay/tests/integration/runtime_uring.rs +++ b/rs/moq-relay/tests/integration/runtime_uring.rs @@ -404,7 +404,8 @@ async fn uring_workers_drain_on_the_trigger() { let arriving = tokio::spawn(connect(client, url)); let pending = admissions.next().await.expect("pending admission"); trigger.start(); - let goaway = tokio::time::timeout(TIMEOUT, established.draining().expect("connected").recv()) + let draining = established.draining().expect("connected"); + let goaway = tokio::time::timeout(TIMEOUT, moq_net::kio::wait(|waiter| draining.poll(waiter))) .await .expect("no GOAWAY after the trigger") .expect("session closed without a GOAWAY"); @@ -417,10 +418,14 @@ async fn uring_workers_drain_on_the_trigger() { let (_arrival_lease, arrival_consumer) = moq_auth::lease::Producer::new(moq_auth::Grant::new(all(), all())); pending.grant(arrival_consumer); let arrival = arriving.await.expect("arrival task"); - let goaway = tokio::time::timeout(Duration::from_secs(1), arrival.draining().expect("connected").recv()) - .await - .expect("an arrival mid-drain was not sent a GOAWAY") - .expect("arrival closed without a GOAWAY"); + let draining = arrival.draining().expect("connected"); + let goaway = tokio::time::timeout( + Duration::from_secs(1), + moq_net::kio::wait(|waiter| draining.poll(waiter)), + ) + .await + .expect("an arrival mid-drain was not sent a GOAWAY") + .expect("arrival closed without a GOAWAY"); assert_eq!(goaway.uri(), "", "expected a reconnect-to-me GOAWAY"); tokio::time::timeout(TIMEOUT, arrival.closed()) .await diff --git a/rs/moq-relay/tests/integration/shutdown_signal.rs b/rs/moq-relay/tests/integration/shutdown_signal.rs index ee214e825cf..63e2a81d90b 100644 --- a/rs/moq-relay/tests/integration/shutdown_signal.rs +++ b/rs/moq-relay/tests/integration/shutdown_signal.rs @@ -153,10 +153,13 @@ async fn sigint_drains_sessions_before_exiting_inner() { // handler, registered above. assert_eq!(unsafe { libc::raise(libc::SIGINT) }, 0, "failed to raise SIGINT"); - let goaway = tokio::time::timeout(Duration::from_secs(5), draining.recv()) - .await - .expect("no GOAWAY within 5s of SIGINT") - .expect("session closed without a GOAWAY"); + let goaway = tokio::time::timeout( + Duration::from_secs(5), + moq_net::kio::wait(|waiter| draining.poll(waiter)), + ) + .await + .expect("no GOAWAY within 5s of SIGINT") + .expect("session closed without a GOAWAY"); // Empty URI: the relay is restarting, not moving. The window itself is the // sender's own timer here (only moq-transport draft-17+ puts it on the wire), // so it is the elapsed time below that proves it was honored. @@ -192,10 +195,13 @@ async fn a_signal_right_after_ready_drains_inner() { // for SIGTERM, so in a process of its own (nextest) the relay's handler is all // that keeps this from killing the test. assert_eq!(unsafe { libc::raise(libc::SIGTERM) }, 0, "failed to raise SIGTERM"); - let goaway = tokio::time::timeout(Duration::from_secs(5), draining.recv()) - .await - .expect("no GOAWAY within 5s of SIGTERM") - .expect("session closed without a GOAWAY"); + let goaway = tokio::time::timeout( + Duration::from_secs(5), + moq_net::kio::wait(|waiter| draining.poll(waiter)), + ) + .await + .expect("no GOAWAY within 5s of SIGTERM") + .expect("session closed without a GOAWAY"); assert_eq!(goaway.uri(), "", "expected a reconnect-to-me GOAWAY"); tokio::time::timeout(Duration::from_secs(15), run) .await @@ -256,7 +262,9 @@ async fn an_embedder_owns_the_signals_inner() { // The signal is the embedder's: the relay keeps serving without a GOAWAY. assert!( - tokio::time::timeout(DRAIN_TIMEOUT / 2, draining.recv()).await.is_err(), + tokio::time::timeout(DRAIN_TIMEOUT / 2, moq_net::kio::wait(|waiter| draining.poll(waiter))) + .await + .is_err(), "the relay drained on a signal the embedder owns" ); assert!(!run.is_finished(), "the relay exited on a signal the embedder owns"); @@ -264,10 +272,13 @@ async fn an_embedder_owns_the_signals_inner() { // Until the embedder decides. let started = std::time::Instant::now(); trigger.start(); - let goaway = tokio::time::timeout(Duration::from_secs(5), draining.recv()) - .await - .expect("no GOAWAY within 5s of the trigger") - .expect("session closed without a GOAWAY"); + let goaway = tokio::time::timeout( + Duration::from_secs(5), + moq_net::kio::wait(|waiter| draining.poll(waiter)), + ) + .await + .expect("no GOAWAY within 5s of the trigger") + .expect("session closed without a GOAWAY"); assert_eq!(goaway.uri(), "", "expected a reconnect-to-me GOAWAY"); tokio::time::timeout(Duration::from_secs(15), run) @@ -302,10 +313,13 @@ async fn a_session_arriving_mid_drain_gets_what_is_left_inner() { let _straggler = straggler(port).await; trigger.start(); let deadline = std::time::Instant::now() + DRAIN_TIMEOUT; - let goaway = tokio::time::timeout(Duration::from_secs(5), draining.recv()) - .await - .expect("no GOAWAY within 5s of the trigger") - .expect("session closed without a GOAWAY"); + let goaway = tokio::time::timeout( + Duration::from_secs(5), + moq_net::kio::wait(|waiter| draining.poll(waiter)), + ) + .await + .expect("no GOAWAY within 5s of the trigger") + .expect("session closed without a GOAWAY"); assert_eq!(goaway.uri(), "", "expected a reconnect-to-me GOAWAY"); let timeout = goaway.timeout().expect("the GOAWAY carries its deadline"); assert!( @@ -318,10 +332,14 @@ async fn a_session_arriving_mid_drain_gets_what_is_left_inner() { tokio::time::sleep(DRAIN_TIMEOUT / 2).await; let dialed = std::time::Instant::now(); let arrival = connect(&client, port).await; - let goaway = tokio::time::timeout(Duration::from_secs(1), arrival.draining().expect("connected").recv()) - .await - .expect("an arrival mid-drain was not sent a GOAWAY") - .expect("arrival closed without a GOAWAY"); + let draining = arrival.draining().expect("connected"); + let goaway = tokio::time::timeout( + Duration::from_secs(1), + moq_net::kio::wait(|waiter| draining.poll(waiter)), + ) + .await + .expect("an arrival mid-drain was not sent a GOAWAY") + .expect("arrival closed without a GOAWAY"); assert_eq!(goaway.uri(), "", "expected a reconnect-to-me GOAWAY"); // Only what is left of the window, not a fresh one that would outlive the diff --git a/rs/moq-stats/src/aggregate.rs b/rs/moq-stats/src/aggregate.rs index 4cb0c7c3a85..5362551d7f0 100644 --- a/rs/moq-stats/src/aggregate.rs +++ b/rs/moq-stats/src/aggregate.rs @@ -2024,7 +2024,7 @@ mod tests { let mut traffic = traffic(&agg); let frame = read_until_bytes(&mut traffic, "acme/room", 100).await; let snap = frame.get("acme/room").expect("entry"); - assert!(snap.is_announced()); + assert!(snap.announces_started > snap.announces_ended); assert_eq!(snap.active_broadcasts(), 2); assert_eq!(snap.active_subscriptions(), 3); @@ -2035,7 +2035,7 @@ mod tests { let frame = traffic.next().await.expect("read").expect("frame"); let snap = frame.get("acme/room").expect("entry"); assert_eq!(snap.bytes, 100, "cumulative totals stay"); - assert!(!snap.is_announced(), "no phantom announcement"); + assert_eq!(snap.announces_started, snap.announces_ended, "no phantom announcement"); assert_eq!(snap.active_broadcasts(), 0, "no phantom broadcasts"); assert_eq!(snap.active_subscriptions(), 0, "no phantom subscriptions"); } diff --git a/rs/moq-stats/src/produce.rs b/rs/moq-stats/src/produce.rs index 3ebca806820..e587d4562d4 100644 --- a/rs/moq-stats/src/produce.rs +++ b/rs/moq-stats/src/produce.rs @@ -2166,7 +2166,10 @@ mod tests { assert_eq!(totals.publisher.subscriptions_started, 2); assert_eq!(totals.publisher.active_subscriptions(), 0); assert_eq!(totals.subscriber.bytes, 1500, "the publisher's ingress"); - assert!(!totals.subscriber.is_announced(), "the publisher left"); + assert_eq!( + totals.subscriber.announces_started, totals.subscriber.announces_ended, + "the publisher left" + ); assert_eq!(totals.sessions.sessions_started, 2); assert_eq!(totals.sessions.active(), 1, "the publisher's session left with it"); } diff --git a/rs/moq-tokio/src/client.rs b/rs/moq-tokio/src/client.rs index a7575963533..410fa067362 100644 --- a/rs/moq-tokio/src/client.rs +++ b/rs/moq-tokio/src/client.rs @@ -912,7 +912,12 @@ mod tests { #[test] fn test_cli_version_help_lists_every_parseable_name() { let help = Cli::render_help(Cli::command(), true).expect("long help"); - for name in moq_net::Version::names() { + // Every wire code this crate parses, including the opt-in `-wip` drafts. + let codes = (0xff0dad00..=0xff0dadff).chain(0xff000000..=0xff0000ff); + for name in codes + .filter_map(moq_net::Version::from_code) + .map(|version| version.as_str()) + { assert!(help.contains(name), "missing {name} from --connect-version help"); } } diff --git a/rs/moq-tokio/src/connection.rs b/rs/moq-tokio/src/connection.rs index 24eb9cf367a..5d5c2d62f2f 100644 --- a/rs/moq-tokio/src/connection.rs +++ b/rs/moq-tokio/src/connection.rs @@ -2361,7 +2361,7 @@ mod tests { assert_eq!(out_rx.peek(), Some(moq_net::bandwidth::Rate::from_bps(9_000))); // Closing the source is what retires the arm, so we stop polling a dead one. - src.abort(moq_net::Error::Cancel).unwrap(); + drop(src); poll_forward(&mut bw, &out, &waiter); assert!(bw.is_none()); } @@ -2445,11 +2445,6 @@ mod tests { source.is_used(), "new demand must be forwarded without unrelated activity" ); - out.abort(moq_net::Error::Cancel).unwrap(); - forward.poll(&out, &kio::Waiter::noop(), || { - panic!("a closed output must not acquire a source") - }); - assert!(!source.is_used()); } const WALK_TIMEOUT: Duration = Duration::from_secs(30); @@ -3376,7 +3371,9 @@ mod tests { .expect("the drained session stopped serving") .unwrap(); assert!( - connection.draining().and_then(|goaway| goaway.peek()).is_some(), + connection + .draining() + .is_some_and(|goaway| matches!(goaway.poll(&kio::Waiter::noop()), Poll::Ready(Ok(_)))), "the GOAWAY never arrived" ); assert_eq!( diff --git a/rs/moq-tokio/src/server.rs b/rs/moq-tokio/src/server.rs index cf7369c3224..51fde53c749 100644 --- a/rs/moq-tokio/src/server.rs +++ b/rs/moq-tokio/src/server.rs @@ -2554,7 +2554,12 @@ mod tests { } let help = Cli::render_help(Cli::command(), true).unwrap(); - for name in moq_net::Version::names() { + // Every wire code this crate parses, including the opt-in `-wip` drafts. + let codes = (0xff0dad00..=0xff0dadff).chain(0xff000000..=0xff0000ff); + for name in codes + .filter_map(moq_net::Version::from_code) + .map(|version| version.as_str()) + { assert!(help.contains(name), "missing {name} from --server-version help"); } } diff --git a/rs/moq-tokio/src/websocket.rs b/rs/moq-tokio/src/websocket.rs index 1500c4aeb04..eaf5ee1004e 100644 --- a/rs/moq-tokio/src/websocket.rs +++ b/rs/moq-tokio/src/websocket.rs @@ -969,13 +969,11 @@ mod tests { fn moqt_18_and_newer_pin_to_qmux01() { // The literals in `qmux_versions_for` must stay the IETF draft ALPNs; // otherwise the pin silently stops matching. - assert_eq!( - QMUX01_ONLY_ALPNS - .iter() - .map(|&a| moq_net::Version::from_alpn(a).map(|v| v.code())) - .collect::>(), - vec![Some(0xff000012), Some(0xff000016)] - ); + let drafts: Vec<_> = [0xff000012, 0xff000016] + .into_iter() + .map(|code| moq_net::Version::from_code(code).unwrap().alpn()) + .collect(); + assert_eq!(QMUX01_ONLY_ALPNS, drafts); for &alpn in QMUX01_ONLY_ALPNS { assert_eq!(qmux_versions_for(alpn), &[qmux::Version::QMux01]); } diff --git a/rs/moq-tokio/tests/integration/broadcast.rs b/rs/moq-tokio/tests/integration/broadcast.rs index 2389afda82f..da8edbd1104 100644 --- a/rs/moq-tokio/tests/integration/broadcast.rs +++ b/rs/moq-tokio/tests/integration/broadcast.rs @@ -1153,7 +1153,13 @@ async fn broadcast_rejoin_skips_a_stale_warm_cache() { "moq-lite-04", "moq-lite-05", ]; - for version in moq_net::Version::names().filter(|version| !pre06.contains(version)) { + let every = (0xff0dad00..=0xff0dadff) + .chain(0xff000000..=0xff0000ff) + .filter_map(moq_net::Version::from_code); + for version in every + .map(|version| version.as_str()) + .filter(|version| !pre06.contains(version)) + { rejoin_skips_a_stale_warm_cache(version).await; } } @@ -1278,7 +1284,11 @@ async fn rejoin_skips_a_stale_warm_cache(version: &str) { #[tracing_test::traced_test] #[tokio::test] async fn broadcast_rejoin_replays_a_current_warm_cache() { - for version in moq_net::Version::names() { + for version in (0xff0dad00..=0xff0dadff) + .chain(0xff000000..=0xff0000ff) + .filter_map(moq_net::Version::from_code) + .map(|version| version.as_str()) + { for open in [false, true] { rejoin_replays_a_current_warm_cache(version, open).await; } @@ -3396,12 +3406,15 @@ async fn goaway_test(scheme: &str, version: &str, expect_wire_timeout: bool) { // Trigger the drain and observe the GOAWAY. start_drain_tx.send(()).expect("send drain signal"); - let goaway = tokio::time::timeout(TIMEOUT, draining.recv()) + let goaway = tokio::time::timeout(TIMEOUT, moq_net::kio::wait(|waiter| draining.poll(waiter))) .await .expect("goaway timed out") .expect("session closed before GOAWAY"); assert_eq!(goaway.uri(), "https://elsewhere.example/"); - assert!(draining.peek().is_some()); + assert!(matches!( + draining.poll(&moq_net::kio::Waiter::noop()), + std::task::Poll::Ready(Ok(_)) + )); if expect_wire_timeout { assert_eq!( goaway.timeout(), @@ -3489,7 +3502,8 @@ async fn goaway_timeout_force_close_moq_transport_22_quic() { .expect("client connect failed"); // Observe the GOAWAY but do NOT leave. - let goaway = tokio::time::timeout(TIMEOUT, connection.draining().expect("connected").recv()) + let draining = connection.draining().expect("connected"); + let goaway = tokio::time::timeout(TIMEOUT, moq_net::kio::wait(|waiter| draining.poll(waiter))) .await .expect("goaway timed out") .expect("session closed before GOAWAY"); diff --git a/rs/moq-transcode/src/active.rs b/rs/moq-transcode/src/active.rs index 7a729fdabfa..5d8ee6a49dc 100644 --- a/rs/moq-transcode/src/active.rs +++ b/rs/moq-transcode/src/active.rs @@ -2,8 +2,8 @@ //! //! Nothing is encoded until a consumer asks for a rung, so a transcoder that is //! publishing a catalog and a transcoder that is saturating a GPU look identical -//! from the outside. Broadcast demand ([`moq_net::broadcast::Demand`]) closes -//! half the gap: it says *someone* is watching. This module closes the other +//! from the outside. Track demand ([`moq_net::track::Demand`]) closes half the +//! gap: it says *someone* is watching. This module closes the other //! half by naming *which* renditions are being produced, and counting what each //! one produced, which is what a caller pricing or admitting the work needs. //! diff --git a/swift/Sources/Moq/Epoch.swift b/swift/Sources/Moq/Epoch.swift index 4845249640f..6ad94122326 100644 --- a/swift/Sources/Moq/Epoch.swift +++ b/swift/Sources/Moq/Epoch.swift @@ -8,10 +8,3 @@ import MoqFFI public func mintEpoch() -> String { moqMintEpoch() } - -/// The wall-clock time an epoch encodes, to the millisecond. Throws unless -/// `epoch` is a lowercase hyphenated UUIDv7. -public func epochTime(_ epoch: String) throws -> Date { - let ms = try moqEpochTimeMs(epoch: epoch) - return Date(timeIntervalSince1970: Double(ms) / 1000) -} diff --git a/swift/Tests/MoqTests/SmokeTests.swift b/swift/Tests/MoqTests/SmokeTests.swift index 30277fe7ed5..4f02156cbdf 100644 --- a/swift/Tests/MoqTests/SmokeTests.swift +++ b/swift/Tests/MoqTests/SmokeTests.swift @@ -216,7 +216,6 @@ final class SmokeTests: XCTestCase { let broadcast = try origin.createBroadcast(path: "epoch") let first = mintEpoch() - XCTAssertEqual(try epochTime("018bcfe5-6800-7000-8000-000000000000"), Date(timeIntervalSince1970: 1_700_000_000)) try broadcast.announce(route: AnnounceRoute(epoch: first)) guard case .start(let start) = try await nextRoute(announced) else { return XCTFail("expected an announcement") @@ -230,7 +229,7 @@ final class SmokeTests: XCTestCase { } XCTAssertEqual(restart.route.epoch, second) - XCTAssertThrowsError(try epochTime("not-an-epoch")) + XCTAssertThrowsError(try broadcast.announce(route: AnnounceRoute(epoch: "not-an-epoch"))) } func testDynamicServesARequestUnderAPrefix() async throws {