diff --git a/.env.example b/.env.example index 3ef1215..d327c35 100644 --- a/.env.example +++ b/.env.example @@ -32,13 +32,6 @@ LIVEKIT_API_SECRET= SFU_INGRESS_CAPACITY= SFU_EGRESS_CAPACITY= -# Optional one-shot TURN/UDP transport for a controller-selected edge after -# peer and SFU recovery are exhausted. Set all four values together. -# SELECTED_EDGE_TURN_URLS=turn:turn.example.com:3478?transport=udp -# SELECTED_EDGE_TURN_SHARED_SECRET=replace-with-32-or-more-independent-random-bytes -# SELECTED_EDGE_TURN_CREDENTIAL_TTL_SECONDS=120 -# SELECTED_EDGE_TURN_ALLOCATION_CAPACITY=replace-with-measured-positive-integer - # Ordinary peer ICE discovery. Production requires at least one STUN URL. # Separate multiple URLs with commas. STUN_URLS= diff --git a/docs/project-memory.md b/docs/project-memory.md index 93513cc..196c5e7 100644 --- a/docs/project-memory.md +++ b/docs/project-memory.md @@ -27,7 +27,7 @@ Last updated: 2026-08-24 ## Current Source And Production - Canonical source is the root `main` branch. Audit/integration boundaries require it to be clean, and new branches/worktrees start from its exact commit only after accepted truth is merged. -- Canonical `main` at `a5b1fc62f609b996ead54f2a5db42ddd2f374fb9` admits up to `20` Viewers and implements the uniform `1/2/3` endpoint cap, single Browser `screener-v8` wire, exact prepare candidate, first-decoded-frame commit, strict rollback revision, decoded-progress stall, staged operation deadline, and one event-driven room controller. Its source still contains the superseded selected-TURN path; removing that path and retaining only direct peer plus SFU is the active implementation boundary. SFU admission uses exact publication and Viewer-subscription handles with explicit no-default ingress/egress capacities and a listener-fenced dedicated LiveKit owner. +- Current source admits up to `20` Viewers and implements the uniform `1/2/3` endpoint cap, single Browser `screener-v8` wire, exact prepare candidate, first-decoded-frame commit, strict rollback revision, decoded-progress stall, a two-stage direct/SFU operation deadline, and one event-driven room controller. Application TURN URL/credential schema, messages, candidates, admission, relay-only clients, UI, gate, configuration, and tests are removed; stale TURN environment keys fail startup. SFU admission uses exact publication and Viewer-subscription handles with explicit no-default ingress/egress capacities and a listener-fenced dedicated LiveKit owner. - Current source and production implement the memory-resident room model: random free four-digit codes, the configurable dormant lease, exact-Host resume, restart loss, local creation-preference replay, orthogonal Viewer grants and `open | password | disabled` code entry, and direct invitation/policy controls. SQLite room persistence, `ROOM_DATABASE_PATH`, `ROOM_TTL_SECONDS`, and the writable room StateDirectory are removed from the v8 runtime; the stopped v7 SQLite snapshot exists only in the verified rollback backup. - Production runs exact `a5b1fc62f609b996ead54f2a5db42ddd2f374fb9`, release `a5b1fc6`, wire `screener-v8`; `/opt/screener/backups/a5b1fc6-precutover-20260823T160319Z` is its verified pre-cutover boundary. - Production implements the endpoint-cap and staged route-controller contract, admits 20 Viewers, and uses LiveKit admission of one publication ingress plus 20 subscription egress handles on a dedicated non-auto-create instance. It still has the superseded four-value selected-TURN environment tuple and authenticated coturn relay configuration; accepted truth is not deployed until both are removed atomically. diff --git a/docs/session-handoff.md b/docs/session-handoff.md index 1e1014b..dbf2a12 100644 --- a/docs/session-handoff.md +++ b/docs/session-handoff.md @@ -32,9 +32,10 @@ Current high-signal boundary: - docs/status.md owns the exact source/production/hold state. - docs/todo.md is the only executable work ledger and owns ordering. - ADR-0005 owns the accepted automatic route model: direct/STUN peer followed - only by the dedicated LiveKit SFU/UDP fallback. Production release `a5b1fc6` - has the staged route deadline but still contains the superseded selected-TURN - surface; its atomic removal is the active boundary. Real external-network + only by the dedicated LiveKit SFU/UDP fallback. Current source removes the + complete selected-TURN application surface; production release `a5b1fc6` + still contains its old source/configuration, so atomic deployment is the + active boundary. Real external-network and SFU media evidence remains open. Do not rerun broad or executable/native test suites unless the current acceptance boundary specifically requires them. diff --git a/docs/status.md b/docs/status.md index 7d532c0..e78030a 100644 --- a/docs/status.md +++ b/docs/status.md @@ -18,10 +18,11 @@ This is the current execution index. Git history owns completed timelines; [veri - Current source implements one steady outbound media-copy cap for every non-server endpoint: default `2`, static `1/2/3`, a peer child or Host publication consumes one slot, upstream receive is free, and no Browser/UA/visibility tier exists. One shared guard owns the accounting and transition work is bounded by `min(C + 1, 3)`. - Current source and production admission support one Host plus up to `20` Viewers. Focused controller coverage includes `C=1/2/3`, a 20-Viewer graph, and the passed local 20-Viewer Browser smoke; real external-network media and resource measurements remain open. - The single Browser `screener-v8` boundary rejects stale Browser and executable-sender wires before room authority and carries the exact deployment cap plus exact prepare candidate in the authenticated route contract. -- Current source also implements explicit no-default SFU ingress/egress capacities and exact `reserved | committed | draining` accounting. The listener owner reconciles its dedicated non-auto-create LiveKit namespace before serving, creates each generation before token issuance, releases only after delete-plus-absence proof, and reclaims Host-offline generations through bounded exact-participant checks. The superseded selected-TURN source remains present at this truth checkpoint and is the active removal scope. +- Current source also implements explicit no-default SFU ingress/egress capacities and exact `reserved | committed | draining` accounting. The listener owner reconciles its dedicated non-auto-create LiveKit namespace before serving, creates each generation before token issuance, releases only after delete-plus-absence proof, and reclaims Host-offline generations through bounded exact-participant checks. Application TURN URL/credential schema, messages, candidates, admission, relay-only clients, UI, gate, configuration, and tests are removed; stale TURN environment keys fail startup. - Current source implements automatic P2P-first routing with one active upstream, an acyclic source-reachable graph, exact Browser v7 prepare tuple, first-decoded-frame make-before-break, strict rollback revision, and bounded wait or failure. - One event-driven controller owns the committed graph and at most one room-serial child operation. The operation owns one deterministic candidate list/cursor, one current candidate and reservations, fact version, and one total deadline split only across direct and SFU stages. Failed candidates are operation-local; exhaustion retires invalid/overflow edges but retains healthy bootstrap media. Bounded-gap preflights resources, replans the remaining suffix, and restores a healthy retired edge through one final ordinary candidate. Admission commits synchronously before graph promotion; release uses the resource-set difference and room disposal returns all resources. One Host publication serves exact Viewer subscription handles, and actual global SFU release wakes registered waiting rooms. - Parent selection is deterministic rather than random or score-based: hard eligibility filters, then shallowest result, remaining sender capacity, stable join order, and peer identity. When a departed Host root frees a Host slot, a new or reparable child prefers that shallower Host result over another root's free slot; healthy old edges remain sticky and the controller does not periodically rebalance the whole graph. The exact candidate child's first new decoded frame is the only application media-ready event. Moderate bitrate, FPS, or visual degradation remains with stock WebRTC/LiveKit adaptation and diagnostics; only hard connection failure or a non-paused decoded-frame stall invalidates a route edge. Quality-driven local reparenting and topology evidence are accepted research/TODO work, not current behavior. +- The direct-to-SFU source passed repository typecheck, all 531 Web tests in 41 files, and client/server production builds. Focused tests cover the direct-stage wake, Host-full virtual SFU bootstrap without a premature route error, SFU reuse, and a departed-root scenario in which new demand uses the freed Host slot without moving healthy descendants. No Native/executable suite or heterogeneous-network media run was performed. - Browser initial capture and source switching already request available share audio through one shared capture boundary and visibly retain video-only sharing when no audio track is returned. - Live 64/128/256 audio-ceiling mutation and paused-share video codec switching are accepted but not yet implemented on canonical `main`. H.264 startup blur/low-FPS reports remain an evidence-led diagnosis, not a presumed codec defect. Public-server and fully local distribution packages are later work after the functional, NAT, and real-network ledger is complete. - Current source and production implement one process-memory RoomStore with random four-digit codes, a default 24-hour dormant lease, exact-Host resume, restart loss, local Host preference replay, independent token invitations and `open | password | disabled` code entry, and no SQLite runtime. The Host UI separates the invitation, room-code entry policy, password, grant update, and grant revoke controls. diff --git a/docs/verification-status.md b/docs/verification-status.md index 5efeddd..694fe08 100644 --- a/docs/verification-status.md +++ b/docs/verification-status.md @@ -26,7 +26,7 @@ Capacity and routing entries below describe the exact tested source or release o - Exact route source `4e0a922` passed repository hygiene, TypeScript typecheck, and 84 focused controller, runtime, protocol, transition, decoded-stall, and SFU-admission tests. During the same convergence, the full 557-test Web suite and client/server builds passed before the final no-consumer cleanup. The two independent route audits' model and wire findings are incorporated, and final static review found no remaining P0/P1. No Browser, Native, Go, helper, hardware, or executable test ran in the final pass; this is source and state-machine evidence, not production or heterogeneous-network proof. - Exact source `6b87732` passed the required one-Host/20-Viewer local Browser smoke on Windows Node 24.19.0 / Chrome 151.0.7922.174, using 720p30 synthetic canvas media, endpoint cap `2`, a 30-second measurement window, and report SHA-256 `1347E036BB9AC35D51B4ED9B215290E742715128C7F3056D8FEB185CC84456D6`. All six checks passed: 20/20 Viewers had live clean upstreams and increasing decoded frames, Host and relay sender counts stayed at two, topology was coherent, and no orphan SFU publication appeared. Maximum authenticate-to-first-decoded latency was 958 ms; the final graph had two Host children and eighteen peer descendants. The application Node process consumed about 0.95 additional CPU seconds over the 54-second observation window, peaked near 284 MiB RSS, and settled near 162 MiB; isolated Chromium averaged 761% CPU over its valid 25.97-second process interval. The harness did not configure LiveKit, so its observed SFU usage was zero and this loopback is not a public-network, SFU, quality, or endurance claim. - Exact memory-room source `2ba155782bd5c4a193484c007ee69007e562ddb9` passed repository hygiene, TypeScript typecheck, all 548 Web tests, and client/server builds. The focused room/access set contributed 239 passing tests across allocation, dormant leases, grant/code orthogonality, password KDF bounds, stale generation atomicity, protocol v8, HTTP/signaling, local ownership, and creation-profile replay. A fresh Chrome access/privacy gate also passed all fixed checks: pre-DOM fragment consumption, room-scoped session grant, zero request/nginx/application-log secret hits, zero persistent room files, strong grant rotate/revoke coverage, and cleanup. This is local source, state-machine, and loopback-browser evidence; it is not production deployment, restart/rollback, heterogeneous-network, or media-quality evidence. -- Current source TURN-admission unit and signaling tests cover independent peer and Host-SFU logical edges, shared capacity exhaustion, exact reserve/commit/drain/release identity, current-share cleanup, and graceful shutdown. This is local authorization evidence, not real coturn, network-media, or production-capacity proof. +- The direct-to-SFU source passed repository typecheck, all 531 Web tests in 41 files, and client/server production builds. The server build now clears its fixed generated-output directory before compilation, and the resulting tree contains zero selected-TURN modules or symbols. Focused controller/runtime coverage proves direct-stage wake, Host-full virtual SFU bootstrap without a premature route error, SFU reuse, and a departed-root sequence where new demand takes the freed Host slot while healthy descendants remain sticky. This is local state-machine/build evidence; no Native/executable suite or heterogeneous-network media run was performed. - Chrome 151 synthetic `1/3/5/8` and 720p30 runs preserved bounded fanout and decoding. A `9461e20`-era three-Viewer rerun kept Host2/Browser Viewer1, decoded every Viewer, and recovered a hard first-level relay departure in 5,380 ms; it did not exercise quality-triggered MBB. Sixteen-Viewer cap2/cap3 loopbacks decoded all Viewers and a ten-second follow-up recorded guarded sender and aggregate Chromium CPU evidence, but resolution/FPS remained low and the ordinary-PC data cannot rank caps, select product policy, or prove sustainable heterogeneous-network quality. See [peer-assisted media](./research/peer-assisted-media.md). - Chrome 151 with local LiveKit 1.13.5/client 2.22.0 proved one SFU/UDP root after `q,f` -> `q,h`: decoded frames advanced 3 -> 23, rendered frames reached 26, Host used one edge, leaves were clean, and TURN was absent. This is functional evidence only. See [low-server media routes](./research/low-server-media-routes.md). - Host A+B and authenticated P2P Viewer C evidence is sanitized, generation-bound, diagnostic-only, and closed for stale or ambiguous reports. Production can expose a Host-local current SFU `h` snapshot and clears it on publisher identity changes; raw media metadata is not retained. See [realtime quality adaptation](./research/realtime-quality-adaptation.md). diff --git a/native/sender/internal/remote/peer.go b/native/sender/internal/remote/peer.go index acfe31f..57c85a1 100644 --- a/native/sender/internal/remote/peer.go +++ b/native/sender/internal/remote/peer.go @@ -294,11 +294,9 @@ func (peer *peer) snapshot(now time.Time) PeerDiagnostics { func selectedRoute(report webrtc.StatsReport, pair webrtc.ICECandidatePairStats) string { local, localOK := report[pair.LocalCandidateID].(webrtc.ICECandidateStats) remote, remoteOK := report[pair.RemoteCandidateID].(webrtc.ICECandidateStats) - if localOK && local.CandidateType == webrtc.ICECandidateTypeRelay { - return "turn-" + relayTransport(local) - } - if remoteOK && remote.CandidateType == webrtc.ICECandidateTypeRelay { - return "turn-unknown" + if localOK && local.CandidateType == webrtc.ICECandidateTypeRelay || + remoteOK && remote.CandidateType == webrtc.ICECandidateTypeRelay { + return "unknown" } if !localOK || !remoteOK { return "unknown" @@ -306,15 +304,6 @@ func selectedRoute(report webrtc.StatsReport, pair webrtc.ICECandidatePairStats) return "direct-" + candidateProtocol(local) } -func relayTransport(candidate webrtc.ICECandidateStats) string { - switch strings.ToLower(candidate.RelayProtocol) { - case "udp", "tcp", "tls": - return strings.ToLower(candidate.RelayProtocol) - default: - return "unknown" - } -} - func candidateProtocol(candidate webrtc.ICECandidateStats) string { if protocol := strings.ToLower(candidate.Protocol); protocol == "udp" || protocol == "tcp" { return protocol diff --git a/native/sender/internal/remote/peer_test.go b/native/sender/internal/remote/peer_test.go index aefef73..b0f5fef 100644 --- a/native/sender/internal/remote/peer_test.go +++ b/native/sender/internal/remote/peer_test.go @@ -6,7 +6,7 @@ import ( "github.com/pion/webrtc/v4" ) -func TestSelectedRouteTreatsARemoteRelayCandidateAsTURN(t *testing.T) { +func TestSelectedRouteLeavesARemoteRelayCandidateUnknown(t *testing.T) { report := webrtc.StatsReport{ "local": webrtc.ICECandidateStats{ ID: "local", @@ -20,12 +20,12 @@ func TestSelectedRouteTreatsARemoteRelayCandidateAsTURN(t *testing.T) { }, } pair := webrtc.ICECandidatePairStats{LocalCandidateID: "local", RemoteCandidateID: "remote"} - if route := selectedRoute(report, pair); route != "turn-unknown" { - t.Fatalf("selected route = %q, want turn-unknown", route) + if route := selectedRoute(report, pair); route != "unknown" { + t.Fatalf("selected route = %q, want unknown", route) } } -func TestSelectedRouteUsesOnlyTheLocalRelayProtocol(t *testing.T) { +func TestSelectedRouteLeavesALocalRelayCandidateUnknown(t *testing.T) { report := webrtc.StatsReport{ "local": webrtc.ICECandidateStats{ ID: "local", Protocol: "udp", CandidateType: webrtc.ICECandidateTypeRelay, @@ -35,14 +35,8 @@ func TestSelectedRouteUsesOnlyTheLocalRelayProtocol(t *testing.T) { }, } pair := webrtc.ICECandidatePairStats{LocalCandidateID: "local", RemoteCandidateID: "remote"} - if route := selectedRoute(report, pair); route != "turn-unknown" { - t.Fatalf("selected route without local relayProtocol = %q", route) - } - local := report["local"].(webrtc.ICECandidateStats) - local.RelayProtocol = "tcp" - report["local"] = local - if route := selectedRoute(report, pair); route != "turn-tcp" { - t.Fatalf("selected route with local relayProtocol = %q", route) + if route := selectedRoute(report, pair); route != "unknown" { + t.Fatalf("selected route = %q, want unknown", route) } } diff --git a/native/sender/internal/remote/session.go b/native/sender/internal/remote/session.go index 6ca6d35..7b86676 100644 --- a/native/sender/internal/remote/session.go +++ b/native/sender/internal/remote/session.go @@ -453,12 +453,6 @@ func (session *Session) handle(message serverMessage) error { return session.handleRouteUpdate(message) case "sfu-config": return session.handleUnsupportedRouteMessage(message, nil) - case "selected-edge-turn": - if message.EdgeKind == "host-sfu-ingress" { - connectionID := message.NewConnectionID - return session.handleUnsupportedRouteMessage(message, &connectionID) - } - return nil case "viewer-quality-evidence": // Fixed-HIGH observes no viewer feedback; accepting this current-wire // message must not turn it into a room-wide quality controller. diff --git a/native/sender/internal/remote/session_test.go b/native/sender/internal/remote/session_test.go index 4f5cc32..b904ba4 100644 --- a/native/sender/internal/remote/session_test.go +++ b/native/sender/internal/remote/session_test.go @@ -430,20 +430,8 @@ func TestPeerAssistedHostUsesOnlyAuthoritativeRoutes(t *testing.T) { if err = session.handle(sfuUpdate); err != nil { t.Fatal(err) } - selected := serverMessage{Type: "selected-edge-turn", EdgeKind: "host-sfu-ingress", RouteRevision: 9, NewConnectionID: "connection-new"} - if err = session.handle(selected); err != nil { - t.Fatal(err) - } - connectionID := "connection-new" - assertRouteStatus(t, receiveRouteStatus(t, received), "route-failed", 9, "active", &connectionID) - if err = session.handle(selected); err != nil { - t.Fatal(err) - } - if err = session.handle(sfuUpdate); err != nil { - t.Fatal(err) - } if err = session.handle(serverMessage{Type: "error", Code: "PEER_NOT_FOUND"}); err != nil { - t.Fatal("bounded selected-edge fallback became fatal") + t.Fatal("bounded route failure became fatal") } select { case payload := <-received: diff --git a/native/sender/internal/remote/wire.go b/native/sender/internal/remote/wire.go index 4b1daaf..29e5e24 100644 --- a/native/sender/internal/remote/wire.go +++ b/native/sender/internal/remote/wire.go @@ -112,8 +112,6 @@ type serverMessage struct { RouteAssignment participantRouteAssignment RouteRevision int64 RoutePhase string - EdgeKind string - NewConnectionID string } func readServerMessage(ctx context.Context, conn *websocket.Conn) (serverMessage, error) { @@ -209,18 +207,6 @@ func decodeServerMessage(payload []byte) (serverMessage, error) { return serverMessage{}, errors.New("SFU configuration message is invalid") } return serverMessage{Type: wire.Type, RouteRevision: wire.Revision}, nil - case "selected-edge-turn": - var wire struct { - Type string `json:"type"` - EdgeKind string `json:"edgeKind"` - Revision int64 `json:"revision"` - NewConnectionID string `json:"newConnectionId"` - } - if err := json.Unmarshal(payload, &wire); err != nil || wire.Revision < 0 || wire.Revision > maxRouteRevision || - (wire.EdgeKind != "host-sfu-ingress" && wire.EdgeKind != "peer-selected") || !validOpaqueID(wire.NewConnectionID) { - return serverMessage{}, errors.New("selected-edge TURN message is invalid") - } - return serverMessage{Type: wire.Type, EdgeKind: wire.EdgeKind, RouteRevision: wire.Revision, NewConnectionID: wire.NewConnectionID}, nil case "viewer-quality-evidence": if err := validateViewerQualityEvidence(payload); err != nil { return serverMessage{}, err diff --git a/native/sender/internal/remote/wire_test.go b/native/sender/internal/remote/wire_test.go index ba82882..d2b108c 100644 --- a/native/sender/internal/remote/wire_test.go +++ b/native/sender/internal/remote/wire_test.go @@ -145,7 +145,6 @@ func TestDecodePeerAssistedRouteMessages(t *testing.T) { {`{"type":"media-assignment","mediaAssignment":{"parentPeerId":null,"childPeerIds":["viewer-1"]}}`, "media-assignment"}, {`{"type":"route-update","revision":8,"phase":"active","assignment":{"upstream":{"kind":"none"},"childPeerIds":["viewer-1"],"sfuPublicationGeneration":null}}`, "route-update"}, {`{"type":"sfu-config","revision":8,"url":"wss://example.test","token":"opaque"}`, "sfu-config"}, - {`{"type":"selected-edge-turn","edgeKind":"host-sfu-ingress","revision":8,"hostPeerId":"host-peer","publicationGeneration":"publication-1","oldConnectionId":"connection-old","newConnectionId":"connection-new","expiresAt":"2030-01-01T00:00:00Z","iceServer":{"urls":["turn:example.test?transport=udp"],"username":"1:abcdefghijklmnop","credential":"opaque"}}`, "selected-edge-turn"}, } for _, test := range tests { message, err := decodeServerMessage([]byte(test.payload)) diff --git a/package.json b/package.json index 22fc8a1..d52e652 100644 --- a/package.json +++ b/package.json @@ -11,12 +11,11 @@ "dev": "tsx watch src/server/index.ts", "build": "npm run build:client && npm run build:server", "build:client": "vite build", - "build:server": "tsc -p tsconfig.server.json", + "build:server": "node --input-type=module -e \"import { rmSync } from 'node:fs'; rmSync('dist/server', { recursive: true, force: true })\" && tsc -p tsconfig.server.json", "start": "node dist/server/server/index.js", "gate:peer-topology-loopback": "tsx scripts/peer-assisted-benchmark.ts", "gate:sfu-admission": "vitest run tests/room-route-controller.test.ts tests/hybrid-media-router.test.ts tests/sfu-resource-admission.test.ts tests/sfu-room-control.test.ts tests/livekit-token.test.ts tests/sfu-client.test.ts tests/route-transition.test.ts tests/server-config.test.ts tests/server-http.test.ts tests/server-signal.test.ts", "gate:access-privacy": "npm run build:client && vitest run tests/server-signal.test.ts && tsx scripts/access-privacy-gate.ts", - "gate:turn-udp-allocation": "tsx scripts/turn-allocation-gate.ts", "probe:native-one-viewer": "tsx scripts/native-one-viewer-gate.ts", "typecheck": "tsc -b --pretty false", "test": "vitest run", diff --git a/scripts/native-one-viewer-gate.ts b/scripts/native-one-viewer-gate.ts index e83d5c2..a3ff772 100644 --- a/scripts/native-one-viewer-gate.ts +++ b/scripts/native-one-viewer-gate.ts @@ -342,8 +342,8 @@ async function main(): Promise { ? "Hardware status is the helper's hardware-only MF contract; physical VideoEncode use requires correlated OS telemetry." : "This run proves one WebCodecs object, not one physical or hardware encoder.", windowTargetTitle - ? "The loopback run does not prove public STUN, restrictive networks, TURN, packaging, or endurance." - : "The loopback run does not prove public STUN, restrictive networks, TURN, audio, packaging, or endurance.", + ? "The loopback run does not prove public STUN, SFU, restrictive networks, packaging, or endurance." + : "The loopback run does not prove public STUN, SFU, restrictive networks, audio, packaging, or endurance.", ], }; let server: ScreenerServer | null = null; @@ -998,7 +998,7 @@ function requireChecks(checks: Record): void { } function safeRoute(value: string | undefined): string { - return new Set(["direct-udp", "direct-tcp", "turn-udp", "turn-tcp", "turn-tls", "turn-unknown"]) + return new Set(["direct-udp", "direct-tcp"]) .has(value ?? "") ? value! : "unknown"; } diff --git a/scripts/peer-assisted-benchmark.ts b/scripts/peer-assisted-benchmark.ts index d767832..29d5e5a 100644 --- a/scripts/peer-assisted-benchmark.ts +++ b/scripts/peer-assisted-benchmark.ts @@ -2627,7 +2627,7 @@ export async function main(): Promise { "Headless runs are topology and transport evidence, not representative GPU or power evidence.", "First-frame and recovery timing fields are diagnostics and never determine this loopback gate's status.", "CDP process CPU covers the isolated Chromium instance, not a specific Host or relay page; identity changes or counter resets make that interval unknown, and multicore utilization may exceed 100%.", - "CDP SystemInfo exposes no resident-set field, so peakResidentSetBytes is null; GPU, NIC, glass-to-glass latency, generational visual quality, mobile browsers, and TURN require other measurement.", + "CDP SystemInfo exposes no resident-set field, so peakResidentSetBytes is null; GPU, NIC, glass-to-glass latency, generational visual quality, mobile browsers, and SFU require other measurement.", "The local runner does not start LiveKit; SFU consistency is reported only when an SFU route is actually observed.", "The harness emits raw gate fields and simple invariants; it does not implement a route score or runtime policy.", "BENCHMARK_CANARY=viewer-mbb injects only sanitized control counters; it does not claim detector quality or network performance. Host-candidate and signaling-blackhole canaries remain deferred.", diff --git a/scripts/turn-allocation-gate.ts b/scripts/turn-allocation-gate.ts deleted file mode 100644 index 2d94725..0000000 --- a/scripts/turn-allocation-gate.ts +++ /dev/null @@ -1,126 +0,0 @@ -import { execFile } from "node:child_process"; -import { once } from "node:events"; -import { readFileSync } from "node:fs"; -import { mkdtemp, rm } from "node:fs/promises"; -import { createServer } from "node:http"; -import { tmpdir } from "node:os"; -import { join } from "node:path"; -import { pathToFileURL } from "node:url"; - -import { parseSelectedEdgeTurn } from "../src/server/config"; -import { issueSelectedEdgeTurnCredential } from "../src/server/selected-edge-turn"; - -const allowedProtocols = new Set(["udp", "tcp", "unknown"]); -const allowedErrorBuckets = new Set(["3xx", "4xx", "5xx", "6xx", "701", "other"]); - -export interface TurnAllocationResult { - status: "passed" | "failed"; - relayCandidateCount: number; - protocols: string[]; - errorCodeBuckets: string[]; -} - -export function parseGateEnvironment(input: string): NodeJS.ProcessEnv { - const environment: NodeJS.ProcessEnv = {}; - for (const line of input.split(/\r?\n/)) { - const separator = line.indexOf("="); - if (separator > 0) environment[line.slice(0, separator)] = line.slice(separator + 1); - } - return environment; -} - -export function parseGateResult(input: string): TurnAllocationResult { - let value: unknown; - try { value = JSON.parse(input); } catch { throw new Error("Chrome returned an invalid TURN allocation result"); } - if (!value || typeof value !== "object") throw new Error("Chrome returned an invalid TURN allocation result"); - const raw = value as Record; - if ((raw.status !== "passed" && raw.status !== "failed") || - !Number.isInteger(raw.relayCandidateCount) || Number(raw.relayCandidateCount) < 0 || - !Array.isArray(raw.protocols) || !raw.protocols.every((item) => allowedProtocols.has(item)) || - !Array.isArray(raw.errorCodeBuckets) || !raw.errorCodeBuckets.every((item) => allowedErrorBuckets.has(item))) { - throw new Error("Chrome returned an invalid TURN allocation result"); - } - const protocols = [...new Set(raw.protocols as string[])]; - if (raw.status === "passed" && (Number(raw.relayCandidateCount) === 0 || - protocols.length !== 1 || protocols[0] !== "udp")) { - throw new Error("Chrome returned an invalid TURN allocation result"); - } - return { status: raw.status, relayCandidateCount: Number(raw.relayCandidateCount), - protocols, - errorCodeBuckets: [...new Set(raw.errorCodeBuckets as string[])] }; -} - -const page = ``; - -export async function main(): Promise { - const chromePath = process.env.CHROME_PATH?.trim(); - if (!chromePath) throw new Error("CHROME_PATH is required"); - const stdin = process.stdin.isTTY ? "" : readFileSync(0, "utf8"); - const config = parseSelectedEdgeTurn({ ...parseGateEnvironment(stdin), ...process.env }); - if (!config) throw new Error("The selected-edge TURN tuple is required"); - const grant = issueSelectedEdgeTurnCredential(config, { edgeKind: "host-sfu-ingress", - roomId: "allocation-gate", shareGeneration: "gate-share", revision: 1, - hostPeerId: "gate-host", hostSessionId: "gate-session", - publicationGeneration: "gate-publication", oldConnectionId: "gate-publication", - newConnectionId: "gate-connection" }, Date.now()); - let settle!: (result: TurnAllocationResult) => void; - const resultPromise = new Promise((resolve) => { settle = resolve; }); - const server = createServer((request, response) => { - response.setHeader("Cache-Control", "no-store"); - if (request.url === "/credential") response.end(JSON.stringify(grant.iceServer)); - else if (request.url === "/") response.end(page); - else if (request.url === "/result" && request.method === "POST") { - let body = ""; - request.setEncoding("utf8"); - request.on("data", (chunk) => { if (body.length < 4_096) body += chunk; }); - request.on("end", () => { - try { settle(parseGateResult(body)); response.end(); } - catch { response.statusCode = 400; response.end(); } - }); - } - else { response.statusCode = 404; response.end(); } - }); - const profile = await mkdtemp(join(tmpdir(), "screener-turn-gate-")); - let chrome: ReturnType | undefined; - let chromeClosed: Promise = Promise.resolve(); - let timeout: NodeJS.Timeout | undefined; - try { - server.listen(0, "127.0.0.1"); await once(server, "listening"); - const address = server.address(); - if (!address || typeof address === "string") throw new Error("TURN gate listener failed"); - chrome = execFile(chromePath, ["--headless=new", "--no-first-run", - "--no-default-browser-check", "--disable-extensions", `--user-data-dir=${profile}`, - `http://127.0.0.1:${address.port}/`], { timeout: 20_000, windowsHide: true }, () => {}); - chromeClosed = new Promise((resolve) => { chrome!.once("close", () => resolve()); }); - const chromeFailure = new Promise((_, reject) => { - chrome!.once("error", () => reject(new Error("Chrome failed to start"))); - chrome!.once("exit", () => reject(new Error("Chrome exited before TURN allocation completed"))); - }); - const result = await Promise.race([resultPromise, chromeFailure, - new Promise((_, reject) => { timeout = setTimeout(() => reject(new Error("Chrome did not complete the TURN allocation gate")), 15_000); })]); - process.stdout.write(`${JSON.stringify({ schemaVersion: 1, ...result })}\n`); - return result.status === "passed" ? 0 : 1; - } finally { - if (timeout) clearTimeout(timeout); - if (chrome?.exitCode === null) chrome.kill(); - const serverClose = server.listening - ? new Promise((resolve) => { server.close(() => resolve()); }) - : Promise.resolve(); - await Promise.allSettled([chromeClosed, serverClose]); - await rm(profile, { recursive: true, force: true }); - } -} - -if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) { - main().then((code) => { process.exitCode = code; }).catch((error: Error) => { - process.stderr.write(`${error.message}\n`); process.exitCode = 2; - }); -} diff --git a/src/client/components/StatsGrid.tsx b/src/client/components/StatsGrid.tsx index d15cd58..1bf0162 100644 --- a/src/client/components/StatsGrid.tsx +++ b/src/client/components/StatsGrid.tsx @@ -138,12 +138,6 @@ export function StatsGrid({ label="传输协议" value={metrics.iceProtocol?.toUpperCase() ?? "未知"} /> - {metrics.localCandidateType === "relay" && ( - - )} ; } -export function PathBadge({ metrics }: { metrics: ConnectionMetrics }) { - const presentation = mediaTransportPresentation(metrics); - if (metrics.path === "direct") { - return ( - - - ); - } - if (metrics.path === "relay") { - return ( - - - ); - } - return ( - - - ); -} - export function MediaRouteBadge({ route }: { route: "p2p" | "sfu" }) { const presentation = MEDIA_ROUTE_PRESENTATION[route]; const Icon = presentation.icon === "network" ? Network : Server; diff --git a/src/client/components/status-badge-model.ts b/src/client/components/status-badge-model.ts index ce7bf7a..d789612 100644 --- a/src/client/components/status-badge-model.ts +++ b/src/client/components/status-badge-model.ts @@ -9,34 +9,6 @@ export const ROUTING_STATUS_PRESENTATION = { tone: "neutral", label: "线路分 export type PresentedMediaRoute = keyof typeof MEDIA_ROUTE_PRESENTATION; -export function mediaTransportPresentation( - metrics: Pick< - ConnectionMetrics, - "path" | "iceProtocol" | "localCandidateType" | "localRelayProtocol" - >, -): { label: string; title: string } { - if (metrics.path === "direct") { - const protocol = metrics.iceProtocol?.toUpperCase() ?? null; - return { - label: protocol ?? "直连", - title: protocol ? `此连接使用直连 ${protocol}` : "此连接使用直连传输", - }; - } - if (metrics.path === "relay") { - const protocol = - metrics.localCandidateType === "relay" - ? (metrics.localRelayProtocol?.toUpperCase() ?? null) - : null; - return { - label: protocol ? `TURN/${protocol}` : "TURN", - title: protocol - ? `此连接正在使用 TURN/${protocol} 中继` - : "此连接正在使用 TURN 中继;远端中继协议不可见", - }; - } - return { label: "传输未确定", title: "尚未选出可用的传输路径" }; -} - export function hasPeerRouteEvidence( peer: PeerSnapshot | null | undefined, ): boolean { diff --git a/src/client/lib/diagnostic-export.ts b/src/client/lib/diagnostic-export.ts index a2e0e9e..efcc992 100644 --- a/src/client/lib/diagnostic-export.ts +++ b/src/client/lib/diagnostic-export.ts @@ -7,7 +7,7 @@ const DIAGNOSTIC_METRIC_KEYS = [ "candidatePairResponsesReceived", "intervalCandidatePairResponsesReceived", "candidatePairSampleWindowMs", "captureFramesPerSecond", "mediaSourceFramesPerSecond", "path", - "iceProtocol", "localRelayProtocol", "localCandidateType", + "iceProtocol", "localCandidateType", "remoteCandidateType", "rttMs", "bitrateKbps", "availableOutgoingKbps", "framesPerSecond", "frameWidth", "frameHeight", "resolution", "packetsLost", "intervalPacketsSent", "intervalPacketsReceived", diff --git a/src/client/media/host-provisional-child.ts b/src/client/media/host-provisional-child.ts index 668e30b..7f6246b 100644 --- a/src/client/media/host-provisional-child.ts +++ b/src/client/media/host-provisional-child.ts @@ -113,7 +113,6 @@ export class HostProvisionalChild { } }, }, - false, candidate.connectionId, ); this.signalingPeer = peer; diff --git a/src/client/media/host-sfu-route.ts b/src/client/media/host-sfu-route.ts index 0c29de6..b9a61a1 100644 --- a/src/client/media/host-sfu-route.ts +++ b/src/client/media/host-sfu-route.ts @@ -40,8 +40,6 @@ interface HostPublisherSlot { connected: boolean; active: boolean; failed: boolean; - connectionId: string; - selectedEdgeTurn: boolean; } export interface HostSfuPublisherSnapshot { @@ -72,12 +70,6 @@ export class HostSfuRoute { private resyncing = false; private paused = false; private closed = false; - private lastConfig: Extract | null = - null; - private selectedEdgeTurn: Extract< - ServerMessage, - { type: "selected-edge-turn"; edgeKind: "host-sfu-ingress" } - > | null = null; constructor(private readonly events: HostSfuRouteEvents) {} @@ -97,8 +89,6 @@ export class HostSfuRoute { if (previousRevision !== update.revision) { this.recovery = null; this.lastFailureStage = null; - this.lastConfig = null; - this.selectedEdgeTurn = null; } if (update.phase === "prepare") { if ( @@ -204,8 +194,6 @@ export class HostSfuRoute { ) { return; } - this.lastConfig = message; - if (this.active?.publicationGeneration === publicationGeneration) { this.active.revision = message.revision; if (phase === "active" && this.active.active) { @@ -232,13 +220,6 @@ export class HostSfuRoute { onDisconnected: () => this.handleFailure(slot), onStats: (metrics) => this.handlePublisherStats(slot, metrics), }); - const selectedEdgeTurn = - this.selectedEdgeTurn?.revision === message.revision && - this.selectedEdgeTurn.publicationGeneration === publicationGeneration && - this.selectedEdgeTurn.newConnectionId === candidate?.connectionId - ? this.selectedEdgeTurn - : null; - this.selectedEdgeTurn = null; slot = { revision: message.revision, publicationGeneration, @@ -246,22 +227,12 @@ export class HostSfuRoute { connected: false, active: false, failed: false, - connectionId: selectedEdgeTurn?.newConnectionId ?? publicationGeneration, - selectedEdgeTurn: selectedEdgeTurn !== null, }; this.pending = slot; try { const connected = await publisher.connect({ url: message.url, token: message.token, - ...(selectedEdgeTurn - ? { - rtcConfig: { - iceServers: [selectedEdgeTurn.iceServer], - iceTransportPolicy: "relay", - }, - } - : {}), }); if (!connected) { if (this.pending === slot && this.route.owns(token)) { @@ -291,50 +262,9 @@ export class HostSfuRoute { this.paused = paused; if (paused) { this.clearPending(); - this.selectedEdgeTurn = null; } } - /** Apply one controller-selected TURN grant to the next SFU publisher PC. */ - startSelectedEdgeTurn( - message: Extract< - ServerMessage, - { type: "selected-edge-turn"; edgeKind: "host-sfu-ingress" } - >, - ): boolean { - if ( - this.closed || - !this.route.acceptsConfig(message.revision) || - this.route.getPhase() !== "prepare" || - this.route.getPreparedCandidate()?.transport !== "sfu" || - this.route.getPreparedCandidate()?.connectionId !== message.newConnectionId || - Date.parse(message.expiresAt) <= Date.now() - ) { - return false; - } - const assignment = this.route.getPlannedAssignment(); - const publicationGeneration = assignment?.sfuPublicationGeneration; - const pending = this.pending; - if ( - !publicationGeneration || - publicationGeneration !== message.publicationGeneration || - (pending !== null && - (pending.revision !== message.revision || - pending.publicationGeneration !== publicationGeneration)) - ) { - return false; - } - if (pending?.connectionId === message.newConnectionId) { - return !pending.failed; - } - this.clearPending(); - this.selectedEdgeTurn = message; - if (this.lastConfig?.revision === message.revision) { - void this.acceptConfig(this.lastConfig); - } - return true; - } - updateProfile(profile: QualityProfile): Promise { const slots = this.publishingSlots(); return slots.length === 0 @@ -401,8 +331,6 @@ export class HostSfuRoute { this.resyncing = false; this.route.reset(); this.recovery = null; - this.lastConfig = null; - this.selectedEdgeTurn = null; await this.queueTransition(async () => { const pending = this.pending; const active = this.active; @@ -626,24 +554,17 @@ export class HostSfuRoute { } if (phase === "prepare") { if (matchesPlannedRoute || wasActive) { - this.routeFailed( - revision, - "prepare", - slot.selectedEdgeTurn ? slot.connectionId : null, - ); + this.routeFailed(revision, "prepare", null); } return; } if (!matchesPlannedRoute) { return; } - this.requestRecovery(revision, slot); + this.requestRecovery(revision); } - private requestRecovery( - revision: number, - failedSlot?: HostPublisherSlot, - ): void { + private requestRecovery(revision: number): void { if (!this.recovery || this.recovery.revision !== revision) { this.recovery = { revision, refreshed: false }; } @@ -653,11 +574,7 @@ export class HostSfuRoute { return; } } - this.routeFailed( - revision, - "active", - failedSlot?.selectedEdgeTurn ? failedSlot.connectionId : null, - ); + this.routeFailed(revision, "active", null); } private routeFailed( diff --git a/src/client/pages/HostPage.tsx b/src/client/pages/HostPage.tsx index 323dc4a..7c46335 100644 --- a/src/client/pages/HostPage.tsx +++ b/src/client/pages/HostPage.tsx @@ -38,7 +38,6 @@ import { RoomCodeEntry } from "../components/RoomCodeEntry"; import { qualityLimitationSummary } from "../components/connection-details"; import { MediaRouteBadge, - PathBadge, PeerStatusBadge, SignalStatusBadge, WarningBanner, @@ -1191,20 +1190,6 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { } return; } - if (message.type === "selected-edge-turn") { - if (message.edgeKind === "host-sfu-ingress") { - if ( - peerAssistedRef.current && - message.hostPeerId === hostPeerIdRef.current && - Date.parse(message.expiresAt) > Date.now() - ) { - const route = ensureHostSfuRoute(generation); - route.startSelectedEdgeTurn(message); - } - return; - } - return; - } if (message.type === "code-entry-policy-updated") { const profile = { codeEntryPolicy: message.codeEntryPolicy, @@ -2630,9 +2615,6 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { - {snapshot?.metrics.path === "relay" && ( - - )} )} {showConnectionDetails && diff --git a/src/client/pages/ViewerPage.tsx b/src/client/pages/ViewerPage.tsx index 8ab196d..a09bf1a 100644 --- a/src/client/pages/ViewerPage.tsx +++ b/src/client/pages/ViewerPage.tsx @@ -29,7 +29,6 @@ import { RoomCode } from "../components/RoomCode"; import { qualityLimitationSummary } from "../components/connection-details"; import { MediaRouteBadge, - PathBadge, PeerStatusBadge, SignalStatusBadge, WarningBanner, @@ -92,10 +91,6 @@ type ViewerQualityEvidence = Extract< ServerMessage, { type: "viewer-quality-evidence" } >; -type SelectedEdgeTurn = Extract< - ServerMessage, - { type: "selected-edge-turn"; edgeKind: "peer-selected" } ->; interface SfuUpstreamState { connectionState: "connected" | "reconnecting"; metrics: ConnectionMetrics | null; @@ -105,9 +100,7 @@ interface PendingPeerRoute { parentPeerId: string; peer: ViewerPeer | null; readySent: boolean; - selectedTurn: SelectedEdgeTurn | null; candidateConnectionId: string; - transport: "direct" | "selected-turn"; stream: MediaStream | null; snapshot: PeerSnapshot | null; } @@ -619,9 +612,7 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { parentPeerId: assignment.upstream.peerId, peer: null, readySent: false, - selectedTurn: null, candidateConnectionId: candidate.connectionId, - transport: candidate.transport, stream: null, snapshot: null, }; @@ -805,9 +796,7 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { if (!probe || !currentIceConfig) return null; if (probe.peer) return probe.peer; const peer: ViewerPeer = new ViewerPeer( - probe.selectedTurn - ? { iceServers: [probe.selectedTurn.iceServer] } - : currentIceConfig, + currentIceConfig, { sendSignal: (targetPeerId, payload) => signal.send(viewerSignalMessage(peerAssisted, targetPeerId, payload)), @@ -863,7 +852,6 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { : true; }, }, - probe.transport === "selected-turn", ); probe.peer = peer; return peer; @@ -1048,39 +1036,6 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { } return; } - if (message.type === "selected-edge-turn") { - if (message.edgeKind !== "peer-selected") { - return; - } - if (!peerAssisted) { - return; - } - if (message.parentPeerId === currentPeerId) { - ensureViewerRelay()?.prepareSelectedEdgeTurn( - message, - currentPeerId, - message.revision, - ); - return; - } - const probe = pendingPeer; - if (probe?.selectedTurn?.newConnectionId === message.newConnectionId) { - return; - } - if ( - message.viewerPeerId !== currentPeerId || - probe?.revision !== message.revision || - probe.parentPeerId !== message.parentPeerId || - probe.transport !== "selected-turn" || - probe.candidateConnectionId !== message.newConnectionId || - Date.parse(message.expiresAt) <= Date.now() || - probe.peer !== null - ) { - return; - } - probe.selectedTurn = message; - return; - } if (message.type === "route-update") { if (peerAssisted) { if ( @@ -1128,13 +1083,6 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { } if (message.type === "signal") { if (peerAssisted && pendingPeer?.parentPeerId === message.fromPeerId) { - if ( - message.payload.connectionId === pendingPeer.candidateConnectionId && - pendingPeer.transport === "selected-turn" && - !pendingPeer.selectedTurn - ) { - return; - } if (currentAssignment.parentPeerId !== message.fromPeerId) { await ensurePendingPeerRoute()?.acceptSignal( message.fromPeerId, @@ -1526,9 +1474,6 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { {showConnectionDetails && routePresentation.route && ( <> - {routeMetrics?.path === "relay" && ( - - )} )} @@ -1767,9 +1712,6 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) {

连接数据

- {routeMetrics?.path === "relay" && ( - - )}
{routeMetrics && ( = MAX_AUTOMATIC_RECOVERY_REQUESTS) { this.reportRecoveryExhausted(); return; diff --git a/src/client/webrtc/viewer-relay.ts b/src/client/webrtc/viewer-relay.ts index b9fd1d7..7072889 100644 --- a/src/client/webrtc/viewer-relay.ts +++ b/src/client/webrtc/viewer-relay.ts @@ -1,7 +1,6 @@ import type { IceConfig, PreparedRouteCandidate, - ServerMessage, SignalPayload, } from "../../shared/protocol"; import { @@ -16,10 +15,6 @@ interface ViewerRelayEvents { sendSignal: (peerId: string, payload: SignalPayload) => boolean; onUpdate?: (snapshot: PeerSnapshot | null) => void; } -type SelectedEdgeTurn = Extract< - ServerMessage, - { type: "selected-edge-turn"; edgeKind: "peer-selected" } ->; interface PreparedChild { revision: number; childPeerId: string; @@ -58,40 +53,6 @@ export class ViewerRelay { : null; } - prepareSelectedEdgeTurn( - message: SelectedEdgeTurn, - parentPeerId: string, - routeRevision: number, - now = Date.now(), - ): boolean { - if ( - this.disposed || - message.parentPeerId !== parentPeerId || - message.revision !== routeRevision || - Date.parse(message.expiresAt) <= now || - this.preparedRevision !== routeRevision || - this.preparedCandidate?.transport !== "selected-turn" || - this.preparedCandidate.childPeerId !== message.viewerPeerId || - this.preparedCandidate.connectionId !== message.newConnectionId - ) { - return false; - } - const prepared = this.preparedChild; - if (prepared?.peer.connectionId === message.newConnectionId) { - return !prepared.failed; - } - const stream = this.stream; - if (!stream) return false; - this.discardPreparedPeer(); - this.startPreparedChild( - message.revision, - this.preparedCandidate, - stream, - message, - ); - return true; - } - prepareChild( revision: number, candidate: PreparedRouteCandidate, @@ -338,13 +299,11 @@ export class ViewerRelay { revision: number, candidate: PreparedRouteCandidate, stream: MediaStream, - selectedTurn?: SelectedEdgeTurn, ): void { const peer = this.createPeer( candidate.childPeerId, stream, candidate.connectionId, - selectedTurn, ); this.preparedChild = { revision, @@ -484,14 +443,11 @@ export class ViewerRelay { childPeerId: string, stream: MediaStream, connectionId?: string, - selectedTurn?: SelectedEdgeTurn, ): HostPeer { let peer: HostPeer; peer = new HostPeer( childPeerId, - selectedTurn - ? { iceServers: [selectedTurn.iceServer] } - : this.iceConfig, + this.iceConfig, stream, this.desiredProfile, { @@ -524,7 +480,6 @@ export class ViewerRelay { } }, }, - selectedTurn !== undefined, connectionId, ); return peer; diff --git a/src/server/app.ts b/src/server/app.ts index da0f242..b3fef50 100644 --- a/src/server/app.ts +++ b/src/server/app.ts @@ -11,14 +11,10 @@ import { import { SiteAccess } from "./access-session.js"; import { loadConfig, type ServerConfig } from "./config.js"; import { createIceConfig } from "./ice.js"; -import type { - SelectedEdgeTurnOptions, - SfuFallbackOptions, -} from "./hybrid-media-router.js"; +import type { SfuFallbackOptions } from "./hybrid-media-router.js"; import type { SfuTokenIssuer } from "./livekit-token.js"; import type { SfuRoomControl } from "./sfu-room-control.js"; import { SfuResourceAdmission } from "./sfu-resource-admission.js"; -import { TurnAllocationAdmission } from "./turn-allocation-admission.js"; import { RoomStore, RoomStoreError } from "./room-store.js"; import { SignalingServer, type SignalingOptions } from "./signaling.js"; @@ -56,7 +52,6 @@ export async function createScreenerServer( const livekitFallback = config.livekitFallback; let sfuFallback: SfuFallbackOptions | undefined; let sfuRoomControl: SfuRoomControl | undefined; - let selectedEdgeTurn: SelectedEdgeTurnOptions | undefined; if (livekitFallback) { sfuRoomControl = options.sfuRoomControl ?? @@ -82,14 +77,6 @@ export async function createScreenerServer( roomControl: sfuRoomControl, }; } - if (config.selectedEdgeTurn) { - selectedEdgeTurn = { - config: config.selectedEdgeTurn, - admission: new TurnAllocationAdmission({ - capacity: config.selectedEdgeTurn.allocationCapacity, - }), - }; - } const roomStore = options.roomStore ?? new RoomStore({ @@ -147,7 +134,6 @@ export async function createScreenerServer( peerAssistedMedia: config.peerAssistedMedia, endpointMediaCopyCapacity: config.endpointMediaCopyCapacity, ...(sfuFallback ? { sfuFallback } : {}), - ...(selectedEdgeTurn ? { selectedEdgeTurn } : {}), ice: iceOptions, allowedOrigins: config.allowedOrigins, siteAccessAtUpgrade: (request) => diff --git a/src/server/config.ts b/src/server/config.ts index cec4a10..5d2a753 100644 --- a/src/server/config.ts +++ b/src/server/config.ts @@ -2,7 +2,6 @@ import { MAX_ICE_SERVER_URLS, MAX_VIEWERS_PER_ROOM_LIMIT, stunUrlSchema, - turnUrlSchema, } from "../shared/protocol.js"; import { DEFAULT_ENDPOINT_MEDIA_COPY_CAPACITY, @@ -14,10 +13,6 @@ export type RuntimeEnvironment = "development" | "test" | "production"; const MIN_SITE_ACCESS_PASSWORD_BYTES = 8; const MAX_SITE_ACCESS_PASSWORD_BYTES = 128; const MIN_LIVEKIT_API_SECRET_BYTES = 32; -const MIN_SELECTED_EDGE_TURN_SECRET_BYTES = 32; -const MAX_SELECTED_EDGE_TURN_SECRET_BYTES = 128; -const MIN_SELECTED_EDGE_TURN_TTL_SECONDS = 60; -const MAX_SELECTED_EDGE_TURN_TTL_SECONDS = 10 * 60; const DEFAULT_MAX_VIEWERS_PER_ROOM = 8; const DEFAULT_ROOM_LEASE_SECONDS = 86_400; const MAX_ROOMS = 9_000; @@ -29,6 +24,10 @@ const REMOVED_ENVIRONMENT_VARIABLES = [ "PEER_ICE_TURN_URLS", "PEER_ICE_TURN_SHARED_SECRET", "PEER_ICE_TURN_CREDENTIAL_TTL_SECONDS", + "SELECTED_EDGE_TURN_URLS", + "SELECTED_EDGE_TURN_SHARED_SECRET", + "SELECTED_EDGE_TURN_CREDENTIAL_TTL_SECONDS", + "SELECTED_EDGE_TURN_ALLOCATION_CAPACITY", "PEER_ASSISTED_ROOM_IDS", "HOST_ADMISSION_PASSWORD", "MAX_PEER_RELAY_DOWNSTREAM_EDGES", @@ -46,13 +45,6 @@ export interface LiveKitFallbackConfig { egressCapacity: number; } -export interface SelectedEdgeTurnConfig { - urls: readonly [string]; - sharedSecret: string; - credentialTtlSeconds: number; - allocationCapacity: number; -} - export interface ServerConfig { nodeEnv: RuntimeEnvironment; port: number; @@ -66,7 +58,6 @@ export interface ServerConfig { peerAssistedMedia: boolean; endpointMediaCopyCapacity: number; livekitFallback?: LiveKitFallbackConfig; - selectedEdgeTurn?: SelectedEdgeTurnConfig; stunUrls: readonly string[]; } @@ -281,60 +272,6 @@ function parseStunUrlList(value: string | undefined): string[] { }); } -export function parseSelectedEdgeTurn( - environment: NodeJS.ProcessEnv, -): SelectedEdgeTurnConfig | undefined { - const names = [ - "SELECTED_EDGE_TURN_URLS", "SELECTED_EDGE_TURN_SHARED_SECRET", - "SELECTED_EDGE_TURN_CREDENTIAL_TTL_SECONDS", - "SELECTED_EDGE_TURN_ALLOCATION_CAPACITY", - ] as const; - const configuredNames = names.filter((name) => environment[name]?.trim()); - if (configuredNames.length === 0) { - return undefined; - } - if (configuredNames.length !== names.length) { - throw new Error(`${names.join(", ")} must be configured together`); - } - - const urls = parseUrlList( - environment.SELECTED_EDGE_TURN_URLS, - "SELECTED_EDGE_TURN_URLS", - ); - if (urls.length !== 1 || !turnUrlSchema.safeParse(urls[0]).success) { - throw new Error( - "SELECTED_EDGE_TURN_URLS must contain one UDP TURN URL with transport=udp", - ); - } - const sharedSecret = environment.SELECTED_EDGE_TURN_SHARED_SECRET!; - const secretBytes = Buffer.byteLength(sharedSecret); - if ( - !VISIBLE_ASCII_PATTERN.test(sharedSecret) || - secretBytes < MIN_SELECTED_EDGE_TURN_SECRET_BYTES || - secretBytes > MAX_SELECTED_EDGE_TURN_SECRET_BYTES - ) { - throw new Error( - "SELECTED_EDGE_TURN_SHARED_SECRET must contain 32 to 128 visible ASCII bytes", - ); - } - - return { - urls: [urls[0]!], - sharedSecret, - credentialTtlSeconds: parseBoundedInteger( - environment.SELECTED_EDGE_TURN_CREDENTIAL_TTL_SECONDS, - MIN_SELECTED_EDGE_TURN_TTL_SECONDS, - "SELECTED_EDGE_TURN_CREDENTIAL_TTL_SECONDS", - MIN_SELECTED_EDGE_TURN_TTL_SECONDS, - MAX_SELECTED_EDGE_TURN_TTL_SECONDS, - ), - allocationCapacity: parseRequiredPositiveInteger( - environment.SELECTED_EDGE_TURN_ALLOCATION_CAPACITY, - "SELECTED_EDGE_TURN_ALLOCATION_CAPACITY", - ), - }; -} - function parseOrigins(value: string | undefined, fallback: string): Set { const origins = parseUrlList(value, "ALLOWED_ORIGINS"); return new Set((origins.length > 0 ? origins : [fallback]).map(toOrigin)); @@ -428,25 +365,18 @@ export function loadConfig( MAX_ENDPOINT_MEDIA_COPY_CAPACITY, ); const livekitFallback = parseLiveKitFallback(environment, nodeEnv); - const selectedEdgeTurn = parseSelectedEdgeTurn(environment); if (livekitFallback && !peerAssistedMedia) { throw new Error("LiveKit fallback requires PEER_ASSISTED_MEDIA=true"); } - if (selectedEdgeTurn && (!peerAssistedMedia || !livekitFallback)) { - throw new Error( - "Selected-edge TURN requires peer-assisted media and LiveKit fallback", - ); - } const configuredSecrets = [ siteAccessPassword, livekitFallback?.apiKey, livekitFallback?.apiSecret, - selectedEdgeTurn?.sharedSecret, ].filter((secret): secret is string => secret !== undefined); if (new Set(configuredSecrets).size !== configuredSecrets.length) { throw new Error( - "SITE_ACCESS_PASSWORD, LIVEKIT_API_KEY, LIVEKIT_API_SECRET, and SELECTED_EDGE_TURN_SHARED_SECRET must use independent values", + "SITE_ACCESS_PASSWORD, LIVEKIT_API_KEY, and LIVEKIT_API_SECRET must use independent values", ); } if ( @@ -494,7 +424,6 @@ export function loadConfig( peerAssistedMedia, endpointMediaCopyCapacity, livekitFallback, - selectedEdgeTurn, stunUrls, }; } diff --git a/src/server/hybrid-media-router.ts b/src/server/hybrid-media-router.ts index 04cccfe..f3cb9a7 100644 --- a/src/server/hybrid-media-router.ts +++ b/src/server/hybrid-media-router.ts @@ -11,7 +11,6 @@ import type { } from "../shared/protocol.js"; import { assertEndpointMediaCopyCapacity } from "../shared/media-copy-accounting.js"; import type { SfuTokenIssuer } from "./livekit-token.js"; -import type { SelectedEdgeTurnConfig } from "./config.js"; import { RoomRouteController, type CandidateCursorGuard, @@ -29,13 +28,6 @@ import type { SfuResourceFence, SfuSubscriptionFence, } from "./sfu-resource-admission.js"; -import { - issueSelectedEdgeTurnCredential, - type HostSfuIngressTurnIdentity, - type PeerSelectedEdgeTurnIdentity, - type SelectedEdgeTurnIdentity, -} from "./selected-edge-turn.js"; -import type { TurnAllocationAdmission } from "./turn-allocation-admission.js"; type ErrorCode = Extract["code"]; @@ -47,11 +39,6 @@ interface RouteResourceBase { released: boolean; } -interface TurnRouteResource extends RouteResourceBase { - kind: "turn"; - fence: SelectedEdgeTurnIdentity; -} - interface SfuSubscriptionRouteResource extends RouteResourceBase { kind: "sfu-subscription"; fence: SfuSubscriptionFence; @@ -60,7 +47,6 @@ interface SfuSubscriptionRouteResource extends RouteResourceBase { interface SfuPublicationRouteResource extends RouteResourceBase { kind: "sfu-publication"; fence: SfuResourceFence; - ingressTurn?: TurnRouteResource; } interface OverlapRouteResource extends RouteResourceBase { @@ -69,7 +55,6 @@ interface OverlapRouteResource extends RouteResourceBase { } type RouteResource = - | TurnRouteResource | SfuSubscriptionRouteResource | SfuPublicationRouteResource | OverlapRouteResource; @@ -101,20 +86,12 @@ interface PreparedCandidate { connectionId: string; publicationGeneration?: string; publicationConnectionId?: string; - peerGrant?: Extract< - ServerMessage, - { type: "selected-edge-turn"; edgeKind: "peer-selected" } - >; - hostIngressGrant?: Extract< - ServerMessage, - { type: "selected-edge-turn"; edgeKind: "host-sfu-ingress" } - >; hostSfuConfig?: Extract; viewerSfuConfig?: Extract; } type PrepareResult = - | { kind: "denied"; serverAdmission: "turn" | "sfu" | false } + | { kind: "denied"; serverAdmission: "sfu" | false } | { kind: "ready"; prepared: PreparedCandidate }; export interface SfuFallbackOptions { @@ -127,16 +104,10 @@ export interface SfuFallbackOptions { hostOfflineCheckMs?: number; } -export interface SelectedEdgeTurnOptions { - config: SelectedEdgeTurnConfig; - admission: TurnAllocationAdmission; -} - export interface HybridMediaRouterOptions { roomStore: RoomStore; endpointMediaCopyCapacity: number; sfuFallback?: SfuFallbackOptions; - selectedEdgeTurn?: SelectedEdgeTurnOptions; sendToSession: (sessionId: string, message: ServerMessage) => void; getConnectionId: (roomId: string, viewerPeerId: string) => string | undefined; setConnectionId: (roomId: string, viewerPeerId: string, connectionId: string) => void; @@ -178,9 +149,6 @@ export class HybridMediaRouter { constructor(private readonly options: HybridMediaRouterOptions) { assertEndpointMediaCopyCapacity(options.endpointMediaCopyCapacity); this.now = options.now ?? Date.now; - if (options.selectedEdgeTurn && !options.sfuFallback) { - throw new Error("Selected-edge TURN requires SFU fallback"); - } if (options.sfuFallback) new URL(options.sfuFallback.url); } @@ -190,12 +158,6 @@ export class HybridMediaRouter { this.resourceWaiterRoomIds.clear(); for (const check of this.hostOfflineChecks.values()) clearTimeout(check.timer); this.hostOfflineChecks.clear(); - const turnAdmission = this.options.selectedEdgeTurn?.admission; - if (turnAdmission) { - for (const fence of turnAdmission.beginDrainAll()) { - turnAdmission.completeDrain(fence); - } - } const fallback = this.options.sfuFallback; if (!fallback) return; for (const fence of fallback.admission.beginDrainAll()) this.scheduleSfuDrain(fence); @@ -319,7 +281,7 @@ export class HybridMediaRouter { ); } - selectedEdgeTurnSignalAuthorization(input: { + peerSignalAuthorization(input: { roomId: string; sourcePeerId: string; sourceSessionId: string; @@ -393,14 +355,6 @@ export class HybridMediaRouter { return input.connectionId === edge.connectionId; } - markSelectedEdgeTurnAnswered( - _roomId: string, - _viewerPeerId: string, - _connectionId: string, - ): void { - // Resource admission commits only with exact child media proof. - } - setViewerRelayCapacity( participant: AuthenticatedRouteParticipant, downstreamEdges: number, @@ -637,8 +591,6 @@ export class HybridMediaRouter { endpointMediaCopyCapacity: this.options.endpointMediaCopyCapacity, operationTimeoutMs: this.options.sfuFallback?.prepareTimeoutMs ?? DEFAULT_ROUTE_OPERATION_TIMEOUT_MS, - selectedTurnEnabled: Boolean(this.options.selectedEdgeTurn), - hostSfuSelectedTurnEnabled: Boolean(this.options.selectedEdgeTurn), sfuEnabled: Boolean(this.options.sfuFallback), }); room.controller.upsertParticipant({ @@ -723,9 +675,6 @@ export class HybridMediaRouter { "No usable media route is available", ); } - if (preparation.serverAdmission === "turn" && skipped.exhausted) { - this.resourceWaiterRoomIds.add(roomId); - } continue; } let currentOperation = controller.snapshot().operation; @@ -795,64 +744,13 @@ export class HybridMediaRouter { ? overlapResource(plan.endpointTransition.producerPeerId) : undefined; if (plan.tuple.kind === "peer") { - if (plan.tuple.transport === "direct") { - return { - kind: "ready", - prepared: { - connectionId, - reservation: { kind: "direct", ...(overlap ? { overlap } : {}) }, - }, - }; - } - const parent = this.connectedPeer(roomId, plan.tuple.parentPeerId); - const selected = this.options.selectedEdgeTurn; - if (!parent || !selected) return { kind: "denied", serverAdmission: false }; - const oldConnectionId = - snapshot.upstreamByViewer.get(operation.childPeerId)?.connectionId ?? connectionId; - const fence: PeerSelectedEdgeTurnIdentity = { - edgeKind: "peer-selected", - roomId, - shareGeneration, - revision, - parentPeerId: parent.peerId, - parentSessionId: parent.sessionId, - viewerPeerId: operation.childPeerId, - viewerSessionId: operation.childSessionId, - oldConnectionId, - newConnectionId: connectionId, + return { + kind: "ready", + prepared: { + connectionId, + reservation: { kind: "direct", ...(overlap ? { overlap } : {}) }, + }, }; - if (!selected.admission.reserve(fence)) { - return { kind: "denied", serverAdmission: "turn" }; - } - const resource = turnResource(fence); - try { - const credential = issueSelectedEdgeTurnCredential(selected.config, fence, this.now()); - return { - kind: "ready", - prepared: { - connectionId, - reservation: { - kind: "selected-turn", - edge: resource, - ...(overlap ? { overlap } : {}), - }, - peerGrant: { - type: "selected-edge-turn", - edgeKind: "peer-selected", - revision, - parentPeerId: parent.peerId, - viewerPeerId: operation.childPeerId, - oldConnectionId, - newConnectionId: connectionId, - expiresAt: credential.expiresAt, - iceServer: credential.iceServer, - }, - }, - }; - } catch { - this.releaseResource(resource); - return { kind: "denied", serverAdmission: false }; - } } const fallback = this.options.sfuFallback; @@ -918,9 +816,7 @@ export class HybridMediaRouter { return await this.prepareNewPublication( roomId, - room, operation, - plan, revision, connectionId, shareGeneration, @@ -931,13 +827,11 @@ export class HybridMediaRouter { private async prepareNewPublication( roomId: string, - room: RoomRuntime, operation: OperationSnapshot, - plan: CandidatePlan, revision: number, connectionId: string, shareGeneration: string, - host: { peerId: string; sessionId: string }, + host: { peerId: string }, overlap?: OverlapRouteResource, ): Promise { const fallback = this.options.sfuFallback!; @@ -960,61 +854,8 @@ export class HybridMediaRouter { return { kind: "denied", serverAdmission: "sfu" }; } const subscription = subscriptionResource(subscriptionFence); - let ingressTurn: TurnRouteResource | undefined; - let hostIngressGrant: PreparedCandidate["hostIngressGrant"]; - const publicationConnectionId = - plan.tuple.kind === "sfu" && plan.tuple.ingress === "selected-turn" - ? connectionId - : publicationGeneration; - if (plan.tuple.kind === "sfu" && plan.tuple.ingress === "selected-turn") { - const selected = this.options.selectedEdgeTurn; - if (!selected) { - this.releaseResource(subscription); - fallback.admission.beginDrain(publicationFence); - this.scheduleSfuDrain(publicationFence); - return { kind: "denied", serverAdmission: false }; - } - const fence: HostSfuIngressTurnIdentity = { - edgeKind: "host-sfu-ingress", - roomId, - shareGeneration, - revision, - hostPeerId: host.peerId, - hostSessionId: host.sessionId, - publicationGeneration, - oldConnectionId: - room.controller?.snapshot().hostPublication?.connectionId ?? publicationGeneration, - newConnectionId: connectionId, - }; - if (!selected.admission.reserve(fence)) { - this.releaseResource(subscription); - fallback.admission.beginDrain(publicationFence); - this.scheduleSfuDrain(publicationFence); - return { kind: "denied", serverAdmission: "turn" }; - } - ingressTurn = turnResource(fence); - try { - const credential = issueSelectedEdgeTurnCredential(selected.config, fence, this.now()); - hostIngressGrant = { - type: "selected-edge-turn", - edgeKind: "host-sfu-ingress", - revision, - hostPeerId: host.peerId, - publicationGeneration, - oldConnectionId: fence.oldConnectionId, - newConnectionId: connectionId, - expiresAt: credential.expiresAt, - iceServer: credential.iceServer, - }; - } catch { - this.releaseResource(ingressTurn); - this.releaseResource(subscription); - fallback.admission.beginDrain(publicationFence); - this.scheduleSfuDrain(publicationFence); - return { kind: "denied", serverAdmission: false }; - } - } - const publication = publicationResource(publicationFence, ingressTurn); + const publicationConnectionId = publicationGeneration; + const publication = publicationResource(publicationFence); try { await fallback.roomControl.createRoom(publicationFence); const [hostToken, viewerToken] = await Promise.all([ @@ -1047,7 +888,6 @@ export class HybridMediaRouter { publication, ...(overlap ? { overlap } : {}), }, - hostIngressGrant, hostSfuConfig: { type: "sfu-config", revision, @@ -1097,7 +937,6 @@ export class HybridMediaRouter { const parent = this.connectedPeer(roomId, current.tuple.parentPeerId); if (!parent) return; this.options.sendToSession(child.sessionId, childUpdate); - if (prepared.peerGrant) this.options.sendToSession(child.sessionId, prepared.peerGrant); this.options.sendToSession(parent.sessionId, { type: "route-update", revision: current.revision, @@ -1105,7 +944,6 @@ export class HybridMediaRouter { assignment: assignments.get(parent.peerId) ?? emptyAssignment(), candidate, }); - if (prepared.peerGrant) this.options.sendToSession(parent.sessionId, prepared.peerGrant); return; } const host = this.options.roomStore.getConnectedHost(roomId); @@ -1117,9 +955,6 @@ export class HybridMediaRouter { assignment: assignments.get(host.peerId) ?? emptyAssignment(), candidate, }); - if (prepared.hostIngressGrant) { - this.options.sendToSession(host.sessionId, prepared.hostIngressGrant); - } this.options.sendToSession(host.sessionId, prepared.hostSfuConfig); } this.options.sendToSession(child.sessionId, childUpdate); @@ -1314,10 +1149,6 @@ export class HybridMediaRouter { reservation: CandidateReservation, ): boolean { if (reservation.kind === "direct") return true; - if (reservation.kind === "selected-turn") { - return reservation.edge.kind === "turn" && - this.options.selectedEdgeTurn?.admission.commit(reservation.edge.fence) === true; - } if (reservation.kind === "sfu-reuse") { return reservation.edge.kind === "sfu-subscription" && this.options.sfuFallback?.admission.commitSubscription( @@ -1325,14 +1156,6 @@ export class HybridMediaRouter { ) === true; } if (reservation.publication.kind !== "sfu-publication") return false; - if ( - reservation.publication.ingressTurn && - this.options.selectedEdgeTurn?.admission.commit( - reservation.publication.ingressTurn.fence, - ) !== true - ) { - return false; - } const draining = this.options.sfuFallback?.admission.commitPublication( reservation.publication.fence, ); @@ -1353,19 +1176,10 @@ export class HybridMediaRouter { if (resource.released) return; resource.released = true; if (resource.kind === "overlap") return; - if (resource.kind === "turn") { - const admission = this.options.selectedEdgeTurn?.admission; - if (admission?.beginDrain(resource.fence)) { - admission.completeDrain(resource.fence); - this.wakeResourceWaiters(); - } - return; - } if (resource.kind === "sfu-subscription") { this.options.sfuFallback?.admission.releaseSubscription(resource.fence); return; } - if (resource.ingressTurn) this.releaseResource(resource.ingressTurn); if (this.options.sfuFallback?.admission.beginDrain(resource.fence)) { this.scheduleSfuDrain(resource.fence); } @@ -1585,10 +1399,6 @@ export class HybridMediaRouter { for (const fence of this.options.sfuFallback?.admission.beginDrainRoom(roomId) ?? []) { this.scheduleSfuDrain(fence); } - const turnAdmission = this.options.selectedEdgeTurn?.admission; - for (const fence of turnAdmission?.beginDrainRoom(roomId) ?? []) { - turnAdmission!.completeDrain(fence); - } this.wakeResourceWaiters(); } @@ -1658,10 +1468,6 @@ function overlapResource(endpointPeerId: string): OverlapRouteResource { return { kind: "overlap", endpointPeerId, released: false }; } -function turnResource(fence: SelectedEdgeTurnIdentity): TurnRouteResource { - return { kind: "turn", fence, released: false }; -} - function subscriptionResource( fence: SfuSubscriptionFence, ): SfuSubscriptionRouteResource { @@ -1670,13 +1476,11 @@ function subscriptionResource( function publicationResource( fence: SfuResourceFence, - ingressTurn?: TurnRouteResource, ): SfuPublicationRouteResource { return { kind: "sfu-publication", fence, released: false, - ...(ingressTurn ? { ingressTurn } : {}), }; } diff --git a/src/server/room-route-controller.ts b/src/server/room-route-controller.ts index 4ad9b79..36a77bb 100644 --- a/src/server/room-route-controller.ts +++ b/src/server/room-route-controller.ts @@ -2,12 +2,11 @@ import { MAX_MEDIA_ROUTE_REVISION } from "../shared/protocol.js"; import { assertEndpointMediaCopyCapacity } from "../shared/media-copy-accounting.js"; export type CandidateTuple = - | { kind: "peer"; parentPeerId: string; transport: "direct" | "selected-turn" } - | { kind: "sfu"; publication: "reuse" | "create" | "replace"; ingress: "existing" | "direct" | "selected-turn" }; + | { kind: "peer"; parentPeerId: string; transport: "direct" } + | { kind: "sfu"; publication: "reuse" | "create" | "replace" }; export type CandidateReservation = | { kind: "direct"; overlap?: Resource } - | { kind: "selected-turn"; edge: Resource; overlap?: Resource } | { kind: "sfu-reuse"; edge: Resource; overlap?: Resource } | { kind: "sfu-create"; edge: Resource; publication: Resource; overlap?: Resource }; @@ -18,7 +17,7 @@ export type EndpointRetirement = childSessionId: string; parentPeerId: string; parentSessionId: string; - transport: "direct" | "selected-turn"; + transport: "direct"; connectionId: string; } | { @@ -40,12 +39,10 @@ export interface CandidatePlan { export type CommittedEdge = | { kind: "peer"; childSessionId: string; parentPeerId: string; parentSessionId: string; transport: "direct"; connectionId: string; usable: boolean; physicalActive: boolean } - | { kind: "peer"; childSessionId: string; parentPeerId: string; parentSessionId: string; transport: "selected-turn"; connectionId: string; usable: boolean; physicalActive: boolean; resource: Resource } | { kind: "sfu"; childSessionId: string; publicationGeneration: string; transport: "sfu"; connectionId: string; usable: boolean; physicalActive: boolean; resource: Resource }; export type CommittedEdgeSeed = | Omit, { transport: "direct" }>, "childSessionId" | "parentSessionId"> - | Omit, { transport: "selected-turn" }>, "childSessionId" | "parentSessionId"> | Omit, { transport: "sfu" }>, "childSessionId">; export interface ParticipantInput { @@ -83,8 +80,6 @@ export interface ControllerOptions { hostPeerId: string; endpointMediaCopyCapacity: number; operationTimeoutMs: number; - selectedTurnEnabled?: boolean; - hostSfuSelectedTurnEnabled?: boolean; sfuEnabled?: boolean; } @@ -166,11 +161,9 @@ interface HostPublication { generation: string; hostSessionId: string; connectionId: string; - ingress: "direct" | "selected-turn"; usable: boolean; physicalActive: boolean; resource: Resource; - failedIngress?: { ingress: "direct" | "selected-turn"; factVersion: number }; } export class RoomRouteController { @@ -306,7 +299,6 @@ export class RoomRouteController { generation: string, resource: Resource, connectionId = `publication:${generation}`, - ingress: "direct" | "selected-turn" = "direct", ): void { if (this.operation) throw new Error("Cannot hydrate while an operation is active"); const hostSessionId = this.participants.get(this.options.hostPeerId)?.sessionId; @@ -315,7 +307,6 @@ export class RoomRouteController { generation, hostSessionId, connectionId, - ingress, usable: true, physicalActive: true, resource, @@ -358,10 +349,6 @@ export class RoomRouteController { if (publication.usable) { publication.usable = false; this.touchFacts(); - publication.failedIngress = { - ingress: publication.ingress, - factVersion: this.factVersion, - }; } return true; } @@ -580,7 +567,6 @@ export class RoomRouteController { transport: edge.transport, }; } - if (edge.transport === "selected-turn") released.push(edge.resource); this.upstreamByViewer.delete(retirement.childPeerId); } else { const publication = this.hostPublication; @@ -711,7 +697,8 @@ export class RoomRouteController { if (operation.cursor < operation.candidates.length) return result; const factsChanged = operation.builtAtFactVersion !== this.factVersion; this.blockAndClear(operation, !factsChanged, released); - result.exhausted = !factsChanged; + result.exhausted = + !factsChanged && !this.bootstrapForBlockedDemand(); } return result; } @@ -727,15 +714,10 @@ export class RoomRouteController { if (attempt.tuple.kind === "peer") { const parentSessionId = this.participants.get(attempt.tuple.parentPeerId)?.sessionId; if (!parentSessionId) throw new Error("Candidate parent session is unavailable"); - this.upstreamByViewer.set(operation.childPeerId, attempt.tuple.transport === "direct" ? { + this.upstreamByViewer.set(operation.childPeerId, { kind: "peer", childSessionId: operation.childSessionId, parentPeerId: attempt.tuple.parentPeerId, parentSessionId, transport: "direct", connectionId: attempt.connectionId, usable: true, physicalActive: true, - } : { - kind: "peer", childSessionId: operation.childSessionId, - parentPeerId: attempt.tuple.parentPeerId, parentSessionId, transport: "selected-turn", - connectionId: attempt.connectionId, usable: true, physicalActive: true, - resource: (attempt.reservation as Extract, { kind: "selected-turn" }>).edge, }); } else { const generation = attempt.publicationGeneration!; @@ -749,7 +731,6 @@ export class RoomRouteController { generation, hostSessionId, connectionId: attempt.publicationConnectionId, - ingress: attempt.tuple.ingress === "selected-turn" ? "selected-turn" : "direct", usable: true, physicalActive: true, resource: reservation.publication, @@ -787,9 +768,6 @@ export class RoomRouteController { const failedKey = child.failedTuple?.factVersion === this.factVersion ? child.failedTuple.key : undefined; - const failedIngress = this.hostPublication?.failedIngress?.factVersion === this.factVersion - ? this.hostPublication.failedIngress.ingress - : undefined; const descendants = this.descendantsOf(childPeerId); const parents = [...this.participants.values()] .filter((parent) => parent.sessionId && !parent.departureConfirmed && parent.peerId !== childPeerId && @@ -800,25 +778,16 @@ export class RoomRouteController { const candidates: CandidateTuple[] = []; if (!sfuOnly) { for (const parent of parents) candidates.push({ kind: "peer", parentPeerId: parent.peerId, transport: "direct" }); - if (this.options.selectedTurnEnabled) { - for (const parent of parents) if (parent.role !== "host") { - candidates.push({ kind: "peer", parentPeerId: parent.peerId, transport: "selected-turn" }); - } - } } if (this.options.sfuEnabled) { if (this.hostPublication?.usable && this.hostPublication.physicalActive) { - candidates.push({ kind: "sfu", publication: "reuse", ingress: "existing" }); + candidates.push({ kind: "sfu", publication: "reuse" }); } else if (this.hostPublication || this.publicationFits(childPeerId, false)) { const publication = this.hostPublication ? "replace" : "create"; - candidates.push({ kind: "sfu", publication, ingress: "direct" }); - if (this.options.hostSfuSelectedTurnEnabled) { - candidates.push({ kind: "sfu", publication, ingress: "selected-turn" }); - } + candidates.push({ kind: "sfu", publication }); } } return candidates.filter((candidate, index, all) => tupleKey(candidate) !== failedKey && - !(candidate.kind === "sfu" && candidate.publication !== "reuse" && candidate.ingress === failedIngress) && all.findIndex((other) => tupleKey(other) === tupleKey(candidate)) === index) .map((candidate) => this.planCandidate(childPeerId, candidate)) .filter((candidate): candidate is CandidatePlan => candidate !== null && @@ -971,7 +940,7 @@ export class RoomRouteController { } private assertReservation(tuple: CandidateTuple, reservation: CandidateReservation, overlap: boolean): void { - const expected = tuple.kind === "peer" ? tuple.transport === "direct" ? "direct" : "selected-turn" : + const expected = tuple.kind === "peer" ? "direct" : tuple.publication === "reuse" ? "sfu-reuse" : "sfu-create"; if (reservation.kind !== expected) throw new Error("Candidate reservation kind does not match tuple"); if (overlap && !("overlap" in reservation && reservation.overlap !== undefined)) { @@ -1036,7 +1005,7 @@ export class RoomRouteController { private retireInvalidOperationEdge(childPeerId: string, released: Resource[]): boolean { const edge = this.upstreamByViewer.get(childPeerId); if (!edge || !this.edgeRequiresMove(childPeerId, edge)) return false; - if (edge.transport !== "direct" && edge.physicalActive) released.push(edge.resource); + if (edge.kind === "sfu" && edge.physicalActive) released.push(edge.resource); edge.physicalActive = false; edge.usable = false; if (edge.kind === "sfu" && !this.hasSfuSubscribers() && this.hostPublication) { @@ -1139,7 +1108,7 @@ export class RoomRouteController { operation: ChildOperation, stage: CandidateStage, ): number { - const stages = candidateStages(operation.candidates); + const stages = this.operationStages(operation); const index = stages.indexOf(stage); if (index === -1 || index === stages.length - 1) { return operation.deadlineAtMs; @@ -1154,6 +1123,40 @@ export class RoomRouteController { ); } + private operationStages( + operation: ChildOperation, + ): CandidateStage[] { + const stages = candidateStages(operation.candidates); + if ( + !stages.includes("sfu") && + this.bootstrapCandidateAvailable() + ) { + stages.push("sfu"); + } + return stages; + } + + private bootstrapCandidateAvailable(): boolean { + if ( + !this.options.sfuEnabled || + this.hostPublication || + this.hostHasPublicationSlot() + ) { + return false; + } + return this.childrenOf(this.options.hostPeerId).some((id) => { + const participant = this.participants.get(id); + const edge = this.upstreamByViewer.get(id); + return Boolean( + participant?.sessionId && + edge?.kind === "peer" && + edge.transport === "direct" && + edge.usable && + edge.physicalActive, + ); + }); + } + private guardMatches(guard: CandidateGuard, operation: ChildOperation, attempt: Attempt): boolean { return guard.childPeerId === operation.childPeerId && guard.childSessionId === operation.childSessionId && guard.revision === attempt.revision && guard.connectionId === attempt.connectionId; @@ -1257,7 +1260,7 @@ export class RoomRouteController { private committedResources(): Resource[] { const resources: Resource[] = []; for (const edge of this.upstreamByViewer.values()) { - if (edge.transport !== "direct" && edge.physicalActive) resources.push(edge.resource); + if (edge.kind === "sfu" && edge.physicalActive) resources.push(edge.resource); } if (this.hostPublication?.physicalActive) resources.push(this.hostPublication.resource); return resources; @@ -1268,38 +1271,18 @@ export class RoomRouteController { sessionId: string, ): { released: Resource[]; retired: boolean } { const released = new Set(); - let retired = false; const ownEdge = this.upstreamByViewer.get(peerId); - if (ownEdge?.kind === "peer" && ownEdge.transport === "selected-turn") { - if (ownEdge.physicalActive) released.add(ownEdge.resource); - ownEdge.physicalActive = false; - ownEdge.usable = false; - retired = true; - } else if (ownEdge) { + if (ownEdge) { ownEdge.childSessionId = sessionId; } for (const edge of this.upstreamByViewer.values()) { if (edge.kind !== "peer" || edge.parentPeerId !== peerId) continue; - if (edge.transport === "selected-turn") { - if (edge.physicalActive) released.add(edge.resource); - edge.physicalActive = false; - edge.usable = false; - retired = true; - } else { - edge.parentSessionId = sessionId; - } + edge.parentSessionId = sessionId; } if (peerId === this.options.hostPeerId && this.hostPublication) { - if (this.hostPublication.ingress === "selected-turn") { - if (this.hostPublication.physicalActive) released.add(this.hostPublication.resource); - this.hostPublication.physicalActive = false; - this.hostPublication.usable = false; - retired = true; - } else { - this.hostPublication.hostSessionId = sessionId; - } + this.hostPublication.hostSessionId = sessionId; } - return { released: [...released], retired }; + return { released: [...released], retired: false }; } private removePublicationGeneration(generation: string): Resource[] { @@ -1362,29 +1345,29 @@ function compareParticipant(left: Participant, right: Participant): number { } function tupleKey(tuple: CandidateTuple): string { - return tuple.kind === "peer" ? `peer:${tuple.parentPeerId}:${tuple.transport}` : `sfu:${tuple.publication}:${tuple.ingress}`; + return tuple.kind === "peer" ? `peer:${tuple.parentPeerId}` : `sfu:${tuple.publication}`; } -type CandidateStage = "direct" | "selected-turn" | "sfu"; +type CandidateStage = "direct" | "sfu"; function candidateStage(tuple: CandidateTuple): CandidateStage { if (tuple.kind === "sfu") { return "sfu"; } - return tuple.transport; + return "direct"; } function candidateStages(candidates: readonly CandidatePlan[]): CandidateStage[] { const stages = new Set( candidates.map((candidate) => candidateStage(candidate.tuple)), ); - return (["direct", "selected-turn", "sfu"] as const).filter((stage) => + return (["direct", "sfu"] as const).filter((stage) => stages.has(stage), ); } function edgeTupleKey(edge: CommittedEdge): string { - return edge.kind === "peer" ? `peer:${edge.parentPeerId}:${edge.transport}` : "sfu:reuse:existing"; + return edge.kind === "peer" ? `peer:${edge.parentPeerId}` : "sfu:reuse"; } function cloneCandidatePlan(plan: CandidatePlan): CandidatePlan { diff --git a/src/server/selected-edge-turn.ts b/src/server/selected-edge-turn.ts deleted file mode 100644 index 9986b52..0000000 --- a/src/server/selected-edge-turn.ts +++ /dev/null @@ -1,90 +0,0 @@ -import { createHmac } from "node:crypto"; - -import type { SelectedEdgeTurnConfig } from "./config.js"; - -export interface PeerSelectedEdgeTurnIdentity { - edgeKind: "peer-selected"; - roomId: string; - shareGeneration: string; - revision: number; - parentPeerId: string; - parentSessionId: string; - viewerPeerId: string; - viewerSessionId: string; - oldConnectionId: string; - newConnectionId: string; -} - -export interface HostSfuIngressTurnIdentity { - edgeKind: "host-sfu-ingress"; - roomId: string; - shareGeneration: string; - revision: number; - hostPeerId: string; - hostSessionId: string; - publicationGeneration: string; - oldConnectionId: string; - newConnectionId: string; -} - -export type SelectedEdgeTurnIdentity = - | PeerSelectedEdgeTurnIdentity - | HostSfuIngressTurnIdentity; - -export function issueSelectedEdgeTurnCredential( - config: SelectedEdgeTurnConfig, - identity: SelectedEdgeTurnIdentity, - nowMs: number, -) { - if (!Number.isFinite(nowMs) || nowMs < 0) { - throw new Error("Selected-edge TURN grant time is invalid"); - } - const expiresAtSeconds = - Math.floor(nowMs / 1_000) + config.credentialTtlSeconds; - const opaquePayload = - identity.edgeKind === "peer-selected" - ? [ - "screener-selected-edge-turn-v1", - identity.roomId, - identity.shareGeneration, - identity.revision, - identity.parentPeerId, - identity.parentSessionId, - identity.viewerPeerId, - identity.viewerSessionId, - identity.oldConnectionId, - identity.newConnectionId, - expiresAtSeconds, - ] - : [ - "screener-selected-edge-turn-v2", - identity.edgeKind, - identity.roomId, - identity.shareGeneration, - identity.revision, - identity.hostPeerId, - identity.hostSessionId, - identity.publicationGeneration, - identity.oldConnectionId, - identity.newConnectionId, - expiresAtSeconds, - ]; - const opaqueIdentity = createHmac("sha256", config.sharedSecret) - .update(JSON.stringify(opaquePayload)) - .digest("base64url") - .slice(0, 32); - const username = `${expiresAtSeconds}:${opaqueIdentity}`; - const credential = createHmac("sha1", config.sharedSecret) - .update(username) - .digest("base64"); - const expiresAtMs = expiresAtSeconds * 1_000; - - return { - iceServer: { - urls: [...config.urls] as [string], - username, - credential, - }, - expiresAt: new Date(expiresAtMs).toISOString(), - }; -} diff --git a/src/server/signaling.ts b/src/server/signaling.ts index 9205de3..c3de900 100644 --- a/src/server/signaling.ts +++ b/src/server/signaling.ts @@ -30,7 +30,6 @@ import { } from "./room-store.js"; import { HybridMediaRouter, - type SelectedEdgeTurnOptions, type SfuFallbackOptions, } from "./hybrid-media-router.js"; import { createIceConfig, type IceConfigOptions } from "./ice.js"; @@ -88,7 +87,6 @@ export interface SignalingOptions { peerAssistedMedia: boolean; endpointMediaCopyCapacity: number; sfuFallback?: SfuFallbackOptions; - selectedEdgeTurn?: SelectedEdgeTurnOptions; ice: IceConfigOptions; allowedOrigins: ReadonlySet; siteAccessAtUpgrade: (request: IncomingMessage) => boolean; @@ -167,7 +165,6 @@ export class SignalingServer { roomStore: options.roomStore, endpointMediaCopyCapacity: options.endpointMediaCopyCapacity, sfuFallback: options.sfuFallback, - selectedEdgeTurn: options.selectedEdgeTurn, sendToSession: (sessionId, message) => this.sendToSession(sessionId, message), getConnectionId: (roomId, viewerPeerId) => @@ -1417,8 +1414,8 @@ export class SignalingServer { message.payload.kind === "description" ? message.payload.description : undefined; - const selectedEdgeAuthorized = - this.hybridMediaRouter!.selectedEdgeTurnSignalAuthorization({ + const candidateAuthorized = + this.hybridMediaRouter!.peerSignalAuthorization({ roomId: source.roomId, sourcePeerId: source.peerId, sourceSessionId: this.socketStates.get(sourceSocket)?.sessionId ?? "", @@ -1433,7 +1430,7 @@ export class SignalingServer { ? parentToChild : childToParent : parentToChild || childToParent; - const authorized = selectedEdgeAuthorized ?? assignedEdgeAuthorized; + const authorized = candidateAuthorized ?? assignedEdgeAuthorized; if (!authorized) { this.sendError( sourceSocket, @@ -1445,7 +1442,7 @@ export class SignalingServer { if ( description?.type === "offer" && - selectedEdgeAuthorized !== "probe" + candidateAuthorized !== "probe" ) { this.setViewerConnectionId( source.roomId, @@ -1458,13 +1455,6 @@ export class SignalingServer { fromPeerId: source.peerId, payload: message.payload, }); - if (selectedEdgeAuthorized && description?.type === "answer") { - this.hybridMediaRouter!.markSelectedEdgeTurnAnswered( - source.roomId, - source.peerId, - message.payload.connectionId, - ); - } } private routePeerAssistedRestart( diff --git a/src/server/turn-allocation-admission.ts b/src/server/turn-allocation-admission.ts deleted file mode 100644 index 1cb4a29..0000000 --- a/src/server/turn-allocation-admission.ts +++ /dev/null @@ -1,214 +0,0 @@ -import type { SelectedEdgeTurnIdentity } from "./selected-edge-turn.js"; - -export interface TurnAllocationAdmissionOptions { - readonly capacity: number; -} - -export interface TurnAllocationUsage { - allocations: number; -} - -export type TurnAllocationState = "reserved" | "committed" | "draining"; - -interface TurnAllocationEntry { - fence: SelectedEdgeTurnIdentity; - state: TurnAllocationState; -} - -export class TurnAllocationAdmission { - private readonly entries = new Map(); - private readonly allocationIdsByRoom = new Map>(); - private readonly capacity: number; - - constructor(options: TurnAllocationAdmissionOptions) { - assertPositiveSafeInteger(options.capacity, "TURN allocation capacity"); - this.capacity = options.capacity; - } - - reserve(fence: SelectedEdgeTurnIdentity): boolean { - assertFence(fence); - const allocationId = fence.newConnectionId; - const existing = this.entries.get(allocationId); - if (existing) { - return existing.state !== "draining" && sameFence(existing.fence, fence); - } - if (this.entries.size >= this.capacity) { - return false; - } - - this.entries.set(allocationId, { - fence: cloneFence(fence), - state: "reserved", - }); - let roomAllocationIds = this.allocationIdsByRoom.get(fence.roomId); - if (!roomAllocationIds) { - roomAllocationIds = new Set(); - this.allocationIdsByRoom.set(fence.roomId, roomAllocationIds); - } - roomAllocationIds.add(allocationId); - return true; - } - - commit(fence: SelectedEdgeTurnIdentity): boolean { - assertFence(fence); - const entry = this.entries.get(fence.newConnectionId); - if (!entry || !sameFence(entry.fence, fence)) { - return false; - } - if (entry.state === "committed") { - return true; - } - if (entry.state !== "reserved") { - return false; - } - entry.state = "committed"; - return true; - } - - beginDrain(fence: SelectedEdgeTurnIdentity): boolean { - assertFence(fence); - const entry = this.entries.get(fence.newConnectionId); - if (!entry || !sameFence(entry.fence, fence)) { - return false; - } - entry.state = "draining"; - return true; - } - - completeDrain(fence: SelectedEdgeTurnIdentity): boolean { - assertFence(fence); - const allocationId = fence.newConnectionId; - const entry = this.entries.get(allocationId); - if ( - !entry || - entry.state !== "draining" || - !sameFence(entry.fence, fence) - ) { - return false; - } - - this.entries.delete(allocationId); - const roomAllocationIds = this.allocationIdsByRoom.get(fence.roomId); - roomAllocationIds?.delete(allocationId); - if (roomAllocationIds?.size === 0) { - this.allocationIdsByRoom.delete(fence.roomId); - } - return true; - } - - beginDrainRoom(roomId: string): readonly SelectedEdgeTurnIdentity[] { - if (!roomId) { - throw new Error("TURN allocation room ID is invalid"); - } - const allocationIds = this.allocationIdsByRoom.get(roomId); - if (!allocationIds) { - return []; - } - const fences: SelectedEdgeTurnIdentity[] = []; - for (const allocationId of allocationIds) { - const entry = this.entries.get(allocationId); - if (!entry) { - throw new Error("TURN allocation room index is inconsistent"); - } - entry.state = "draining"; - fences.push(cloneFence(entry.fence)); - } - return fences; - } - - beginDrainAll(): readonly SelectedEdgeTurnIdentity[] { - return [...this.allocationIdsByRoom.keys()].flatMap((roomId) => - this.beginDrainRoom(roomId), - ); - } - - state(fence: SelectedEdgeTurnIdentity): TurnAllocationState | undefined { - assertFence(fence); - const entry = this.entries.get(fence.newConnectionId); - return entry && sameFence(entry.fence, fence) ? entry.state : undefined; - } - - usage(): TurnAllocationUsage { - return { allocations: this.entries.size }; - } -} - -function cloneFence( - fence: SelectedEdgeTurnIdentity, -): SelectedEdgeTurnIdentity { - return { ...fence }; -} - -function sameFence( - left: SelectedEdgeTurnIdentity, - right: SelectedEdgeTurnIdentity, -): boolean { - if (left.edgeKind !== right.edgeKind) { - return false; - } - if (left.edgeKind === "peer-selected" && right.edgeKind === "peer-selected") { - return ( - left.roomId === right.roomId && - left.shareGeneration === right.shareGeneration && - left.revision === right.revision && - left.parentPeerId === right.parentPeerId && - left.parentSessionId === right.parentSessionId && - left.viewerPeerId === right.viewerPeerId && - left.viewerSessionId === right.viewerSessionId && - left.oldConnectionId === right.oldConnectionId && - left.newConnectionId === right.newConnectionId - ); - } - if ( - left.edgeKind === "host-sfu-ingress" && - right.edgeKind === "host-sfu-ingress" - ) { - return ( - left.roomId === right.roomId && - left.shareGeneration === right.shareGeneration && - left.revision === right.revision && - left.hostPeerId === right.hostPeerId && - left.hostSessionId === right.hostSessionId && - left.publicationGeneration === right.publicationGeneration && - left.oldConnectionId === right.oldConnectionId && - left.newConnectionId === right.newConnectionId - ); - } - return false; -} - -function assertFence(fence: SelectedEdgeTurnIdentity): void { - const commonValues = [ - fence.roomId, - fence.shareGeneration, - fence.oldConnectionId, - fence.newConnectionId, - ]; - const edgeValues = - fence.edgeKind === "peer-selected" - ? [ - fence.parentPeerId, - fence.parentSessionId, - fence.viewerPeerId, - fence.viewerSessionId, - ] - : [ - fence.hostPeerId, - fence.hostSessionId, - fence.publicationGeneration, - ]; - if ( - commonValues.some((value) => value.length === 0) || - edgeValues.some((value) => value.length === 0) || - !Number.isSafeInteger(fence.revision) || - fence.revision < 0 - ) { - throw new Error("TURN allocation fence is invalid"); - } -} - -function assertPositiveSafeInteger(value: number, name: string): void { - if (!Number.isSafeInteger(value) || value <= 0) { - throw new Error(`${name} must be a positive safe integer`); - } -} diff --git a/src/shared/protocol.ts b/src/shared/protocol.ts index 9269174..e3cd6b4 100644 --- a/src/shared/protocol.ts +++ b/src/shared/protocol.ts @@ -11,7 +11,6 @@ export const ROOM_CODE_LENGTH = 4; export const MAX_MEDIA_ROUTE_REVISION = Number.MAX_SAFE_INTEGER; export const MAX_SFU_TOKEN_LENGTH = 8 * 1024; export const MAX_ICE_SERVER_URLS = 8; -export const MAX_TURN_CREDENTIAL_LENGTH = 256; export const MAX_VIEWER_QUALITY_EVIDENCE_BYTES = 2 * 1024; export const MAX_PARENT_EDGE_QUALITY_EVIDENCE_BYTES = 2 * 1024; export const VIEWER_QUALITY_EVIDENCE_INTERVAL_MS = 2_000; @@ -254,46 +253,6 @@ export const stunUrlSchema = z .max(512) .refine(isValidStunUrl, { message: "Invalid STUN URL" }); -function isValidTurnUrl(value: string): boolean { - const match = /^turn:([^/?#\s]+)\?transport=udp$/i.exec(value); - if (!match) { - return false; - } - let authority: URL; - try { - authority = new URL(`http://${match[1]}`); - } catch { - return false; - } - return Boolean( - authority.hostname && - !authority.username && - !authority.password && - authority.pathname === "/" && - !authority.search && - !authority.hash && - (!authority.port || Number(authority.port) > 0) - ); -} - -export const turnUrlSchema = z - .string() - .min(1) - .max(512) - .refine(isValidTurnUrl, { message: "Invalid TURN URL" }); - -const selectedEdgeTurnIceServerSchema = z - .object({ - urls: z.array(turnUrlSchema).length(1), - username: z - .string() - .min(1) - .max(128) - .regex(/^[1-9]\d{0,12}:[A-Za-z0-9_-]{16,64}$/), - credential: z.string().min(1).max(MAX_TURN_CREDENTIAL_LENGTH), - }) - .strict(); - const iceServerSchema = z .object({ urls: z.union([ @@ -348,7 +307,7 @@ export const preparedRouteCandidateSchema = z .object({ childPeerId: opaqueIdSchema, connectionId: opaqueIdSchema, - transport: z.enum(["direct", "selected-turn", "sfu"]), + transport: z.enum(["direct", "sfu"]), }) .strict(); export type PreparedRouteCandidate = z.infer< @@ -779,34 +738,6 @@ export const serverMessageSchema = z.union([ token: z.string().min(1).max(MAX_SFU_TOKEN_LENGTH), }) .strict(), - z.discriminatedUnion("edgeKind", [ - z - .object({ - type: z.literal("selected-edge-turn"), - edgeKind: z.literal("peer-selected"), - revision: mediaRouteRevisionSchema, - parentPeerId: opaqueIdSchema, - viewerPeerId: opaqueIdSchema, - oldConnectionId: opaqueIdSchema, - newConnectionId: opaqueIdSchema, - expiresAt: z.string().datetime(), - iceServer: selectedEdgeTurnIceServerSchema, - }) - .strict(), - z - .object({ - type: z.literal("selected-edge-turn"), - edgeKind: z.literal("host-sfu-ingress"), - revision: mediaRouteRevisionSchema, - hostPeerId: opaqueIdSchema, - publicationGeneration: opaqueIdSchema, - oldConnectionId: opaqueIdSchema, - newConnectionId: opaqueIdSchema, - expiresAt: z.string().datetime(), - iceServer: selectedEdgeTurnIceServerSchema, - }) - .strict(), - ]), z .object({ type: z.literal("quality-settings"), diff --git a/tests/client-state.test.ts b/tests/client-state.test.ts index d792faf..23a4d72 100644 --- a/tests/client-state.test.ts +++ b/tests/client-state.test.ts @@ -1634,7 +1634,7 @@ describe("WebRTC stats parsing", () => { }); }); - it("separates the local TURN transport from the ICE protocol", async () => { + it("keeps an out-of-contract relay candidate unclassified", async () => { const report = new Map>([ [ "transport", @@ -1666,7 +1666,6 @@ describe("WebRTC stats parsing", () => { timestamp: 2_000, candidateType: "relay", protocol: "udp", - relayProtocol: "tls", }, ], [ @@ -1724,91 +1723,14 @@ describe("WebRTC stats parsing", () => { expect(firstMetrics.intervalDecodeMs).toBeNull(); expect(metrics).toMatchObject({ - path: "relay", + path: "unknown", iceProtocol: "udp", - localRelayProtocol: "tls", localCandidateType: "relay", remoteCandidateType: "host", intervalDecodeMs: 20, }); }); - it("does not infer a remote TURN transport", async () => { - const report = new Map([ - [ - "transport", - { - id: "transport", - type: "transport", - timestamp: 1_000, - selectedCandidatePairId: "pair", - }, - ], - [ - "pair", - { - id: "pair", - type: "candidate-pair", - timestamp: 1_000, - transportId: "transport", - localCandidateId: "local", - remoteCandidateId: "remote", - state: "succeeded", - nominated: true, - }, - ], - [ - "local", - { - id: "local", - type: "local-candidate", - timestamp: 1_000, - candidateType: "host", - protocol: "udp", - }, - ], - [ - "remote", - { - id: "remote", - type: "remote-candidate", - timestamp: 1_000, - candidateType: "relay", - protocol: "udp", - }, - ], - [ - "inbound", - { - id: "inbound", - type: "inbound-rtp", - timestamp: 1_000, - kind: "video", - transportId: "transport", - bytesReceived: 1_000, - framesDecoded: 30, - }, - ], - ]) as unknown as RTCStatsReport; - const connection = { - getStats: async () => report, - } as unknown as RTCPeerConnection; - - const metrics = await collectConnectionMetrics( - connection, - "receive", - createStatsAccumulator(), - ); - - expect(metrics).toMatchObject({ - path: "relay", - iceProtocol: "udp", - localRelayProtocol: null, - localCandidateType: "host", - remoteCandidateType: "relay", - }); - }); - it("rebases interval metrics when the RTP stream or its counters reset", async () => { const inbound = ( id: string, diff --git a/tests/diagnostic-export.test.ts b/tests/diagnostic-export.test.ts index 76c0735..589f730 100644 --- a/tests/diagnostic-export.test.ts +++ b/tests/diagnostic-export.test.ts @@ -75,7 +75,7 @@ describe("diagnostic export privacy boundary", () => { trackIdentifier: "sensitive-track-id", rawCandidate: "candidate-sensitive", sdp: "sdp-sensitive", - turnUrl: "turn:secret.example", + privateUrl: "https://secret.example", stunUrl: "stun:secret.example", username: "username-sensitive", credential: "credential-sensitive", diff --git a/tests/host-peer-tracks.test.ts b/tests/host-peer-tracks.test.ts index 01aca65..4871093 100644 --- a/tests/host-peer-tracks.test.ts +++ b/tests/host-peer-tracks.test.ts @@ -305,30 +305,6 @@ function createStream( } as unknown as MediaStream; } -const selectedTurnIceServer = { - urls: ["turn:turn.example.test:3478?transport=udp"], - username: "1787230000:opaque_identity_12345678", - credential: "short-lived-credential", -}; - -function selectedEdgeTurn( - viewerPeerId: string, - oldConnectionId: string, - newConnectionId: string, -) { - return { - type: "selected-edge-turn" as const, - edgeKind: "peer-selected" as const, - revision: 7, - parentPeerId: "selected-parent", - viewerPeerId, - oldConnectionId, - newConnectionId, - expiresAt: "2099-08-20T12:00:00.000Z", - iceServer: selectedTurnIceServer, - }; -} - function createPeer( stream: MediaStream, onUpdate: (snapshot: PeerSnapshot) => void = () => undefined, @@ -1149,9 +1125,11 @@ describe("ViewerRelay downstream ownership", () => { const routeCandidate = ( revision: number, childPeerId: string, - transport: "direct" | "selected-turn" = "direct", - connectionId = `relay-candidate-${revision}`, - ) => ({ childPeerId, connectionId, transport }); + ) => ({ + childPeerId, + connectionId: `relay-candidate-${revision}`, + transport: "direct" as const, + }); it("promotes the exact prepared child connection within the current Viewer cap", async () => { const signals: Array<{ peerId: string; connectionId: string }> = []; @@ -1317,79 +1295,6 @@ describe("ViewerRelay downstream ownership", () => { relay.dispose(); }); - it("replaces the exact prepared child with selected TURN in the same slot", async () => { - const signals: Array<{ peerId: string; connectionId: string }> = []; - const relay = new ViewerRelay( - { iceServers: [] }, - QUALITY_PROFILES["720p30"], - { - sendSignal: (peerId, payload) => { - signals.push({ peerId, connectionId: payload.connectionId }); - return true; - }, - }, - ); - relay.setChildren(["selected-child"]); - relay.setStream(createStream(createTrack("video", "selected-video"), null)); - await vi.waitFor(() => expect(signals).toHaveLength(1)); - const activeConnection = FakePeerConnection.latest!; - const activeConnectionId = signals[0]!.connectionId; - activeConnection.connectionState = "connected"; - expect( - relay.prepareChild( - 7, - routeCandidate(7, "selected-child", "selected-turn", "selected-connection"), - ["selected-child"], - ), - ).toBe(true); - expect(FakePeerConnection.latest).toBe(activeConnection); - - expect( - relay.prepareSelectedEdgeTurn( - selectedEdgeTurn( - "selected-child", - activeConnectionId, - "selected-connection", - ), - "selected-parent", - 7, - ), - ).toBe(true); - await vi.waitFor(() => expect(signals).toHaveLength(2)); - const selectedConnection = FakePeerConnection.latest!; - expect(activeConnection.connectionState).toBe("connected"); - expect(selectedConnection.configurations).toEqual([ - { - iceServers: [selectedTurnIceServer], - iceTransportPolicy: "relay", - }, - ]); - expect(signals[1]).toEqual({ - peerId: "selected-child", - connectionId: "selected-connection", - }); - - await expect( - relay.acceptSignal( - "selected-child", - { - kind: "description", - connectionId: "selected-connection", - description: { type: "answer", sdp: "selected-answer" }, - }, - 7, - ), - ).resolves.toBe(true); - selectedConnection.connectionState = "connected"; - relay.activateChildren(7, ["selected-child"]); - expect(activeConnection.connectionState).toBe("closed"); - expect(relay.getSnapshot("selected-child")?.connectionId).toBe( - "selected-connection", - ); - expect(FakePeerConnection.activeCount).toBe(1); - relay.dispose(); - }); - it("exposes a defensive snapshot of current downstream send metrics", async () => { const relay = new ViewerRelay( { iceServers: [] }, diff --git a/tests/hybrid-media-router.test.ts b/tests/hybrid-media-router.test.ts index a61dc2a..b99f252 100644 --- a/tests/hybrid-media-router.test.ts +++ b/tests/hybrid-media-router.test.ts @@ -8,7 +8,6 @@ import { } from "../src/server/hybrid-media-router.ts"; import { RoomStore, type CreatedRoom } from "../src/server/room-store.ts"; import { SfuResourceAdmission } from "../src/server/sfu-resource-admission.ts"; -import { TurnAllocationAdmission } from "../src/server/turn-allocation-admission.ts"; import { FakeSfuRoomControl } from "./fake-sfu-room-control.ts"; const SHARE_GENERATION = "share_generation_12345678"; @@ -96,7 +95,6 @@ function activeAfter( function harness( endpointMediaCopyCapacity: 1 | 2 | 3, withSfu = false, - withSelectedTurn = false, prepareTimeoutMs?: number, ) { const store = createStore(); @@ -106,9 +104,6 @@ function harness( ? new SfuResourceAdmission({ ingressCapacity: 2, egressCapacity: 20 }) : undefined; const roomControl = withSfu ? new FakeSfuRoomControl() : undefined; - const turnAdmission = withSelectedTurn - ? new TurnAllocationAdmission({ capacity: 20 }) - : undefined; const router = new HybridMediaRouter({ roomStore: store, endpointMediaCopyCapacity, @@ -127,19 +122,6 @@ function harness( }, } : {}), - ...(turnAdmission - ? { - selectedEdgeTurn: { - config: { - urls: ["turn:turn.example.test:3478?transport=udp"] as const, - sharedSecret: "t".repeat(32), - credentialTtlSeconds: 120, - allocationCapacity: 20, - }, - admission: turnAdmission, - }, - } - : {}), sendToSession(sessionId, message) { const messages = sent.get(sessionId) ?? []; messages.push(message); @@ -156,14 +138,14 @@ function harness( }, getShareGeneration: () => SHARE_GENERATION, }); - return { store, sent, admission, turnAdmission, router }; + return { store, sent, admission, router }; } describe("HybridMediaRouter v7 runtime", () => { it("wakes at the derived direct boundary and prepares SFU automatically", async () => { vi.useFakeTimers(); vi.setSystemTime(0); - const { store, sent, router } = harness(2, true, false, 300); + const { store, sent, router } = harness(2, true, 300); try { const room = await store.createRoom(); const host = connectHost(store, room); @@ -197,6 +179,73 @@ describe("HybridMediaRouter v7 runtime", () => { } }); + it("uses the direct boundary to bootstrap SFU when every Host slot is full", async () => { + vi.useFakeTimers(); + vi.setSystemTime(0); + const { store, sent, router } = harness(2, true, 300); + try { + const room = await store.createRoom(); + const host = connectHost(store, room); + complete(router, host); + + const firstRoot = connectViewer(store, room, "first-root"); + complete(router, firstRoot); + await vi.waitFor(() => + expect(preparedFor(sent, firstRoot.sessionId)?.candidate.transport).toBe( + "direct", + ), + ); + const firstPrepare = preparedFor(sent, firstRoot.sessionId)!; + router.handleRouteReady(firstRoot, { + type: "route-ready", + revision: firstPrepare.revision, + phase: "prepare", + }); + router.setViewerRelayCapacity(firstRoot, 2); + + const secondRoot = connectViewer(store, room, "second-root"); + complete(router, secondRoot); + await vi.waitFor(() => + expect(preparedFor(sent, secondRoot.sessionId)?.candidate.transport).toBe( + "direct", + ), + ); + const secondPrepare = preparedFor(sent, secondRoot.sessionId)!; + router.handleRouteReady(secondRoot, { + type: "route-ready", + revision: secondPrepare.revision, + phase: "prepare", + }); + + const waiting = connectViewer(store, room, "waiting"); + complete(router, waiting); + await vi.waitFor(() => + expect(preparedFor(sent, waiting.sessionId)?.candidate.transport).toBe( + "direct", + ), + ); + + await vi.advanceTimersByTimeAsync(150); + await vi.waitFor(() => + expect(preparedFor(sent, secondRoot.sessionId)?.candidate.transport).toBe( + "sfu", + ), + ); + expect( + sent + .get(waiting.sessionId) + ?.some( + (message) => + message.type === "error" && + message.message === "No usable media route is available", + ), + ).toBe(false); + } finally { + await router.close(); + vi.useRealTimers(); + } + }); + it("orders exact prepare, commits only child proof, and rolls back above P", async () => { const { store, sent, router } = harness(2); const room = await store.createRoom(); @@ -309,124 +358,4 @@ describe("HybridMediaRouter v7 runtime", () => { await router.close(); }); - it("uses selected TURN only as the exact Viewer-parent edge transport", async () => { - const { store, sent, turnAdmission, router } = harness(1, true, true); - const room = await store.createRoom(); - const host = connectHost(store, room); - complete(router, host); - const parent = connectViewer(store, room, "selected_parent"); - complete(router, parent); - await vi.waitFor(() => expect(preparedFor(sent, parent.sessionId)).toBeDefined()); - const parentPrepare = preparedFor(sent, parent.sessionId)!; - router.handleRouteReady(parent, { - type: "route-ready", - revision: parentPrepare.revision, - phase: "prepare", - }); - router.setViewerRelayCapacity(parent, 1); - - const child = connectViewer(store, room, "selected_child"); - complete(router, child); - await vi.waitFor(() => - expect(preparedFor(sent, child.sessionId)?.candidate.transport).toBe("direct"), - ); - const direct = preparedFor(sent, child.sessionId)!; - expect(direct.assignment.upstream).toEqual({ - kind: "peer", - peerId: parent.peerId, - }); - router.handleRouteReady(child, { - type: "route-ready", - revision: direct.revision, - phase: "prepare", - }); - - router.handleRouteFailed(child, { - type: "route-failed", - revision: direct.revision, - phase: "active", - connectionId: direct.candidate.connectionId, - }); - await vi.waitFor(() => - expect(preparedFor(sent, child.sessionId)?.candidate.transport).toBe( - "selected-turn", - ), - ); - const selected = preparedFor(sent, child.sessionId)!; - expect( - sent - .get(child.sessionId) - ?.some( - (message) => - message.type === "selected-edge-turn" && - message.edgeKind === "peer-selected" && - message.newConnectionId === selected.candidate.connectionId, - ), - ).toBe(true); - router.handleRouteReady(child, { - type: "route-ready", - revision: selected.revision, - phase: "prepare", - }); - await vi.waitFor(() => - expect(turnAdmission?.usage()).toEqual({ allocations: 1 }), - ); - await router.close(); - expect(turnAdmission?.usage()).toEqual({ allocations: 0 }); - }); - - it("uses selected TURN as one Host-SFU ingress after direct ingress fails", async () => { - const { store, sent, turnAdmission, router } = harness(1, true, true); - const room = await store.createRoom(); - const host = connectHost(store, room); - complete(router, host); - const first = connectViewer(store, room, "ingress_first"); - complete(router, first); - await vi.waitFor(() => expect(preparedFor(sent, first.sessionId)).toBeDefined()); - const direct = preparedFor(sent, first.sessionId)!; - router.handleRouteReady(first, { - type: "route-ready", - revision: direct.revision, - phase: "prepare", - }); - - const waiting = connectViewer(store, room, "ingress_waiting"); - complete(router, waiting); - await vi.waitFor(() => - expect(preparedFor(sent, first.sessionId)?.candidate.transport).toBe("sfu"), - ); - const directIngress = preparedFor(sent, first.sessionId)!; - router.handleRouteFailed(first, { - type: "route-failed", - revision: directIngress.revision, - phase: "prepare", - connectionId: directIngress.candidate.connectionId, - }); - - await vi.waitFor(() => { - const prepared = preparedFor(sent, first.sessionId); - expect(prepared?.revision).toBeGreaterThan(directIngress.revision); - return prepared; - }); - const selectedIngress = preparedFor(sent, first.sessionId)!; - expect( - sent - .get(host.sessionId) - ?.some( - (message) => - message.type === "selected-edge-turn" && - message.edgeKind === "host-sfu-ingress" && - message.revision === selectedIngress.revision, - ), - ).toBe(true); - router.handleRouteReady(first, { - type: "route-ready", - revision: selectedIngress.revision, - phase: "prepare", - }); - await vi.waitFor(() => - expect(turnAdmission?.usage()).toEqual({ allocations: 1 }), - ); - await router.close(); - }); }); diff --git a/tests/protocol.test.ts b/tests/protocol.test.ts index 764d7c3..21ae805 100644 --- a/tests/protocol.test.ts +++ b/tests/protocol.test.ts @@ -792,7 +792,7 @@ describe("client signaling protocol", () => { }); describe("server signaling protocol", () => { - it("rejects the removed all-room TURN credential wire", () => { + it("accepts only STUN URLs in the authenticated peer ICE config", () => { expect( serverMessageSchema.safeParse({ type: "ice-config", @@ -1156,54 +1156,6 @@ describe("server signaling protocol", () => { ).toBe(false); }); - it("accepts only explicitly classified selected-edge grants", () => { - const iceServer = { - urls: ["turn:relay.example.test:3478?transport=udp"], - username: `1787076000:${"a".repeat(32)}`, - credential: "credential", - }; - expect( - serverMessageSchema.safeParse({ - type: "selected-edge-turn", - edgeKind: "peer-selected", - revision: 1, - parentPeerId: "parent_12345678", - viewerPeerId: "viewer_12345678", - oldConnectionId: "old_12345678", - newConnectionId: "new_12345678", - expiresAt: "2026-08-20T12:00:00.000Z", - iceServer, - }).success, - ).toBe(true); - expect( - serverMessageSchema.safeParse({ - type: "selected-edge-turn", - edgeKind: "host-sfu-ingress", - revision: 1, - hostPeerId: "host_12345678", - publicationGeneration: "publication_12345678", - oldConnectionId: "publication_12345678", - newConnectionId: "new_12345678", - expiresAt: "2026-08-20T12:00:00.000Z", - iceServer, - }).success, - ).toBe(true); - expect( - serverMessageSchema.safeParse({ - type: "selected-edge-turn", - edgeKind: "host-sfu-ingress", - revision: 1, - hostPeerId: "host_12345678", - publicationGeneration: "publication_12345678", - oldConnectionId: "publication_12345678", - newConnectionId: "new_12345678", - parentPeerId: "unexpected_12345678", - expiresAt: "2026-08-20T12:00:00.000Z", - iceServer, - }).success, - ).toBe(false); - }); - it("accepts every bounded hybrid upstream shape", () => { for (const upstream of [ { kind: "none" }, diff --git a/tests/room-route-controller.test.ts b/tests/room-route-controller.test.ts index 6ea7500..6ec0a33 100644 --- a/tests/room-route-controller.test.ts +++ b/tests/room-route-controller.test.ts @@ -19,7 +19,7 @@ const F = "viewer_f_12345678"; function controller( capacity: 1 | 2 | 3 = 2, - options: { selectedTurnEnabled?: boolean; sfuEnabled?: boolean } = {}, + options: { sfuEnabled?: boolean } = {}, ) { const routes = new RoomRouteController({ hostPeerId: HOST, @@ -149,10 +149,7 @@ describe("RoomRouteController", () => { ); it("keeps healthy branches while 18 silent Viewers fall back serially", () => { - const routes = controller(2, { - selectedTurnEnabled: true, - sfuEnabled: true, - }); + const routes = controller(2, { sfuEnabled: true }); addViewer(routes, B, 2); routes.hydrateHostPublication( "publication_generation", @@ -186,33 +183,13 @@ describe("RoomRouteController", () => { }); routes.operationExpired(directOperation.wakeAtMs); - const selectedOperation = routes.snapshot().operation!; - expect( - selectedOperation.candidates[selectedOperation.cursor]!.tuple, - ).toMatchObject({ kind: "peer", transport: "selected-turn" }); - beginCandidate(routes, { - nowMs: selectedOperation.wakeAtMs - 1, - connectionId: `${peerId}_silent_turn`, - reservation: { - kind: "selected-turn", - edge: `${peerId}_turn_resource`, - }, - }); - const selectedExpired = routes.operationExpired( - selectedOperation.wakeAtMs, - ); - expect(selectedExpired.released).toEqual([ - `${peerId}_turn_resource`, - ]); - const sfuOperation = routes.snapshot().operation!; expect(sfuOperation.candidates[sfuOperation.cursor]!.tuple).toEqual({ kind: "sfu", publication: "reuse", - ingress: "existing", }); const prepared = beginCandidate(routes, { - nowMs: selectedOperation.wakeAtMs + 1, + nowMs: directOperation.wakeAtMs + 1, connectionId: `${peerId}_sfu`, reservation: { kind: "sfu-reuse", @@ -227,10 +204,10 @@ describe("RoomRouteController", () => { revision: prepared.current!.revision, connectionId: `${peerId}_sfu`, }, - selectedOperation.wakeAtMs + 2, + directOperation.wakeAtMs + 2, ).accepted, ).toBe(true); - nowMs = selectedOperation.wakeAtMs + 3; + nowMs = directOperation.wakeAtMs + 3; expect(routes.snapshot().upstreamByViewer.get(B)).toMatchObject({ kind: "sfu", @@ -333,10 +310,7 @@ describe("RoomRouteController", () => { }); it("advances one candidate cursor without resetting the operation deadline", () => { - const routes = controller(2, { - selectedTurnEnabled: true, - sfuEnabled: true, - }); + const routes = controller(2, { sfuEnabled: true }); addViewer(routes, A, 2); const operation = routes.reconcile(1_000).operation!; const deadline = operation.deadlineAtMs; @@ -394,69 +368,21 @@ describe("RoomRouteController", () => { expect(routes.reconcile(deadline + 1).operation).toBeUndefined(); routes.touchExternalFacts(); expect(routes.reconcile(deadline + 2).operation?.childPeerId).toBe(A); - - const overlap = controller(2, { selectedTurnEnabled: true }); - addViewer(overlap, A, 1); - addViewer(overlap, B, 0); - overlap.hydrateEdge(A, peerEdge(HOST, "a_parent")); - overlap.hydrateEdge(B, peerEdge(A, "b_direct")); - expect(overlap.invalidateEdge({ - childPeerId: B, - childSessionId: `${B}_session`, - parentSessionId: `${A}_session`, - routeRevision: 0, - connectionId: "b_direct", - })).toBe(true); - overlap.reconcile(0); - const hostAttempt = beginCandidate(overlap, { - nowMs: 1, - connectionId: "b_host", - reservation: { kind: "direct" }, - }).operation!; - overlap.candidateFailed({ - childPeerId: B, - childSessionId: `${B}_session`, - revision: hostAttempt.current!.revision, - connectionId: "b_host", - }, 2); - expect(() => beginCandidate(overlap, { - nowMs: 3, - connectionId: "b_turn", - reservation: { kind: "selected-turn", edge: "turn_edge" }, - })).toThrow("endpoint overlap reservation"); - beginCandidate(overlap, { - nowMs: 3, - connectionId: "b_turn", - reservation: { - kind: "selected-turn", - edge: "turn_edge", - overlap: "overlap_slot", - }, - }); - overlap.setEffectiveCapacity(A, `${A}_session`, 0); - expect(overlap.reconcile(4).released).toEqual([ - "turn_edge", - "overlap_slot", - ]); }); - it("reserves one derived deadline stage for direct, selected TURN, and SFU", () => { - const routes = controller(2, { - selectedTurnEnabled: true, - sfuEnabled: true, - }); + it("reserves one derived deadline stage for direct and SFU", () => { + const routes = controller(2, { sfuEnabled: true }); addViewer(routes, B, 2); routes.hydrateEdge(B, peerEdge(HOST, "b_from_host")); addViewer(routes, A, 0); const operation = routes.reconcile(0).operation!; expect(operation.deadlineAtMs).toBe(10_000); - expect(operation.wakeAtMs).toBe(Math.floor(10_000 / 3)); + expect(operation.wakeAtMs).toBe(5_000); expect(operation.candidates.map((candidate) => candidate.tuple)).toEqual( expect.arrayContaining([ { kind: "peer", parentPeerId: B, transport: "direct" }, - { kind: "peer", parentPeerId: B, transport: "selected-turn" }, - { kind: "sfu", publication: "create", ingress: "direct" }, + { kind: "sfu", publication: "create" }, ]), ); @@ -470,31 +396,14 @@ describe("RoomRouteController", () => { expect(directExpired.exhausted).toBeUndefined(); expect(routes.snapshot().operation).toMatchObject({ deadlineAtMs: 10_000, - wakeAtMs: Math.floor((10_000 * 2) / 3), + wakeAtMs: 10_000, current: undefined, }); - expect( - routes.snapshot().operation!.candidates[ - routes.snapshot().operation!.cursor - ]!.tuple, - ).toMatchObject({ kind: "peer", transport: "selected-turn" }); - - beginCandidate(routes, { - nowMs: operation.wakeAtMs + 1, - connectionId: "silent_selected_turn", - reservation: { kind: "selected-turn", edge: "turn_edge" }, - }); - const selectedExpired = routes.operationExpired( - Math.floor((10_000 * 2) / 3), - ); - expect(selectedExpired).toMatchObject({ accepted: true }); - expect(selectedExpired.released).toEqual(["turn_edge"]); expect( routes.snapshot().operation!.candidates[ routes.snapshot().operation!.cursor ]!.tuple, ).toMatchObject({ kind: "sfu" }); - expect(routes.snapshot().operation!.wakeAtMs).toBe(10_000); const prepared = beginCandidate(routes, { nowMs: 7_000, @@ -524,11 +433,8 @@ describe("RoomRouteController", () => { }); }); - it("reaches a healthy SFU reuse stage after silent peer transports", () => { - const routes = controller(2, { - selectedTurnEnabled: true, - sfuEnabled: true, - }); + it("reaches a healthy SFU reuse stage after silent peer transport", () => { + const routes = controller(2, { sfuEnabled: true }); addViewer(routes, B, 2); routes.hydrateHostPublication( "publication_generation", @@ -551,8 +457,7 @@ describe("RoomRouteController", () => { expect(operation.candidates.map((candidate) => candidate.tuple)).toEqual( expect.arrayContaining([ { kind: "peer", parentPeerId: B, transport: "direct" }, - { kind: "peer", parentPeerId: B, transport: "selected-turn" }, - { kind: "sfu", publication: "reuse", ingress: "existing" }, + { kind: "sfu", publication: "reuse" }, ]), ); @@ -562,12 +467,6 @@ describe("RoomRouteController", () => { reservation: { kind: "direct" }, }); routes.operationExpired(operation.wakeAtMs); - beginCandidate(routes, { - nowMs: operation.wakeAtMs + 1, - connectionId: "silent_selected", - reservation: { kind: "selected-turn", edge: "turn_edge" }, - }); - routes.operationExpired(Math.floor((10_000 * 2) / 3)); expect( routes.snapshot().operation!.candidates[ @@ -576,7 +475,6 @@ describe("RoomRouteController", () => { ).toEqual({ kind: "sfu", publication: "reuse", - ingress: "existing", }); const prepared = beginCandidate(routes, { nowMs: 7_000, @@ -605,6 +503,37 @@ describe("RoomRouteController", () => { }); }); + it("reserves a stage for Host-full SFU bootstrap before it is a tuple", () => { + const routes = controller(2, { sfuEnabled: true }); + addViewer(routes, B, 2); + routes.hydrateEdge(B, peerEdge(HOST, "b_from_host")); + addViewer(routes, C, 0); + routes.hydrateEdge(C, peerEdge(HOST, "c_from_host")); + addViewer(routes, A, 0); + + const operation = routes.reconcile(0).operation!; + expect(operation.candidates.some((candidate) => candidate.tuple.kind === "sfu")) + .toBe(false); + expect(operation.wakeAtMs).toBe(5_000); + beginCandidate(routes, { + nowMs: 1, + connectionId: "a_silent_direct", + reservation: { kind: "direct" }, + }); + const blocked = routes.operationExpired(operation.wakeAtMs); + expect(blocked).toMatchObject({ accepted: true }); + expect(blocked.exhausted).toBe(false); + expect(blocked.released).toEqual([]); + expect(routes.snapshot().operation).toBeUndefined(); + + const bootstrap = routes.reconcile(operation.wakeAtMs + 1).operation!; + expect(bootstrap.childPeerId).toBe(C); + expect(bootstrap.candidates[0]!.tuple).toMatchObject({ + kind: "sfu", + publication: "create", + }); + }); + it("retains another exact failed edge while one child operation is busy", () => { const routes = controller(3); addViewer(routes, A, 1); @@ -662,6 +591,53 @@ describe("RoomRouteController", () => { expect(routes.reconcile(3).operation?.childPeerId).toBe(B); }); + it("uses a freed Host slot for new demand without moving healthy descendants", () => { + const routes = controller(2); + addViewer(routes, A, 2); + addViewer(routes, B, 2); + addViewer(routes, C, 0); + addViewer(routes, D, 0); + routes.hydrateEdge(A, peerEdge(HOST, "a_root")); + routes.hydrateEdge(B, peerEdge(HOST, "b_root")); + routes.hydrateEdge(C, peerEdge(A, "c_from_a")); + routes.hydrateEdge(D, peerEdge(B, "d_from_b")); + + expect(routes.confirmDeparture(B)).toBe(true); + const repair = routes.reconcile(0).operation!; + expect(repair.childPeerId).toBe(D); + expect(repair.candidates[0]!.tuple).toEqual({ + kind: "peer", + parentPeerId: A, + transport: "direct", + }); + commitCurrent(routes, 1, "d_from_a"); + expect(routes.reconcile(2).operation).toBeUndefined(); + expect(routes.snapshot().upstreamByViewer.has(B)).toBe(false); + + addViewer(routes, E, 0); + const join = routes.reconcile(3).operation!; + expect(join.childPeerId).toBe(E); + expect(join.candidates[0]!.tuple).toEqual({ + kind: "peer", + parentPeerId: HOST, + transport: "direct", + }); + commitCurrent(routes, 4, "e_from_host"); + + expect(routes.snapshot().upstreamByViewer.get(C)).toMatchObject({ + parentPeerId: A, + connectionId: "c_from_a", + }); + expect(routes.snapshot().upstreamByViewer.get(D)).toMatchObject({ + parentPeerId: A, + connectionId: "d_from_a", + }); + expect(routes.snapshot().upstreamByViewer.get(E)).toMatchObject({ + parentPeerId: HOST, + connectionId: "e_from_host", + }); + }); + it("keeps grace media, then reparents a confirmed relay departure", () => { const routes = controller(2); addViewer(routes, A, 2); @@ -757,7 +733,6 @@ describe("RoomRouteController", () => { expect(routes.snapshot().operation?.candidates[1]?.tuple).toEqual({ kind: "sfu", publication: "replace", - ingress: "direct", }); expect(() => beginCandidate(routes, { nowMs: 3, @@ -810,7 +785,6 @@ describe("RoomRouteController", () => { expect(bootstrapOperation.candidates[0]?.tuple).toEqual({ kind: "sfu", publication: "create", - ingress: "direct", }); expect(() => beginCandidate(bootstrap, { nowMs: 2, @@ -905,7 +879,7 @@ describe("RoomRouteController", () => { routes.touchExternalFacts(); const operation = routes.reconcile(0).operation!; expect(operation.candidates[0]).toMatchObject({ - tuple: { kind: "sfu", publication: "replace", ingress: "direct" }, + tuple: { kind: "sfu", publication: "replace" }, endpointTransition: { kind: "overlap", producerPeerId: HOST }, }); const prepared = beginCandidate(routes, { @@ -941,30 +915,25 @@ describe("RoomRouteController", () => { expect(routes.reconcile(3).operation?.childPeerId).toBe(B); }); - it("releases session-bound transports while rebinding direct media", () => { + it("rebinds direct and SFU media to replacement sessions", () => { const peerRoutes = controller(2); addViewer(peerRoutes, A, 1); addViewer(peerRoutes, B, 0); - peerRoutes.hydrateEdge(A, { - kind: "peer", - parentPeerId: HOST, - transport: "selected-turn", - connectionId: "a_selected", - usable: true, - physicalActive: true, - resource: "a_turn_allocation", - }); - peerRoutes.hydrateEdge(B, peerEdge(HOST, "b_direct")); + peerRoutes.hydrateEdge(A, peerEdge(HOST, "a_direct")); + peerRoutes.hydrateEdge(B, peerEdge(A, "b_direct")); expect(peerRoutes.upsertParticipant({ peerId: A, role: "viewer", sessionId: "a_replacement", effectiveDownstreamCapacity: 1, - })).toEqual(["a_turn_allocation"]); + })).toEqual([]); expect(peerRoutes.snapshot().upstreamByViewer.get(A)).toMatchObject({ - childSessionId: `${A}_session`, - usable: false, - physicalActive: false, + childSessionId: "a_replacement", + usable: true, + physicalActive: true, + }); + expect(peerRoutes.snapshot().upstreamByViewer.get(B)).toMatchObject({ + parentSessionId: "a_replacement", }); expect(peerRoutes.upsertParticipant({ peerId: B, @@ -981,14 +950,13 @@ describe("RoomRouteController", () => { const sfuRoutes = controller(1, { sfuEnabled: true }); addViewer(sfuRoutes, A, 0); sfuRoutes.hydrateHostPublication( - "selected_publication", - "selected_ingress_resource", - "selected_ingress_connection", - "selected-turn", + "publication", + "publication_resource", + "publication_connection", ); sfuRoutes.hydrateEdge(A, { kind: "sfu", - publicationGeneration: "selected_publication", + publicationGeneration: "publication", transport: "sfu", connectionId: "subscription_connection", usable: true, @@ -1000,12 +968,11 @@ describe("RoomRouteController", () => { role: "host", sessionId: "host_replacement", effectiveDownstreamCapacity: 1, - })).toEqual(["selected_ingress_resource"]); + })).toEqual([]); expect(sfuRoutes.snapshot().hostPublication).toMatchObject({ - hostSessionId: "host_session", - ingress: "selected-turn", - usable: false, - physicalActive: false, + hostSessionId: "host_replacement", + usable: true, + physicalActive: true, }); expect(sfuRoutes.upsertParticipant({ peerId: A, @@ -1157,57 +1124,6 @@ describe("RoomRouteController", () => { expect(routes.snapshot().hostPublication).toBeNull(); }); - it("retires one exact peer edge for a C=3 bounded gap", () => { - const routes = controller(3, { selectedTurnEnabled: true }); - addViewer(routes, A, 3); - routes.hydrateEdge(A, peerEdge(HOST, "a_direct")); - for (const peerId of [B, C, D]) { - addViewer(routes, peerId, 0); - routes.hydrateEdge(peerId, peerEdge(A, `${peerId}_from_a`)); - } - routes.invalidateEdge({ - childPeerId: B, - childSessionId: `${B}_session`, - parentSessionId: `${A}_session`, - routeRevision: 0, - connectionId: `${B}_from_a`, - }); - const first = routes.reconcile(0).operation!; - expect(first.candidates[0]?.tuple).toMatchObject({ - kind: "peer", - parentPeerId: HOST, - transport: "direct", - }); - expect(skipCandidate(routes, 1).accepted).toBe(true); - const bounded = routes.snapshot().operation!; - expect(bounded.candidates[bounded.cursor]).toMatchObject({ - tuple: { kind: "peer", parentPeerId: A, transport: "selected-turn" }, - endpointTransition: { - kind: "bounded-gap", - producerPeerId: A, - retire: { kind: "edge", childPeerId: B, connectionId: `${B}_from_a` }, - }, - }); - const gap = routes.retireCurrentCandidateProducer(cursorGuard(bounded), 2); - expect(gap).toMatchObject({ accepted: true, released: [] }); - const prepared = beginCandidate(routes, { - nowMs: 3, - connectionId: "b_selected_turn", - reservation: { kind: "selected-turn", edge: "b_turn_resource" }, - }).operation!; - expect(routes.candidateReady({ - childPeerId: B, - childSessionId: `${B}_session`, - revision: prepared.current!.revision, - connectionId: "b_selected_turn", - }, 4).accepted).toBe(true); - expect(routes.snapshot().upstreamByViewer.get(B)).toMatchObject({ - parentPeerId: A, - transport: "selected-turn", - resource: "b_turn_resource", - }); - }); - it("retries an exact failed tuple only after a new external fact", () => { const routes = controller(1); addViewer(routes, A, 0); diff --git a/tests/route-transition.test.ts b/tests/route-transition.test.ts index 5e2a051..2b57d05 100644 --- a/tests/route-transition.test.ts +++ b/tests/route-transition.test.ts @@ -78,7 +78,7 @@ type SubscriberEvents = Parameters< const candidate = ( revision: number, childPeerId = "viewer_12345678", - transport: "direct" | "selected-turn" | "sfu" = "direct", + transport: "direct" | "sfu" = "direct", connectionId = `candidate_${revision}_12345678`, ) => ({ childPeerId, connectionId, transport }); @@ -290,67 +290,6 @@ describe("minimal route transition contracts", () => { expect(publishers[1]?.disconnect).not.toHaveBeenCalled(); }); - it("replaces one pending Host publication with selected TURN transport", async () => { - const publishers: ReturnType[] = []; - const route = new HostSfuRoute({ - getStream: () => ({}) as MediaStream, - getProfile: () => QUALITY_PROFILES["720p30"], - reconcileChildren: () => undefined, - send: () => true, - createPublisher: () => { - const publisher = createFakePublisher([], `publisher-${publishers.length + 1}`); - publishers.push(publisher); - return publisher; - }, - }); - - route.accept({ - revision: 2, - phase: "prepare", - assignment: hostAssignment("publication-b"), - candidate: candidate( - 2, - "viewer_12345678", - "sfu", - "selected-connection", - ), - }); - expect( - route.startSelectedEdgeTurn({ - type: "selected-edge-turn", - edgeKind: "host-sfu-ingress", - revision: 2, - hostPeerId: "host_12345678", - publicationGeneration: "publication-b", - oldConnectionId: "publication-b", - newConnectionId: "selected-connection", - expiresAt: "2099-01-01T00:00:00.000Z", - iceServer: { - urls: ["turn:turn.example.test:3478?transport=udp"], - username: "1787230000:opaque_identity_12345678", - credential: "short-lived-credential", - }, - }), - ).toBe(true); - await route.acceptConfig(sfuConfig(2)); - - await vi.waitFor(() => expect(publishers).toHaveLength(1)); - expect(publishers[0]?.connect).toHaveBeenCalledWith({ - url: "wss://sfu.example.test", - token: "token-2", - rtcConfig: { - iceServers: [ - { - urls: ["turn:turn.example.test:3478?transport=udp"], - username: "1787230000:opaque_identity_12345678", - credential: "short-lived-credential", - }, - ], - iceTransportPolicy: "relay", - }, - }); - }); - it("uses the subscriber first decoded frame as the only SFU prepare ready", async () => { const log: string[] = []; const messages: ClientMessage[] = []; diff --git a/tests/selected-edge-turn.test.ts b/tests/selected-edge-turn.test.ts deleted file mode 100644 index b67b8e4..0000000 --- a/tests/selected-edge-turn.test.ts +++ /dev/null @@ -1,83 +0,0 @@ -import { createHmac } from "node:crypto"; -import { describe, expect, it } from "vitest"; -import { issueSelectedEdgeTurnCredential } from "../src/server/selected-edge-turn.ts"; -const config = { - urls: ["turn:turn.example.test:3478?transport=udp"] as [string], - sharedSecret: "t".repeat(32), - credentialTtlSeconds: 120, - allocationCapacity: 4, -}; -const identity = { - edgeKind: "peer-selected" as const, - roomId: "123456789012", - shareGeneration: "share_generation_12345678", - revision: 7, - parentPeerId: "parent_peer_12345678", - parentSessionId: "parent_session_12345678", - viewerPeerId: "viewer_peer_12345678", - viewerSessionId: "viewer_session_12345678", - oldConnectionId: "old_connection_12345678", - newConnectionId: "new_connection_12345678", -}; -describe("selected-edge TURN", () => { - it("issues an opaque, expiring coturn REST credential bound to the edge", () => { - const grant = issueSelectedEdgeTurnCredential(config, identity, 1_700_000_000_000); - expect(grant.expiresAt).toBe("2023-11-14T22:15:20.000Z"); - expect(grant.iceServer.username).toMatch(/^1700000120:[A-Za-z0-9_-]{32}$/); - expect(grant.iceServer.credential).toBe( - createHmac("sha1", config.sharedSecret) - .update(grant.iceServer.username) - .digest("base64"), - ); - const serialized = JSON.stringify(grant); - for (const value of Object.values(identity).filter((value) => typeof value === "string")) { - expect(serialized).not.toContain(String(value)); - } - expect( - issueSelectedEdgeTurnCredential( - config, - { ...identity, shareGeneration: "share_generation_87654321" }, - 1_700_000_000_000, - ).iceServer.username, - ).not.toBe(grant.iceServer.username); - }); - - it("binds a Host-to-SFU ingress grant to its publication generation", () => { - const grant = issueSelectedEdgeTurnCredential( - config, - { - edgeKind: "host-sfu-ingress", - roomId: "1", - shareGeneration: "share_generation_12345678", - revision: 7, - hostPeerId: "host_peer_12345678", - hostSessionId: "host_session_12345678", - publicationGeneration: "publication_generation_12345678", - oldConnectionId: "publication_generation_12345678", - newConnectionId: "new_connection_12345678", - }, - 1_700_000_000_000, - ); - - expect(grant.iceServer.username).toMatch( - /^1700000120:[A-Za-z0-9_-]{32}$/, - ); - expect( - issueSelectedEdgeTurnCredential( - config, - { - edgeKind: "host-sfu-ingress", - roomId: "1", - shareGeneration: "share_generation_12345678", - revision: 7, - hostPeerId: "host_peer_12345678", - hostSessionId: "host_session_12345678", - publicationGeneration: "publication_generation_87654321", - oldConnectionId: "publication_generation_87654321", - newConnectionId: "new_connection_12345678", - }, - 1_700_000_000_000, - ).iceServer.username, - ).not.toBe(grant.iceServer.username); - }); -}); diff --git a/tests/server-config.test.ts b/tests/server-config.test.ts index 4e5a58a..9698821 100644 --- a/tests/server-config.test.ts +++ b/tests/server-config.test.ts @@ -21,78 +21,8 @@ describe("server configuration", () => { expect(config.peerAssistedMedia).toBe(false); expect(config.endpointMediaCopyCapacity).toBe(2); expect(config.livekitFallback).toBeUndefined(); - expect(config.selectedEdgeTurn).toBeUndefined(); }); - it("enables selected-edge TURN only from its complete post-SFU tuple", () => { - const base = { - PEER_ASSISTED_MEDIA: "true", - LIVEKIT_URL: "wss://livekit.test", - LIVEKIT_API_KEY: "test-key", - LIVEKIT_API_SECRET: "s".repeat(32), - ...liveKitAdmission, - }; - const turn = { - SELECTED_EDGE_TURN_URLS: "turn:turn.test:3478?transport=udp", - SELECTED_EDGE_TURN_SHARED_SECRET: "t".repeat(32), - SELECTED_EDGE_TURN_CREDENTIAL_TTL_SECONDS: "120", - SELECTED_EDGE_TURN_ALLOCATION_CAPACITY: "4", - }; - expect(loadConfig({ ...base, ...turn }).selectedEdgeTurn).toEqual({ - urls: [turn.SELECTED_EDGE_TURN_URLS], - sharedSecret: turn.SELECTED_EDGE_TURN_SHARED_SECRET, - credentialTtlSeconds: 120, - allocationCapacity: 4, - }); - expect(loadConfig({ ...base, ...turn }).stunUrls).toEqual([]); - for (const name of Object.keys(turn)) { - const partial = { ...base, ...turn }; - delete partial[name as keyof typeof partial]; - expect(() => loadConfig(partial)).toThrow("must be configured together"); - } - }); - - it.each([ - "turn:turn.test:3478", - "turns:turn.test:5349?transport=udp", - "turn:user@turn.test:3478?transport=udp", - "turn:turn.test:3478?transport=tcp", - ])("rejects a non-canonical selected-edge TURN URL: %s", (url) => { - expect(() => - loadConfig({ - PEER_ASSISTED_MEDIA: "true", - LIVEKIT_URL: "wss://livekit.test", - LIVEKIT_API_KEY: "test-key", - LIVEKIT_API_SECRET: "s".repeat(32), - ...liveKitAdmission, - SELECTED_EDGE_TURN_URLS: url, - SELECTED_EDGE_TURN_SHARED_SECRET: "t".repeat(32), - SELECTED_EDGE_TURN_CREDENTIAL_TTL_SECONDS: "120", - SELECTED_EDGE_TURN_ALLOCATION_CAPACITY: "4", - }), - ).toThrow("one UDP TURN URL"); - }); - - it.each(["0", "-1", "1.5", "9007199254740992"])( - "rejects an invalid selected-edge TURN allocation capacity of %s", - (capacity) => { - expect(() => - loadConfig({ - PEER_ASSISTED_MEDIA: "true", - LIVEKIT_URL: "wss://livekit.test", - LIVEKIT_API_KEY: "test-key", - LIVEKIT_API_SECRET: "s".repeat(32), - ...liveKitAdmission, - SELECTED_EDGE_TURN_URLS: - "turn:turn.test:3478?transport=udp", - SELECTED_EDGE_TURN_SHARED_SECRET: "t".repeat(32), - SELECTED_EDGE_TURN_CREDENTIAL_TTL_SECONDS: "120", - SELECTED_EDGE_TURN_ALLOCATION_CAPACITY: capacity, - }), - ).toThrow("SELECTED_EDGE_TURN_ALLOCATION_CAPACITY must be a positive integer"); - }, - ); - it("enables LiveKit fallback only for a complete credential tuple", () => { const config = loadConfig({ PEER_ASSISTED_MEDIA: "true", @@ -394,7 +324,7 @@ describe("server configuration", () => { ); }); - it("requires STUN but not TURN in production", () => { + it("requires STUN in production", () => { expect(() => loadConfig({ NODE_ENV: "production", @@ -483,6 +413,10 @@ describe("server configuration", () => { "PEER_ICE_TURN_URLS", "PEER_ICE_TURN_SHARED_SECRET", "PEER_ICE_TURN_CREDENTIAL_TTL_SECONDS", + "SELECTED_EDGE_TURN_URLS", + "SELECTED_EDGE_TURN_SHARED_SECRET", + "SELECTED_EDGE_TURN_CREDENTIAL_TTL_SECONDS", + "SELECTED_EDGE_TURN_ALLOCATION_CAPACITY", ] as const)("rejects removed TURN configuration even when %s is blank", (name) => { expect(() => loadConfig({ [name]: "" })).toThrow( `${name} is no longer supported`, diff --git a/tests/sfu-client.test.ts b/tests/sfu-client.test.ts index 24bdd9f..9befce4 100644 --- a/tests/sfu-client.test.ts +++ b/tests/sfu-client.test.ts @@ -558,37 +558,6 @@ describe("SfuPublisher", () => { expect(room.disconnect).toHaveBeenCalledWith(false); }); - it("passes a controller-selected relay-only RTC configuration to LiveKit", async () => { - const iceServer = { - urls: ["turn:relay.example.test:3478?transport=udp"], - username: `1787076000:${"a".repeat(32)}`, - credential: "short-lived-credential", - }; - const publisher = new SfuPublisher(); - - await expect( - publisher.connect({ - ...connection, - rtcConfig: { - iceServers: [iceServer], - iceTransportPolicy: "relay", - }, - }), - ).resolves.toBe(true); - - expect(livekit.state.rooms[0].connect).toHaveBeenCalledWith( - connection.url, - connection.token, - { - autoSubscribe: false, - rtcConfig: { - iceServers: [iceServer], - iceTransportPolicy: "relay", - }, - }, - ); - }); - it("retains the bounded stage after connection setup fails", async () => { livekit.state.connectGate = Promise.reject(new Error("connect failed")); const publisher = new SfuPublisher(); diff --git a/tests/status-badge.test.ts b/tests/status-badge.test.ts index 069f25e..73f99f6 100644 --- a/tests/status-badge.test.ts +++ b/tests/status-badge.test.ts @@ -3,7 +3,6 @@ import { describe, expect, it } from "vitest"; import { hasPeerRouteEvidence, MEDIA_ROUTE_PRESENTATION, - mediaTransportPresentation, ROUTING_STATUS_PRESENTATION, viewerRouteEvidence, } from "../src/client/components/status-badge-model.ts"; @@ -89,50 +88,4 @@ describe("route status badges", () => { ).toBe("p2p"); }); - it("exposes TURN only from current assigned-edge evidence", () => { - const relayPeer = { - ...peerSnapshot, - metrics: { ...EMPTY_METRICS, path: "relay" as const }, - }; - expect( - viewerRouteEvidence( - { kind: "peer", peerId: "host-peer" }, - relayPeer, - null, - ).evidence?.metrics?.path, - ).toBe("relay"); - }); - - it("states a TURN protocol only from local relay evidence", () => { - expect( - mediaTransportPresentation({ - ...EMPTY_METRICS, - path: "relay", - localCandidateType: "relay", - localRelayProtocol: "udp", - }).label, - ).toBe("TURN/UDP"); - expect( - mediaTransportPresentation({ - ...EMPTY_METRICS, - path: "relay", - localCandidateType: "relay", - localRelayProtocol: "tls", - }).label, - ).toBe("TURN/TLS"); - expect( - mediaTransportPresentation({ - ...EMPTY_METRICS, - path: "relay", - localCandidateType: "host", - }).label, - ).toBe("TURN"); - expect( - mediaTransportPresentation({ - ...EMPTY_METRICS, - path: "direct", - iceProtocol: "udp", - }).label, - ).toBe("UDP"); - }); }); diff --git a/tests/turn-allocation-admission.test.ts b/tests/turn-allocation-admission.test.ts deleted file mode 100644 index 8311b8d..0000000 --- a/tests/turn-allocation-admission.test.ts +++ /dev/null @@ -1,140 +0,0 @@ -import { describe, expect, it } from "vitest"; - -import type { SelectedEdgeTurnIdentity } from "../src/server/selected-edge-turn.ts"; -import { TurnAllocationAdmission } from "../src/server/turn-allocation-admission.ts"; - -function peerFence( - roomId: string, - viewerPeerId: string, - newConnectionId = `connection-${roomId}-${viewerPeerId}`, -): SelectedEdgeTurnIdentity { - return { - edgeKind: "peer-selected", - roomId, - shareGeneration: `share-${roomId}`, - revision: 3, - parentPeerId: `parent-${viewerPeerId}`, - parentSessionId: `parent-session-${viewerPeerId}`, - viewerPeerId, - viewerSessionId: `viewer-session-${viewerPeerId}`, - oldConnectionId: `old-${viewerPeerId}`, - newConnectionId, - }; -} - -function ingressFence(roomId: string): SelectedEdgeTurnIdentity { - return { - edgeKind: "host-sfu-ingress", - roomId, - shareGeneration: `share-${roomId}`, - revision: 4, - hostPeerId: `host-${roomId}`, - hostSessionId: `host-session-${roomId}`, - publicationGeneration: `publication-${roomId}`, - oldConnectionId: `publication-${roomId}`, - newConnectionId: `ingress-${roomId}`, - }; -} - -describe("TURN allocation admission", () => { - it("admits independent peer and Host-SFU edges up to deployment capacity", () => { - const admission = new TurnAllocationAdmission({ capacity: 3 }); - const first = peerFence("room-a", "viewer-a"); - const second = peerFence("room-a", "viewer-b"); - const ingress = ingressFence("room-b"); - - expect(admission.reserve(first)).toBe(true); - expect(admission.reserve(second)).toBe(true); - expect(admission.reserve(ingress)).toBe(true); - expect(admission.usage()).toEqual({ allocations: 3 }); - }); - - it("keeps reserve, commit, and drain charged until exact completion", () => { - const admission = new TurnAllocationAdmission({ capacity: 1 }); - const fence = peerFence("room-a", "viewer-a"); - - expect(admission.reserve(fence)).toBe(true); - expect(admission.reserve(fence)).toBe(true); - expect(admission.state(fence)).toBe("reserved"); - expect(admission.commit(fence)).toBe(true); - expect(admission.commit(fence)).toBe(true); - expect(admission.state(fence)).toBe("committed"); - expect(admission.beginDrain(fence)).toBe(true); - expect(admission.beginDrain(fence)).toBe(true); - expect(admission.state(fence)).toBe("draining"); - expect(admission.usage()).toEqual({ allocations: 1 }); - expect(admission.completeDrain(fence)).toBe(true); - expect(admission.completeDrain(fence)).toBe(false); - expect(admission.usage()).toEqual({ allocations: 0 }); - }); - - it("rejects stale exact identity and allocation ID reuse", () => { - const admission = new TurnAllocationAdmission({ capacity: 2 }); - const current = peerFence("room-a", "viewer-a", "same-connection"); - const staleRevision = { ...current, revision: current.revision - 1 }; - const staleGeneration = { - ...current, - shareGeneration: `${current.shareGeneration}-stale`, - }; - const wrongEdge = peerFence("room-a", "viewer-b", "same-connection"); - - expect(admission.reserve(current)).toBe(true); - expect(admission.reserve(staleRevision)).toBe(false); - expect(admission.reserve(staleGeneration)).toBe(false); - expect(admission.reserve(wrongEdge)).toBe(false); - expect(admission.commit(staleRevision)).toBe(false); - expect(admission.beginDrain(staleGeneration)).toBe(false); - expect(admission.beginDrain(wrongEdge)).toBe(false); - expect(admission.state(current)).toBe("reserved"); - expect(admission.usage()).toEqual({ allocations: 1 }); - expect(admission.beginDrain(current)).toBe(true); - expect(admission.completeDrain(current)).toBe(true); - expect(admission.usage()).toEqual({ allocations: 0 }); - }); - - it("shares capacity exhaustion between Host ingress and peer edges", () => { - const admission = new TurnAllocationAdmission({ capacity: 1 }); - const first = ingressFence("room-a"); - const second = peerFence("room-b", "viewer-b"); - - expect(admission.reserve(first)).toBe(true); - expect(admission.reserve(second)).toBe(false); - expect(admission.state(first)).toBe("reserved"); - expect(admission.state(second)).toBeUndefined(); - expect(admission.usage()).toEqual({ allocations: 1 }); - }); - - it("returns capacity exhaustion immediately without queueing a retry", () => { - const admission = new TurnAllocationAdmission({ capacity: 1 }); - const active = ingressFence("room-a"); - const rejected = peerFence("room-b", "viewer-b"); - - expect(admission.reserve(active)).toBe(true); - expect(admission.reserve(rejected)).toBe(false); - expect(admission.beginDrain(active)).toBe(true); - expect(admission.completeDrain(active)).toBe(true); - expect(admission.state(rejected)).toBeUndefined(); - expect(admission.usage()).toEqual({ allocations: 0 }); - expect(admission.reserve(rejected)).toBe(true); - }); - - it("makes shutdown drain explicit while restart ownership stays process-local", () => { - const previousOwner = new TurnAllocationAdmission({ capacity: 2 }); - const first = peerFence("room-a", "viewer-a"); - const second = ingressFence("room-b"); - previousOwner.reserve(first); - previousOwner.reserve(second); - previousOwner.commit(first); - - const draining = previousOwner.beginDrainAll(); - expect(draining).toHaveLength(2); - expect(previousOwner.usage()).toEqual({ allocations: 2 }); - const restartedOwner = new TurnAllocationAdmission({ capacity: 2 }); - expect(restartedOwner.usage()).toEqual({ allocations: 0 }); - - for (const fence of draining) { - expect(previousOwner.completeDrain(fence)).toBe(true); - } - expect(previousOwner.usage()).toEqual({ allocations: 0 }); - }); -}); diff --git a/tests/turn-allocation-gate.test.ts b/tests/turn-allocation-gate.test.ts deleted file mode 100644 index 3faefea..0000000 --- a/tests/turn-allocation-gate.test.ts +++ /dev/null @@ -1,48 +0,0 @@ -import { describe, expect, it } from "vitest"; - -import { parseGateEnvironment, parseGateResult } from "../scripts/turn-allocation-gate"; - -describe("TURN UDP allocation gate", () => { - it("parses credential input without truncating values containing equals", () => { - expect( - parseGateEnvironment( - "SELECTED_EDGE_TURN_URLS=turn:relay.example:3478?transport=udp\n" + - "SELECTED_EDGE_TURN_SHARED_SECRET=secret=value==\n", - ), - ).toEqual({ - SELECTED_EDGE_TURN_URLS: "turn:relay.example:3478?transport=udp", - SELECTED_EDGE_TURN_SHARED_SECRET: "secret=value==", - }); - }); - - it("returns only allowlisted allocation fields from Chrome output", () => { - const secret = "must-not-leak"; - const result = parseGateResult( - '{"status":"passed","relayCandidateCount":1,"protocols":["udp","udp"],' + - `"errorCodeBuckets":[],"credential":"${secret}"}`, - ); - - expect(result).toEqual({ - status: "passed", - relayCandidateCount: 1, - protocols: ["udp"], - errorCodeBuckets: [], - }); - expect(JSON.stringify(result)).not.toContain(secret); - }); - - it("rejects pending or raw Chrome output with a fixed message", () => { - expect(() => parseGateResult('"pending"')).toThrow( - "Chrome returned an invalid TURN allocation result", - ); - expect(() => parseGateResult("credential=must-not-leak")).toThrow( - "Chrome returned an invalid TURN allocation result", - ); - expect(() => parseGateResult( - '{"status":"passed","relayCandidateCount":0,"protocols":[],"errorCodeBuckets":[]}', - )).toThrow("Chrome returned an invalid TURN allocation result"); - expect(() => parseGateResult( - '{"status":"passed","relayCandidateCount":1,"protocols":["tcp"],"errorCodeBuckets":[]}', - )).toThrow("Chrome returned an invalid TURN allocation result"); - }); -}); diff --git a/tsconfig.tools.json b/tsconfig.tools.json index 2b18416..5f0a3b5 100644 --- a/tsconfig.tools.json +++ b/tsconfig.tools.json @@ -21,7 +21,6 @@ "vitest.config.ts", "scripts/native-one-viewer-gate.ts", "scripts/native-one-viewer-ledger.ts", - "scripts/turn-allocation-gate.ts", "tests" ] }