refactor: simplify audited mechanisms and quality evidence (#314)

This commit is contained in:
AAA authored and GitHub committed 2026-08-28 18:19:15 +08:00
1 parent 7653164192
commit fe1e31c8bb
29 files changed
+761 -1107

No files matched your search

+45 -15
View File
@@ -6,6 +6,12 @@ on:
branches:
- main
workflow_dispatch:
inputs:
native_checks:
description: Run the optional Native sender research checks
required: false
default: false
type: boolean
permissions:
contents: read
@@ -18,26 +24,50 @@ jobs:
uses: actions/checkout@v6
with:
fetch-depth: 0
- name: Validate tracked project state
run: sh scripts/check-project-state.sh
env:
SCREENER_BASE_SHA: ${{ github.event.pull_request.base.sha || github.event.before }}
- name: Set up Go
uses: actions/setup-go@v6
with:
go-version: 1.26.6
cache-dependency-path: native/sender/go.sum
- name: Check native sender
working-directory: native/sender
run: |
go test ./...
go vet ./...
GOOS=windows GOARCH=amd64 CGO_ENABLED=0 go build -o "$RUNNER_TEMP/screener-sender.exe" ./cmd/screener-sender
- name: Set up Node.js
uses: actions/setup-node@v6
with:
node-version-file: .node-version
cache: npm
- name: Validate tracked project state
run: sh scripts/check-project-state.sh
env:
SCREENER_BASE_SHA: ${{ github.event.pull_request.base.sha || github.event.before }}
- name: Detect Native sender scope
id: native-scope
shell: bash
env:
BASE_SHA: ${{ github.event.pull_request.base.sha || github.event.before }}
EXPLICIT_NATIVE_CHECKS: ${{ inputs.native_checks || false }}
run: |
run_native=false
if [ "$EXPLICIT_NATIVE_CHECKS" = "true" ]; then
run_native=true
elif [ "$GITHUB_EVENT_NAME" != "workflow_dispatch" ]; then
base="$BASE_SHA"
case "$base" in
""|0000000000000000000000000000000000000000)
base=$(git rev-parse HEAD^ 2>/dev/null || true)
;;
esac
if [ -n "$base" ] && ! git diff --quiet "$base" HEAD -- native/sender; then
run_native=true
fi
fi
echo "run=$run_native" >> "$GITHUB_OUTPUT"
- name: Set up Go
if: steps.native-scope.outputs.run == 'true'
uses: actions/setup-go@v6
with:
go-version: 1.26.6
cache-dependency-path: native/sender/go.sum
- name: Check native sender
if: steps.native-scope.outputs.run == 'true'
working-directory: native/sender
run: |
go test ./...
go vet ./...
GOOS=windows GOARCH=amd64 CGO_ENABLED=0 go build -o "$RUNNER_TEMP/screener-sender.exe" ./cmd/screener-sender
- name: Install dependencies
run: npm ci
- name: Type-check, test, and build
@@ -232,8 +232,8 @@ deterministic candidate filters, ordering, cursor, reservations, and total
deadline. Candidate Peer parents need a usable source path and available steady
capacity; clear native paths are tried before the remaining deterministic Peer
candidates but are not a hard eligibility tier. Only one candidate runs at a
time and the old route keeps playing. The same Viewer compares fresh old and
candidate receive windows. A P2P candidate commits only after first decoded
time and the old route keeps playing. The same Viewer compares fresh,
overlapping old and candidate receive windows. A P2P candidate commits only after first decoded
frame and three consecutive windows with no freeze/pause, no lower pixel area or
rounded FPS, and a strict improvement in at least one of those dimensions.
Bitrate does not rank P2P candidates because codec and content phase make it
+1 -1
View File
@@ -111,7 +111,7 @@ per-share convergence gate is enabled, Screener uses an exact persistent native
sender limitation only to trigger one measured experiment through the existing
serial operation. The same Viewer must then prove a P2P candidate strictly
improves delivered pixel area or rounded FPS without regressing either dimension
or producing a freeze/pause across three fresh paired windows. A persistently
or producing a freeze/pause across three fresh overlapping windows. A persistently
limited exact candidate sender cannot commit. Ordinary quality moves remain
Peer-to-Peer. SFU may
join quality work only as Host fanout relief when every current Host-origin Peer
-17
View File
@@ -11,23 +11,6 @@ No implementation task is currently authorized. The audit root-cause
corrections are integrated and verified in current source but remain undeployed;
deployment requires explicit authority.
## Parked Repository Work
1. **Bounded mechanism simplification.** Remove only the proved zero-consumer or
duplicate owners: coarse `mediaAssignment` wire residue, legacy SFU room-name
acceptance, orphan `scripts/browser-codec-preflight.ts`, and the access gate's
test-title scan. The same bounded pass may remove the unused SFU publisher
snapshot callback, test-only production helpers, unused SFU token allowlist
input, debug ordinals after participant retirement, and share-unowned sender
evidence caches. Keep the production
`src/client/webrtc/video-codec-preflight.ts`, live internal media-assignment
helpers, and destructive-path safety gates.
2. **Tooling ownership correction.** Give every CDP call a deadline, move shared
Browser harness code out of the Native owner, type-check the access/privacy
gate, run repository hygiene under the pinned Node version, ignore untracked
drafts in canonical-document gates, and scope rejected Native checks to
Native changes or explicit research runs.
## Parked Product Work
1. **Broader quality optimization.** Weighted/global optimization,
+11 -9
View File
@@ -7,10 +7,16 @@ import { pathToFileURL } from "node:url";
import { createRoomResponseSchema } from "../src/shared/protocol";
import { createScreenerServer, type ScreenerServer } from "../src/server/app";
import type { ServerConfig } from "../src/server/config";
import { CdpConnection, cleanupRun, createPage, evaluate, reservePort, waitForVersion } from "./native-one-viewer-gate";
import {
CdpConnection,
cleanupRun,
createPage,
evaluate,
reservePort,
waitForVersion,
} from "./browser-gate-harness";
const root = resolve(import.meta.dirname, "..");
const nginxConfigPath = resolve(root, "deploy/nginx/share.bonfire.icu.conf.example");
const teardownTest = "strongly rotates and revokes every Viewer generation and media edge";
const fragmentProbe = `(() => {
let domContentLoaded = false;
const state = { fragmentAtStart: location.hash.length > 0, replaced: false, beforeDomContentLoaded: false };
@@ -32,7 +38,7 @@ interface GateReport {
cookieGrantMatches: number; resourceGrantMatches: number;
requestLeakCount: number; nginxLogLeakCount: number; nginxConfigSafe: boolean;
appLogLeakCount: number; persistentRoomFileCount: number; leakAuditCompleted: boolean;
rotateRevokeCasePresent: boolean; cleanupPassed: boolean;
cleanupPassed: boolean;
}
interface BrowserAudit {
fragmentSeenAtDocumentStart: boolean; fragmentReplaceObserved: boolean;
@@ -65,8 +71,7 @@ async function main(): Promise<void> {
otherSessionGrantMatches: 0, localGrantMatches: 0, cookieGrantMatches: 0,
resourceGrantMatches: 0, requestLeakCount: 0, nginxLogLeakCount: 0,
nginxConfigSafe: false, appLogLeakCount: 0, persistentRoomFileCount: 0,
leakAuditCompleted: false, rotateRevokeCasePresent: false,
cleanupPassed: false,
leakAuditCompleted: false, cleanupPassed: false,
};
const chromePath = process.env.CHROME_PATH?.trim() ?? "";
const sitePassword = secret();
@@ -84,9 +89,6 @@ async function main(): Promise<void> {
console.error = (...values: unknown[]) => { appLogs.push(diagnosticText(values)); };
try {
report.rotateRevokeCasePresent = (await readFile(
resolve(root, "tests/server-signal.test.ts"), "utf8",
)).includes(teardownTest);
await access(chromePath);
appPort = await reservePort();
do { debugPort = await reservePort(); } while (debugPort === appPort);
@@ -197,7 +199,7 @@ async function main(): Promise<void> {
report.leakAuditCompleted && report.requestLeakCount === 0 &&
report.nginxLogLeakCount === 0 && report.nginxConfigSafe &&
report.appLogLeakCount === 0 && report.persistentRoomFileCount === 0 &&
report.rotateRevokeCasePresent && report.cleanupPassed;
report.cleanupPassed;
process.stdout.write(`${JSON.stringify(report)}\n`);
process.exitCode = report.passed ? 0 : 1;
}
-384
View File
@@ -1,384 +0,0 @@
import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process";
import { mkdir, mkdtemp, readFile, writeFile } from "node:fs/promises";
import { createServer, type Server } from "node:http";
import { tmpdir } from "node:os";
import { join, resolve } from "node:path";
import { AccessToken, TrackSource } from "livekit-server-sdk";
import {
CdpConnection,
cleanupRun,
createPage,
reservePort,
waitForVersion,
type PageHandle,
} from "./native-one-viewer-gate";
const WIDTH = 1_920;
const HEIGHT = 1_080;
const FRAME_RATE = 30;
const MAX_BITRATE = 5_000_000;
const SETTLE_MS = 5_000;
const SAMPLE_MS = 10_000;
interface RuntimeResult<T> {
result: { value?: T };
exceptionDetails?: {
text?: string;
exception?: { description?: string };
};
}
interface ProcessInfo {
type: string;
id: number;
cpuTime: number;
}
interface ProcessSnapshot {
processInfo: ProcessInfo[];
}
async function evaluate<T>(
cdp: CdpConnection,
page: PageHandle,
expression: string,
timeoutMs: number,
): Promise<T> {
const value = await cdp.call<RuntimeResult<T>>(
"Runtime.evaluate",
{
expression,
awaitPromise: true,
returnByValue: true,
userGesture: true,
},
page.sessionId,
Date.now() + timeoutMs,
);
if (value.exceptionDetails) {
throw new Error(
value.exceptionDetails.exception?.description ??
value.exceptionDetails.text ??
"Browser evaluation failed",
);
}
return value.result.value as T;
}
function sourceHtml(): string {
return `<!doctype html><title>Codec Preflight Source</title>
<style>html,body{margin:0;overflow:hidden;background:#000}canvas{display:block;width:100vw;height:100vh}</style>
<canvas width="${WIDTH}" height="${HEIGHT}"></canvas><script>
const c=document.querySelector('canvas'),x=c.getContext('2d',{alpha:false});let f=0;
function draw(){f+=1;for(let r=0;r<14;r+=1)for(let q=0;q<24;q+=1){
x.fillStyle='hsl('+((f*29+r*71+q*43)%360)+' 80% '+(35+(f+r+q)%30)+'%)';
x.fillRect(q*80,r*78,80,78)}x.fillStyle='#fff';x.fillRect(f*13%1920,0,12,1080);
x.fillStyle='#000';x.font='60px monospace';x.fillText(String(f).padStart(8,'0'),24,72)}
draw();setInterval(draw,1000/${FRAME_RATE});</script>`;
}
async function startServer(port: number): Promise<Server> {
const livekitBundle = await readFile(
resolve("node_modules/livekit-client/dist/livekit-client.umd.js"),
);
const server = createServer((request, response) => {
const path = new URL(request.url ?? "/", `http://127.0.0.1:${port}`).pathname;
if (path === "/livekit.js") {
response.setHeader("content-type", "text/javascript; charset=utf-8");
response.end(livekitBundle);
return;
}
response.setHeader("content-type", "text/html; charset=utf-8");
response.end(path === "/source"
? sourceHtml()
: "<!doctype html><title>Codec Preflight</title><script src='/livekit.js'></script><video autoplay muted playsinline style='width:960px'></video>");
});
await new Promise<void>((resolveListen, rejectListen) => {
server.once("error", rejectListen);
server.listen(port, "127.0.0.1", resolveListen);
});
return server;
}
function browserProbe(): string {
return `(() => {
const WIDTH=${WIDTH},HEIGHT=${HEIGHT},FRAME_RATE=${FRAME_RATE},MAX_BITRATE=${MAX_BITRATE};
const SETTLE_MS=${SETTLE_MS},SAMPLE_MS=${SAMPLE_MS};let stream=null,track=null;
const delay=(ms)=>new Promise((resolve)=>setTimeout(resolve,ms));
async function waitFor(predicate,timeout,label){const deadline=performance.now()+timeout;
while(performance.now()<deadline){if(await predicate())return;await delay(50)}
throw new Error('Timed out waiting for '+label)}
function preferences(codec){const all=RTCRtpSender.getCapabilities('video')?.codecs||[];
const mime=codec==='h264'?'video/h264':'video/vp8';const primary=all.filter((item)=>item.mimeType.toLowerCase()===mime);
if(!primary.length)throw new Error(mime+' unavailable');
const repair=new Set(['video/rtx','video/red','video/ulpfec','video/flexfec-03']);
return [...primary,...all.filter((item)=>repair.has(item.mimeType.toLowerCase()))]}
async function gathered(pc){if(pc.iceGatheringState==='complete')return;
await new Promise((resolve)=>{const changed=()=>{if(pc.iceGatheringState!=='complete')return;
pc.removeEventListener('icegatheringstatechange',changed);resolve()};pc.addEventListener('icegatheringstatechange',changed)})}
async function negotiate(a,b){await a.setLocalDescription(await a.createOffer());await gathered(a);
await b.setRemoteDescription(a.localDescription);await b.setLocalDescription(await b.createAnswer());
await gathered(b);await a.setRemoteDescription(b.localDescription)}
function statsValue(item){if(!item)return null;const keys=['id','timestamp','bytesSent','bytesReceived',
'framesEncoded','framesDecoded','framesDropped','framesPerSecond','frameWidth','frameHeight','totalEncodeTime',
'totalDecodeTime','qualityLimitationReason','qualityLimitationDurations','encoderImplementation',
'decoderImplementation','powerEfficientEncoder','powerEfficientDecoder','frames'];const value={};
for(const key of keys)value[key]=item[key]??null;return value}
async function snapshot(sender,receiver,rendered){const [sendReport,receiveReport]=await Promise.all([sender.getStats(),receiver.getStats()]);
let outbound=null,inbound=null;sendReport.forEach((item)=>{if(item.type==='outbound-rtp'&&item.kind==='video'&&!item.isRemote)outbound=item});
receiveReport.forEach((item)=>{if(item.type==='inbound-rtp'&&item.kind==='video'&&!item.isRemote)inbound=item});
const sendCodec=outbound?.codecId?sendReport.get(outbound.codecId):null;
const receiveCodec=inbound?.codecId?receiveReport.get(inbound.codecId):null;
const mediaSource=outbound?.mediaSourceId?sendReport.get(outbound.mediaSourceId):null;
const codec=(item)=>item?{mimeType:item.mimeType||null,sdpFmtpLine:item.sdpFmtpLine||null}:null;
return {rendered:rendered(),sender:statsValue(outbound),receiver:statsValue(inbound),
mediaSource:statsValue(mediaSource),senderCodec:codec(sendCodec),receiverCodec:codec(receiveCodec)}}
function delta(after,before,key){const a=after?.[key],b=before?.[key];return Number.isFinite(a)&&Number.isFinite(b)&&a>=b?a-b:null}
function summarize(before,after,sampleMs){const encoded=delta(after.sender,before.sender,'framesEncoded');
const decoded=delta(after.receiver,before.receiver,'framesDecoded');const bytes=delta(after.sender,before.sender,'bytesSent');
const sourceFrames=delta(after.mediaSource,before.mediaSource,'frames');
const encodeTime=delta(after.sender,before.sender,'totalEncodeTime');const decodeTime=delta(after.receiver,before.receiver,'totalDecodeTime');
return {senderCodec:after.senderCodec,receiverCodec:after.receiverCodec,sender:after.sender,receiver:after.receiver,
measured:{sampleMs,encodedFps:encoded===null?null:encoded*1000/sampleMs,
decodedFps:decoded===null?null:decoded*1000/sampleMs,bitrateKbps:bytes===null?null:bytes*8/sampleMs,
encodeMsPerFrame:encodeTime!==null&&encoded>0?encodeTime*1000/encoded:null,
decodeMsPerFrame:decodeTime!==null&&decoded>0?decodeTime*1000/decoded:null,
sourceFps:sourceFrames===null?after.mediaSource?.framesPerSecond:sourceFrames*1000/sampleMs,
encodedSourceRatio:sourceFrames>0&&encoded!==null?encoded/sourceFrames:null,
renderedFrames:delta(after,before,'rendered')}}}
async function setup(){stream=await navigator.mediaDevices.getDisplayMedia({video:{
width:{ideal:WIDTH,max:WIDTH},height:{ideal:HEIGHT,max:HEIGHT},frameRate:{ideal:FRAME_RATE,max:FRAME_RATE}},audio:false});
track=stream.getVideoTracks()[0];if(!track)throw new Error('No display video track');track.contentHint='motion';
return {settings:track.getSettings(),constraints:track.getConstraints(),contentHint:track.contentHint,label:track.label}}
function capabilities(){const simple=(codec)=>({mimeType:codec.mimeType,sdpFmtpLine:codec.sdpFmtpLine||null});
return {sender:(RTCRtpSender.getCapabilities('video')?.codecs||[]).map(simple),
receiver:(RTCRtpReceiver.getCapabilities('video')?.codecs||[]).map(simple)}}
async function edge(codec,sourceTrack){
const sendPc=new RTCPeerConnection({iceServers:[]}),receivePc=new RTCPeerConnection({iceServers:[]});
const transceiver=sendPc.addTransceiver(sourceTrack,{direction:'sendonly'});transceiver.setCodecPreferences(preferences(codec));
let receiver=null,remoteTrack=null;receivePc.addEventListener('track',(event)=>{receiver=event.receiver;remoteTrack=event.track});
await negotiate(sendPc,receivePc);await waitFor(()=>sendPc.connectionState==='connected'&&receivePc.connectionState==='connected',10000,'connection');
await waitFor(()=>receiver&&remoteTrack,5000,'remote track');const video=document.createElement('video');let rendered=0;
video.autoplay=true;video.muted=true;video.playsInline=true;document.body.append(video);
video.srcObject=new MediaStream([remoteTrack]);if(video.requestVideoFrameCallback){const frame=()=>{rendered+=1;video.requestVideoFrameCallback(frame)};video.requestVideoFrameCallback(frame)}
await video.play();const parameters=transceiver.sender.getParameters();if(!parameters.encodings.length)parameters.encodings=[{}];
parameters.encodings[0].maxBitrate=MAX_BITRATE;parameters.encodings[0].maxFramerate=FRAME_RATE;
parameters.encodings[0].scaleResolutionDownBy=1;parameters.degradationPreference='balanced';await transceiver.sender.setParameters(parameters);
await waitFor(async()=>{const report=await receiver.getStats();let frames=0;report.forEach((item)=>{if(item.type==='inbound-rtp'&&item.kind==='video')frames=Math.max(frames,item.framesDecoded||0)});return frames>0&&rendered>0},10000,'decoded frame');
return {sendPc,receivePc,sender:transceiver.sender,receiver,remoteTrack,rendered:()=>rendered}}
async function run(codec,count,topology){if(!track)throw new Error('Capture not initialized');const edges=[];
try{edges.push(await edge(codec,track));
if(topology==='relay')edges.push(await edge(codec,edges[0].remoteTrack));
else for(let index=1;index<count;index+=1)edges.push(await edge(codec,track));await delay(SETTLE_MS);
const before=await Promise.all(edges.map((item)=>snapshot(item.sender,item.receiver,item.rendered))),started=performance.now();
await delay(SAMPLE_MS);const sampleMs=performance.now()-started;
const after=await Promise.all(edges.map((item)=>snapshot(item.sender,item.receiver,item.rendered)));
return edges.map((_,index)=>summarize(before[index],after[index],sampleMs))}
finally{for(const item of edges){item.sendPc.close();item.receivePc.close()}await delay(300)}}
async function runLivekit(codec,url,publisherToken,subscriberToken){if(!track)throw new Error('Capture not initialized');
const sdk=globalThis.LivekitClient,publisher=new sdk.Room({dynacast:true}),subscriber=new sdk.Room();let remoteTrack=null,rendered=0;
subscriber.on(sdk.RoomEvent.TrackSubscribed,(nextTrack)=>{if(nextTrack.kind==='video')remoteTrack=nextTrack});
try{await publisher.connect(url,publisherToken,{autoSubscribe:false,rtcConfig:{iceServers:[]}});
await subscriber.connect(url,subscriberToken,{rtcConfig:{iceServers:[]}});
const publication=await publisher.localParticipant.publishTrack(track,{source:sdk.Track.Source.ScreenShare,
videoCodec:codec,backupCodec:false,screenShareEncoding:{maxBitrate:MAX_BITRATE,maxFramerate:FRAME_RATE},
degradationPreference:'balanced'});
await waitFor(()=>publication.videoTrack?.sender&&remoteTrack,15000,'LiveKit publication');
const video=document.createElement('video');video.autoplay=true;video.muted=true;video.playsInline=true;document.body.append(video);
video.srcObject=new MediaStream([remoteTrack.mediaStreamTrack]);if(video.requestVideoFrameCallback){const frame=()=>{rendered+=1;video.requestVideoFrameCallback(frame)};video.requestVideoFrameCallback(frame)}
await video.play();const receiver={getStats:async()=>{const report=await remoteTrack.getRTCStatsReport();if(!report)throw new Error('LiveKit receiver stats unavailable');return report}};
await waitFor(async()=>{const report=await receiver.getStats();let frames=0;report?.forEach((item)=>{if(item.type==='inbound-rtp'&&item.kind==='video')frames=Math.max(frames,item.framesDecoded||0)});return frames>0&&rendered>0},15000,'LiveKit decoded frame');
await delay(SETTLE_MS);const sender=publication.videoTrack.sender;
const before=await snapshot(sender,receiver,()=>rendered),started=performance.now();await delay(SAMPLE_MS);
const sampleMs=performance.now()-started,after=await snapshot(sender,receiver,()=>rendered);
return {...summarize(before,after,sampleMs),senderParameters:sender.getParameters()}}
finally{await Promise.allSettled([publisher.disconnect(false),subscriber.disconnect(false)]);await delay(300)}}
function stop(){stream?.getTracks().forEach((item)=>item.stop())}
Object.defineProperty(globalThis,'__CODEC_PREFLIGHT__',{value:{setup,capabilities,run,runLivekit,stop}});
})()`;
}
async function processSnapshot(cdp: CdpConnection): Promise<ProcessSnapshot> {
return cdp.call<ProcessSnapshot>(
"SystemInfo.getProcessInfo",
{},
undefined,
Date.now() + 5_000,
);
}
function processCpu(
before: ProcessSnapshot,
after: ProcessSnapshot,
wallMs: number,
): Record<string, unknown> {
const baseline = new Map(
before.processInfo.map((item) => [`${item.type}:${item.id}`, item.cpuTime]),
);
let cpuSeconds = 0;
let stableProcesses = 0;
for (const item of after.processInfo) {
const previous = baseline.get(`${item.type}:${item.id}`);
if (previous === undefined || item.cpuTime < previous) continue;
cpuSeconds += item.cpuTime - previous;
stableProcesses += 1;
}
const wallSeconds = wallMs / 1_000;
return {
wallSeconds,
cpuSeconds,
logicalCoreEquivalent: cpuSeconds / wallSeconds,
stableProcesses,
processIdentityChanged:
stableProcesses !== before.processInfo.length ||
stableProcesses !== after.processInfo.length,
};
}
async function main(): Promise<void> {
const chromePath = process.env.CHROME_PATH?.trim();
if (!chromePath) throw new Error("CHROME_PATH must point to Chrome");
const codec = process.env.CODEC_PREFLIGHT_CODEC?.trim().toLowerCase();
if (codec !== "vp8" && codec !== "h264") {
throw new Error("CODEC_PREFLIGHT_CODEC must be vp8 or h264");
}
const edges = Number(process.env.CODEC_PREFLIGHT_EDGES ?? "1");
if (edges !== 1 && edges !== 2) {
throw new Error("CODEC_PREFLIGHT_EDGES must be 1 or 2");
}
const topology = process.env.CODEC_PREFLIGHT_TOPOLOGY?.trim() || "direct";
if (topology !== "direct" && topology !== "relay" && topology !== "livekit") {
throw new Error("CODEC_PREFLIGHT_TOPOLOGY must be direct, relay, or livekit");
}
const [debugPort, pagePort] = await Promise.all([reservePort(), reservePort()]);
const profile = await mkdtemp(join(tmpdir(), "screener-codec-preflight-"));
let chrome: ChildProcessWithoutNullStreams | null = null;
let cdp: CdpConnection | null = null;
let server: Server | null = null;
try {
server = await startServer(pagePort);
chrome = spawn(chromePath, [
`--remote-debugging-port=${debugPort}`,
`--user-data-dir=${profile}`,
"--no-first-run",
"--no-default-browser-check",
"--disable-extensions",
"--disable-background-timer-throttling",
"--disable-backgrounding-occluded-windows",
"--disable-renderer-backgrounding",
"--autoplay-policy=no-user-gesture-required",
"--auto-select-tab-capture-source-by-title=Codec Preflight Source",
"--window-size=1920,1080",
"about:blank",
], { stdio: "pipe", windowsHide: true });
chrome.stdout.resume();
chrome.stderr.resume();
const version = await waitForVersion(debugPort, chrome);
cdp = await CdpConnection.connect(version.webSocketDebuggerUrl, Date.now() + 5_000);
const gpuInfo = await cdp.call<unknown>("SystemInfo.getInfo", {}, undefined, Date.now() + 5_000);
await createPage(cdp, `http://127.0.0.1:${pagePort}/source`, undefined, true);
const page = await createPage(cdp, `http://127.0.0.1:${pagePort}/test`, undefined, true);
await evaluate<void>(cdp, page, browserProbe(), 5_000);
const capabilities = await evaluate<unknown>(
cdp,
page,
"globalThis.__CODEC_PREFLIGHT__.capabilities()",
5_000,
);
const capture = await evaluate<unknown>(
cdp,
page,
"globalThis.__CODEC_PREFLIGHT__.setup()",
15_000,
);
const before = await processSnapshot(cdp);
const started = performance.now();
let media: unknown;
if (topology === "livekit") {
const livekitUrl = process.env.CODEC_PREFLIGHT_LIVEKIT_URL?.trim();
const apiKey = process.env.CODEC_PREFLIGHT_LIVEKIT_KEY?.trim();
const apiSecret = process.env.CODEC_PREFLIGHT_LIVEKIT_SECRET?.trim();
if (!livekitUrl || !apiKey || !apiSecret) {
throw new Error("LiveKit preflight credentials are missing");
}
const room = `codec-preflight-${Date.now()}`;
const token = async (identity: string, publish: boolean) => {
const access = new AccessToken(apiKey, apiSecret, { identity, ttl: 60 });
access.addGrant({
roomJoin: true,
room,
canPublish: publish,
canPublishSources: publish ? [TrackSource.SCREEN_SHARE] : undefined,
canSubscribe: !publish,
canPublishData: false,
});
return access.toJwt();
};
const [publisherToken, subscriberToken] = await Promise.all([
token("host", true),
token("viewer", false),
]);
media = await evaluate<unknown>(
cdp,
page,
`globalThis.__CODEC_PREFLIGHT__.runLivekit(${JSON.stringify(codec)}, ${JSON.stringify(livekitUrl)}, ${JSON.stringify(publisherToken)}, ${JSON.stringify(subscriberToken)})`,
60_000,
);
} else {
media = await evaluate<unknown>(
cdp,
page,
`globalThis.__CODEC_PREFLIGHT__.run(${JSON.stringify(codec)}, ${edges}, ${JSON.stringify(topology)})`,
45_000,
);
}
const wallMs = performance.now() - started;
const after = await processSnapshot(cdp);
await evaluate<void>(cdp, page, "globalThis.__CODEC_PREFLIGHT__.stop()", 5_000);
const report = {
schemaVersion: 1,
browser: version.Browser,
configuration: {
width: WIDTH,
height: HEIGHT,
frameRate: FRAME_RATE,
maxBitrate: MAX_BITRATE,
contentHint: "motion",
settleMs: SETTLE_MS,
sampleMs: SAMPLE_MS,
edges,
topology,
},
gpuInfo,
capabilities,
capture,
codec,
media,
cpu: processCpu(before, after, wallMs),
};
const outputDirectory = resolve("benchmark-results");
await mkdir(outputDirectory, { recursive: true });
const output = join(
outputDirectory,
`browser-codec-preflight-${codec}-${topology}-${edges}.json`,
);
await writeFile(output, `${JSON.stringify(report, null, 2)}\n`, "utf8");
console.log(output);
} finally {
await cleanupRun({
cdp,
native: null,
chrome,
server: null,
profile,
ports: [debugPort],
});
if (server) {
await new Promise<void>((resolveClose) => server!.close(() => resolveClose()));
}
}
}
await main();
+538
View File
@@ -0,0 +1,538 @@
import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process";
import { lstat, readdir, rm } from "node:fs/promises";
import { createConnection, createServer } from "node:net";
import { tmpdir } from "node:os";
import { basename, dirname, join, resolve } from "node:path";
import WebSocket from "ws";
export interface PageHandle {
sessionId: string;
}
interface CdpResult<T> {
result: { value?: T };
exceptionDetails?: unknown;
}
export interface CleanupResult {
browserExited: boolean;
nativeExited: boolean;
serverClosed: boolean;
portsClosed: boolean;
profileRemoved: boolean;
}
interface AsyncCloseable {
close(): Promise<void>;
}
export interface CleanupResources {
cdp: CdpConnection | null;
native: ChildProcessWithoutNullStreams | null;
chrome: ChildProcessWithoutNullStreams | null;
server: AsyncCloseable | null;
profile: string | null;
ports: number[];
}
export class CdpConnection {
private nextId = 1;
private readonly pending = new Map<number, {
resolve: (value: unknown) => void;
reject: (error: Error) => void;
timeout: NodeJS.Timeout;
}>();
private closed = false;
private constructor(private readonly socket: WebSocket) {
socket.on("message", (raw) => this.onMessage(raw.toString()));
socket.on("close", () => this.closePending());
socket.on("error", () => this.closePending());
}
static async connect(url: string, deadline: number): Promise<CdpConnection> {
const socket = new WebSocket(url, { maxPayload: 16 * 1024 * 1024 });
try {
await withDeadline(() => new Promise<void>((resolveOpen, rejectOpen) => {
socket.once("open", resolveOpen);
socket.once("error", rejectOpen);
}), deadline);
} catch (error) {
socket.terminate();
throw error;
}
return new CdpConnection(socket);
}
call<T>(
method: string,
params: Record<string, unknown>,
sessionId: string | undefined,
deadline: number,
): Promise<T> {
if (this.closed || this.socket.readyState !== WebSocket.OPEN) {
return Promise.reject(new Error(`Cannot call ${method}: CDP is closed`));
}
const id = this.nextId++;
return new Promise<T>((resolveCall, rejectCall) => {
const timeout = setTimeout(() => {
this.pending.delete(id);
rejectCall(new Error(`CDP command timed out: ${method}`));
}, Math.max(0, deadline - Date.now()));
this.pending.set(id, {
resolve: (value) => {
clearTimeout(timeout);
resolveCall(value as T);
},
reject: (error) => {
clearTimeout(timeout);
rejectCall(error);
},
timeout,
});
try {
this.socket.send(JSON.stringify({ id, method, params, sessionId }));
} catch (error) {
clearTimeout(timeout);
this.pending.delete(id);
rejectCall(error instanceof Error ? error : new Error("CDP send failed"));
}
});
}
close(): void {
if (!this.closed) this.socket.close();
}
private onMessage(raw: string): void {
let message: { id?: number; result?: unknown; error?: unknown };
try {
message = JSON.parse(raw) as typeof message;
} catch {
return;
}
if (message.id === undefined) return;
const pending = this.pending.get(message.id);
if (!pending) return;
this.pending.delete(message.id);
if (message.error) {
pending.reject(new Error("CDP command failed"));
} else {
pending.resolve(message.result ?? {});
}
}
private closePending(): void {
if (this.closed) return;
this.closed = true;
for (const pending of this.pending.values()) {
clearTimeout(pending.timeout);
pending.reject(new Error("CDP connection closed"));
}
this.pending.clear();
}
}
export function withDeadline<T>(
operation: () => Promise<T>,
deadline: number,
now: () => number = Date.now,
): Promise<T> {
const remaining = Math.max(0, deadline - now());
return new Promise<T>((resolveValue, rejectValue) => {
const timeout = setTimeout(
() => rejectValue(new Error("Operation deadline exceeded")),
remaining,
);
Promise.resolve().then(operation).then(
(value) => {
clearTimeout(timeout);
resolveValue(value);
},
(error: unknown) => {
clearTimeout(timeout);
rejectValue(error);
},
);
});
}
type JsonFetch = (
input: string,
init: { signal: AbortSignal },
) => Promise<{ ok: boolean; json(): Promise<unknown> }>;
export async function fetchJsonBefore<T>(
url: string,
deadline: number,
fetcher: JsonFetch = fetch,
): Promise<T> {
const controller = new AbortController();
const abort = setTimeout(
() => controller.abort(),
Math.max(0, deadline - Date.now()),
);
try {
const response = await withDeadline(
() => fetcher(url, { signal: controller.signal }),
deadline,
);
if (!response.ok) throw new Error("HTTP response was not successful");
return await withDeadline(() => response.json() as Promise<T>, deadline);
} finally {
clearTimeout(abort);
controller.abort();
}
}
export async function waitForSample<T>(
sample: (deadline: number) => Promise<T>,
retainAndAccept: (value: T) => boolean,
timeoutMs: number,
pollMs = 100,
): Promise<void> {
const deadline = Date.now() + timeoutMs;
while (Date.now() < deadline) {
let value: T | undefined;
try {
value = await withDeadline(() => sample(deadline), deadline);
} catch {}
if (value !== undefined && retainAndAccept(value)) return;
await delay(Math.min(pollMs, Math.max(0, deadline - Date.now())));
}
throw new Error("Bounded browser stage timed out");
}
export async function createPage(
cdp: CdpConnection,
url: string,
probe?: string,
newWindow = false,
browserContextId?: string,
): Promise<PageHandle> {
const deadline = Date.now() + 15_000;
const created = await cdp.call<{ targetId: string }>(
"Target.createTarget",
{
url: "about:blank",
background: false,
newWindow,
...(browserContextId ? { browserContextId } : {}),
},
undefined,
deadline,
);
const attached = await cdp.call<{ sessionId: string }>(
"Target.attachToTarget",
{ targetId: created.targetId, flatten: true },
undefined,
deadline,
);
const page = { sessionId: attached.sessionId };
await Promise.all([
cdp.call("Page.enable", {}, page.sessionId, deadline),
cdp.call("Runtime.enable", {}, page.sessionId, deadline),
]);
if (probe) {
await cdp.call(
"Page.addScriptToEvaluateOnNewDocument",
{ source: probe },
page.sessionId,
deadline,
);
}
await cdp.call("Page.navigate", { url }, page.sessionId, deadline);
await waitForSample(
(sampleDeadline) => evaluate(
cdp,
page,
"document.readyState === 'complete'",
sampleDeadline,
),
Boolean,
Math.max(1, deadline - Date.now()),
);
return page;
}
export async function evaluate<T>(
cdp: CdpConnection,
page: PageHandle,
expression: string,
deadline: number,
): Promise<T> {
const value = await cdp.call<CdpResult<T>>(
"Runtime.evaluate",
{
expression,
awaitPromise: true,
returnByValue: true,
userGesture: true,
},
page.sessionId,
deadline,
);
if (value.exceptionDetails) throw new Error("Browser evaluation failed");
return value.result.value as T;
}
export async function reservePort(): Promise<number> {
const server = createServer();
return new Promise((resolvePort, rejectPort) => {
server.once("error", rejectPort);
server.listen(0, "127.0.0.1", () => {
const address = server.address();
if (!address || typeof address === "string") {
rejectPort(new Error("Port reservation failed"));
return;
}
server.close((error) =>
error ? rejectPort(error) : resolvePort(address.port),
);
});
});
}
export async function waitForVersion(
port: number,
child: ChildProcessWithoutNullStreams,
): Promise<{ Browser: string; webSocketDebuggerUrl: string }> {
const deadline = Date.now() + 20_000;
while (Date.now() < deadline && child.exitCode === null) {
try {
return await fetchJsonBefore<{ Browser: string; webSocketDebuggerUrl: string }>(
`http://127.0.0.1:${port}/json/version`,
deadline,
);
} catch {}
await delay(Math.min(100, Math.max(0, deadline - Date.now())));
}
throw new Error("Chrome startup timed out");
}
export async function cleanupRun(
resources: CleanupResources,
): Promise<CleanupResult> {
if (resources.cdp) {
try {
await resources.cdp.call(
"Browser.close",
{},
undefined,
Date.now() + 2_000,
);
} catch {}
resources.cdp.close();
}
const browserExited = resources.chrome
? await stopProcessTree(resources.chrome)
: true;
if (resources.native && processRunning(resources.native)) {
try {
resources.native.stdin.write("\n");
} catch {}
}
const nativeExited = resources.native
? await stopProcessTree(resources.native)
: true;
let serverClosed = true;
if (resources.server) {
try {
await withDeadline(() => resources.server!.close(), Date.now() + 5_000);
} catch {
serverClosed = false;
}
}
const portsClosed = await waitForPortsClosed(
[...new Set(resources.ports)],
5_000,
);
const profileRemoved = resources.profile
? await removeGateProfile(resources.profile)
: true;
return {
browserExited,
nativeExited,
serverClosed,
portsClosed,
profileRemoved,
};
}
function processRunning(child: ChildProcessWithoutNullStreams): boolean {
return child.exitCode === null && child.signalCode === null;
}
async function stopProcessTree(
child: ChildProcessWithoutNullStreams,
): Promise<boolean> {
if (!processRunning(child)) return true;
if (await waitForExit(child, 2_000)) return true;
if (process.platform === "win32" && child.pid !== undefined) {
await runBounded(
"taskkill.exe",
["/PID", String(child.pid), "/T", "/F"],
5_000,
);
} else {
child.kill("SIGTERM");
if (await waitForExit(child, 2_000)) return true;
child.kill("SIGKILL");
}
return waitForExit(child, 3_000);
}
async function waitForExit(
child: ChildProcessWithoutNullStreams,
timeoutMs: number,
): Promise<boolean> {
if (!processRunning(child)) return true;
return new Promise<boolean>((resolveExit) => {
const onExit = () => {
clearTimeout(timeout);
resolveExit(true);
};
const timeout = setTimeout(() => {
child.off("exit", onExit);
resolveExit(!processRunning(child));
}, timeoutMs);
child.once("exit", onExit);
});
}
async function runBounded(
command: string,
args: string[],
timeoutMs: number,
environment: NodeJS.ProcessEnv = process.env,
): Promise<boolean> {
const child = spawn(command, args, {
stdio: "pipe",
windowsHide: true,
env: environment,
});
child.stdout.resume();
child.stderr.resume();
try {
const code = await withDeadline(
() => new Promise<number | null>((resolveExit, rejectExit) => {
child.once("error", rejectExit);
child.once("exit", resolveExit);
}),
Date.now() + timeoutMs,
);
return code === 0;
} catch {
child.kill();
return false;
}
}
async function waitForPortsClosed(
ports: number[],
timeoutMs: number,
): Promise<boolean> {
if (ports.length === 0) return true;
try {
await waitForSample(
(deadline) => Promise.all(ports.map((port) => portClosed(port, deadline))),
(closed) => closed.every(Boolean),
timeoutMs,
);
return true;
} catch {
return false;
}
}
function portClosed(port: number, deadline: number): Promise<boolean> {
return withDeadline(() => new Promise<boolean>((resolveClosed) => {
const socket = createConnection({ host: "127.0.0.1", port });
const finish = (closed: boolean) => {
socket.destroy();
resolveClosed(closed);
};
socket.once("connect", () => finish(false));
socket.once("error", () => finish(true));
}), deadline);
}
export const profileCleanupScript = String.raw`
$ErrorActionPreference = 'Stop'
$root = [IO.Path]::GetFullPath($env:SCREENER_GATE_PROFILE_TO_REMOVE)
$temp = [IO.Path]::GetFullPath([IO.Path]::GetTempPath()).TrimEnd([IO.Path]::DirectorySeparatorChar)
$parent = [IO.Path]::GetDirectoryName($root).TrimEnd([IO.Path]::DirectorySeparatorChar)
$name = [IO.Path]::GetFileName($root)
if ($parent -ine $temp -or $name -notmatch '^screener-(native-one-viewer|access-privacy)-[A-Za-z0-9_-]{6}$') { exit 31 }
$rootEntries = @([IO.Directory]::EnumerateFileSystemEntries($parent, $name, [IO.SearchOption]::TopDirectoryOnly))
if ($rootEntries.Count -eq 0) { exit 0 }
if ($rootEntries.Count -ne 1) { exit 34 }
$root = $rootEntries[0]
$rootAttributes = [IO.File]::GetAttributes($root)
if (($rootAttributes -band [IO.FileAttributes]::ReparsePoint) -ne 0) { exit 32 }
if (($rootAttributes -band [IO.FileAttributes]::Directory) -eq 0) { exit 34 }
$pending = [Collections.Generic.Stack[string]]::new()
$pending.Push($root)
while ($pending.Count -gt 0) {
$directory = $pending.Pop()
$attributes = [IO.File]::GetAttributes($directory)
if (($attributes -band [IO.FileAttributes]::ReparsePoint) -ne 0) { exit 32 }
foreach ($entry in [IO.Directory]::EnumerateFileSystemEntries($directory)) {
$attributes = [IO.File]::GetAttributes($entry)
if (($attributes -band [IO.FileAttributes]::ReparsePoint) -ne 0) { exit 32 }
if (($attributes -band [IO.FileAttributes]::Directory) -ne 0) { $pending.Push($entry) }
}
}
Remove-Item -LiteralPath $root -Recurse -Force
$rootEntries = @([IO.Directory]::EnumerateFileSystemEntries($parent, $name, [IO.SearchOption]::TopDirectoryOnly))
if ($rootEntries.Count -ne 0) { exit 33 }
`;
export function isExactGateProfile(
profile: string,
systemTemp: string,
): boolean {
const candidate = resolve(profile);
const temp = resolve(systemTemp);
const same = process.platform === "win32"
? dirname(candidate).toLowerCase() === temp.toLowerCase()
: dirname(candidate) === temp;
return same &&
/^screener-(?:native-one-viewer|access-privacy)-[A-Za-z0-9_-]{6}$/.test(
basename(candidate),
);
}
async function removeGateProfile(profile: string): Promise<boolean> {
if (!isExactGateProfile(profile, tmpdir())) return false;
if (process.platform !== "win32") {
const pending = [profile];
try {
while (pending.length > 0) {
const current = pending.pop()!;
const stat = await lstat(current);
if (stat.isSymbolicLink()) return false;
if (stat.isDirectory()) {
pending.push(
...(await readdir(current)).map((name) => join(current, name)),
);
}
}
await rm(profile, { recursive: true });
return true;
} catch {
return false;
}
}
return runBounded(
"powershell.exe",
["-NoProfile", "-NonInteractive", "-Command", profileCleanupScript],
10_000,
{ ...process.env, SCREENER_GATE_PROFILE_TO_REMOVE: profile },
);
}
function delay(ms: number): Promise<void> {
return new Promise((resolveDelay) => setTimeout(resolveDelay, ms));
}
+4 -5
View File
@@ -6,11 +6,10 @@ import { dirname, extname, relative, resolve, sep } from "node:path";
import { fileURLToPath } from "node:url";
const root = resolve(dirname(fileURLToPath(import.meta.url)), "..");
const repositoryFiles = execFileSync(
"git",
["ls-files", "--cached", "--others", "--exclude-standard"],
{ cwd: root, encoding: "utf8" },
)
const repositoryFiles = execFileSync("git", ["ls-files", "--cached"], {
cwd: root,
encoding: "utf8",
})
.trim()
.split(/\r?\n/u)
.filter((file) => file && existsSync(resolve(root, file)));
+1 -74
View File
@@ -1,4 +1,4 @@
import { basename, dirname, resolve } from "node:path";
import type { CleanupResult } from "./browser-gate-harness";
export interface ViewerIdentityEvidence {
socketCount: number;
@@ -50,14 +50,6 @@ export interface OneViewerIdentity {
pionEdgeGeneration: number;
}
export interface CleanupResult {
browserExited: boolean;
nativeExited: boolean;
serverClosed: boolean;
portsClosed: boolean;
profileRemoved: boolean;
}
interface FinalizableReport {
status: "passed" | "failed";
failedStage: string | null;
@@ -130,62 +122,6 @@ export async function verifyFinalSenderEvidence<T extends CriticalSenderEvidence
}
}
export function withDeadline<T>(
operation: () => Promise<T>,
deadline: number,
now: () => number = Date.now,
): Promise<T> {
const remaining = Math.max(0, deadline - now());
return new Promise<T>((resolveValue, rejectValue) => {
const timeout = setTimeout(() => rejectValue(new Error("Operation deadline exceeded")), remaining);
Promise.resolve().then(operation).then(
(value) => { clearTimeout(timeout); resolveValue(value); },
(error: unknown) => { clearTimeout(timeout); rejectValue(error); },
);
});
}
type JsonFetch = (
input: string,
init: { signal: AbortSignal },
) => Promise<{ ok: boolean; json(): Promise<unknown> }>;
export async function fetchJsonBefore<T>(
url: string,
deadline: number,
fetcher: JsonFetch = fetch,
): Promise<T> {
const controller = new AbortController();
const abort = setTimeout(() => controller.abort(), Math.max(0, deadline - Date.now()));
try {
const response = await withDeadline(
() => fetcher(url, { signal: controller.signal }),
deadline,
);
if (!response.ok) throw new Error("HTTP response was not successful");
return await withDeadline(() => response.json() as Promise<T>, deadline);
} finally {
clearTimeout(abort);
controller.abort();
}
}
export async function waitForSample<T>(
sample: (deadline: number) => Promise<T>,
retainAndAccept: (value: T) => boolean,
timeoutMs: number,
pollMs = 100,
): Promise<void> {
const deadline = Date.now() + timeoutMs;
while (Date.now() < deadline) {
let value: T | undefined;
try { value = await withDeadline(() => sample(deadline), deadline); } catch {}
if (value !== undefined && retainAndAccept(value)) return;
await new Promise((resolveDelay) => setTimeout(resolveDelay, Math.min(pollMs, Math.max(0, deadline - Date.now()))));
}
throw new Error("Bounded browser stage timed out");
}
export async function finalizeGate(
report: FinalizableReport,
gateSucceeded: boolean,
@@ -208,12 +144,3 @@ export async function finalizeGate(
report.failedStage = "cleanup";
}
}
export function isExactGateProfile(profile: string, systemTemp: string): boolean {
const candidate = resolve(profile);
const temp = resolve(systemTemp);
const same = process.platform === "win32"
? dirname(candidate).toLowerCase() === temp.toLowerCase()
: dirname(candidate) === temp;
return same && /^screener-(?:native-one-viewer|access-privacy)-[A-Za-z0-9_-]{6}$/.test(basename(candidate));
}
+11 -311
View File
@@ -1,13 +1,11 @@
import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process";
import { randomBytes } from "node:crypto";
import { closeSync, fsyncSync, openSync, writeFileSync } from "node:fs";
import { lstat, mkdtemp, readdir, rm } from "node:fs/promises";
import { createConnection, createServer } from "node:net";
import { mkdtemp } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join, resolve } from "node:path";
import { pathToFileURL } from "node:url";
import WebSocket from "ws";
import { createScreenerServer, type ScreenerServer } from "../src/server/app";
import { loadConfig } from "../src/server/config";
import {
@@ -22,19 +20,24 @@ import {
waitForSenderStart,
type SenderStartObservation,
} from "./native-one-viewer-ledger";
import {
CdpConnection,
cleanupRun,
createPage,
evaluate,
reservePort,
waitForSample,
waitForVersion,
type PageHandle,
} from "./browser-gate-harness";
import {
captureOneViewerIdentity,
fetchJsonBefore,
finalizeGate,
hasNoCriticalSenderErrors,
hasOneViewerPionProof,
isExactGateProfile,
retainsFirstBridgeSend,
retainsOneViewerIdentity,
waitForSample,
verifyFinalSenderEvidence,
withDeadline,
type CleanupResult,
type OneViewerIdentity,
type SenderIdentityEvidence,
type ViewerIdentityEvidence,
@@ -42,15 +45,6 @@ import {
type Stage = "preflight" | "sender-start" | "node-host" | "viewer-signal" | "viewer-media" | "cleanup";
export interface PageHandle {
sessionId: string;
}
interface CdpResult<T> {
result: { value?: T };
exceptionDetails?: unknown;
}
interface GateReport {
schemaVersion: 1;
status: "passed" | "failed";
@@ -137,82 +131,6 @@ interface ViewerSnapshot {
audioBytesReceived: number;
}
export class CdpConnection {
private nextId = 1;
private readonly pending = new Map<number, {
resolve: (value: unknown) => void;
reject: (error: Error) => void;
timeout: NodeJS.Timeout;
}>();
private constructor(private readonly socket: WebSocket) {
socket.on("message", (raw) => this.onMessage(raw.toString()));
socket.on("close", () => this.closePending());
socket.on("error", () => this.closePending());
}
static async connect(url: string, deadline: number): Promise<CdpConnection> {
const socket = new WebSocket(url, { maxPayload: 16 * 1024 * 1024 });
try {
await withDeadline(() => new Promise<void>((resolveOpen, rejectOpen) => {
socket.once("open", resolveOpen);
socket.once("error", rejectOpen);
}), deadline);
} catch (error) {
socket.terminate();
throw error;
}
return new CdpConnection(socket);
}
call<T>(method: string, params: Record<string, unknown>, sessionId: string | undefined, deadline: number): Promise<T> {
const id = this.nextId++;
return new Promise<T>((resolveCall, rejectCall) => {
const remaining = Math.max(0, deadline - Date.now());
const timeout = setTimeout(() => {
this.pending.delete(id);
rejectCall(new Error("CDP command timed out"));
}, remaining);
this.pending.set(id, {
resolve: (value) => { clearTimeout(timeout); resolveCall(value as T); },
reject: (error) => { clearTimeout(timeout); rejectCall(error); },
timeout,
});
this.socket.send(JSON.stringify({ id, method, params, sessionId }));
});
}
close(): void {
this.socket.close();
}
private onMessage(raw: string): void {
let message: { id?: number; result?: unknown; error?: unknown };
try {
message = JSON.parse(raw) as typeof message;
} catch {
return;
}
if (message.id === undefined) return;
const pending = this.pending.get(message.id);
if (!pending) return;
this.pending.delete(message.id);
if (message.error) {
pending.reject(new Error("CDP command failed"));
} else {
pending.resolve(message.result ?? {});
}
}
private closePending(): void {
for (const pending of this.pending.values()) {
clearTimeout(pending.timeout);
pending.reject(new Error("CDP connection closed"));
}
this.pending.clear();
}
}
const root = resolve(import.meta.dirname, "..");
class ObservedRoomStore extends RoomStore {
@@ -942,44 +860,6 @@ function viewerProbe(): string {
})();`;
}
export async function createPage(
cdp: CdpConnection,
url: string,
probe?: string,
newWindow = false,
browserContextId?: string,
): Promise<PageHandle> {
const deadline = Date.now() + 15_000;
const created = await cdp.call<{ targetId: string }>(
"Target.createTarget", {
url: "about:blank", background: false, newWindow,
...(browserContextId ? { browserContextId } : {}),
}, undefined, deadline,
);
const attached = await cdp.call<{ sessionId: string }>("Target.attachToTarget", { targetId: created.targetId, flatten: true }, undefined, deadline);
const page = { sessionId: attached.sessionId };
await Promise.all([
cdp.call("Page.enable", {}, page.sessionId, deadline),
cdp.call("Runtime.enable", {}, page.sessionId, deadline),
]);
if (probe) await cdp.call("Page.addScriptToEvaluateOnNewDocument", { source: probe }, page.sessionId, deadline);
await cdp.call("Page.navigate", { url }, page.sessionId, deadline);
await waitForSample(
(sampleDeadline) => evaluate(cdp, page, "document.readyState === 'complete'", sampleDeadline),
Boolean,
Math.max(1, deadline - Date.now()),
);
return page;
}
export async function evaluate<T>(cdp: CdpConnection, page: PageHandle, expression: string, deadline: number): Promise<T> {
const value = await cdp.call<CdpResult<T>>("Runtime.evaluate", {
expression, awaitPromise: true, returnByValue: true, userGesture: true,
}, page.sessionId, deadline);
if (value.exceptionDetails) throw new Error("Browser evaluation failed");
return value.result.value as T;
}
function senderSnapshot(cdp: CdpConnection, page: PageHandle, deadline: number): Promise<SenderSnapshot> {
return evaluate(cdp, page, "globalThis.__NATIVE_ONE_VIEWER_GATE__.snapshot()", deadline);
}
@@ -1003,18 +883,6 @@ function safeRoute(value: string | undefined): string {
.has(value ?? "") ? value! : "unknown";
}
export async function reservePort(): Promise<number> {
const server = createServer();
return new Promise((resolvePort, rejectPort) => {
server.once("error", rejectPort);
server.listen(0, "127.0.0.1", () => {
const address = server.address();
if (!address || typeof address === "string") return rejectPort(new Error("Port reservation failed"));
server.close((error) => error ? rejectPort(error) : resolvePort(address.port));
});
});
}
async function readLaunchUrl(child: ChildProcessWithoutNullStreams): Promise<string> {
let buffer = "";
return new Promise((resolveUrl, rejectUrl) => {
@@ -1039,170 +907,6 @@ async function readLaunchUrl(child: ChildProcessWithoutNullStreams): Promise<str
});
}
export async function waitForVersion(port: number, child: ChildProcessWithoutNullStreams): Promise<{ Browser: string; webSocketDebuggerUrl: string }> {
const deadline = Date.now() + 20_000;
while (Date.now() < deadline && child.exitCode === null) {
try {
return await fetchJsonBefore<{ Browser: string; webSocketDebuggerUrl: string }>(
`http://127.0.0.1:${port}/json/version`, deadline,
);
} catch {}
await delay(Math.min(100, Math.max(0, deadline - Date.now())));
}
throw new Error("Chrome startup timed out");
}
export interface CleanupResources {
cdp: CdpConnection | null;
native: ChildProcessWithoutNullStreams | null;
chrome: ChildProcessWithoutNullStreams | null;
server: ScreenerServer | null;
profile: string | null;
ports: number[];
}
export async function cleanupRun(resources: CleanupResources): Promise<CleanupResult> {
if (resources.cdp) {
try { await resources.cdp.call("Browser.close", {}, undefined, Date.now() + 2_000); } catch {}
resources.cdp.close();
}
const browserExited = resources.chrome ? await stopProcessTree(resources.chrome) : true;
if (resources.native && processRunning(resources.native)) {
try { resources.native.stdin.write("\n"); } catch {}
}
const nativeExited = resources.native ? await stopProcessTree(resources.native) : true;
let serverClosed = true;
if (resources.server) {
try { await withDeadline(() => resources.server!.close(), Date.now() + 5_000); } catch { serverClosed = false; }
}
const portsClosed = await waitForPortsClosed([...new Set(resources.ports)], 5_000);
const profileRemoved = resources.profile ? await removeGateProfile(resources.profile) : true;
return { browserExited, nativeExited, serverClosed, portsClosed, profileRemoved };
}
function processRunning(child: ChildProcessWithoutNullStreams): boolean {
return child.exitCode === null && child.signalCode === null;
}
async function stopProcessTree(child: ChildProcessWithoutNullStreams): Promise<boolean> {
if (!processRunning(child)) return true;
if (await waitForExit(child, 2_000)) return true;
if (process.platform === "win32" && child.pid !== undefined) {
await runBounded("taskkill.exe", ["/PID", String(child.pid), "/T", "/F"], 5_000);
} else {
child.kill("SIGTERM");
if (await waitForExit(child, 2_000)) return true;
child.kill("SIGKILL");
}
return waitForExit(child, 3_000);
}
async function waitForExit(child: ChildProcessWithoutNullStreams, timeoutMs: number): Promise<boolean> {
if (!processRunning(child)) return true;
return new Promise<boolean>((resolveExit) => {
const onExit = () => { clearTimeout(timeout); resolveExit(true); };
const timeout = setTimeout(() => {
child.off("exit", onExit);
resolveExit(!processRunning(child));
}, timeoutMs);
child.once("exit", onExit);
});
}
async function runBounded(command: string, args: string[], timeoutMs: number,
environment: NodeJS.ProcessEnv = process.env): Promise<boolean> {
const child = spawn(command, args, { stdio: "pipe", windowsHide: true, env: environment });
child.stdout.resume();
child.stderr.resume();
try {
const code = await withDeadline(() => new Promise<number | null>((resolveExit, rejectExit) => {
child.once("error", rejectExit);
child.once("exit", resolveExit);
}), Date.now() + timeoutMs);
return code === 0;
} catch {
child.kill();
return false;
}
}
async function waitForPortsClosed(ports: number[], timeoutMs: number): Promise<boolean> {
if (ports.length === 0) return true;
try {
await waitForSample(
(deadline) => Promise.all(ports.map((port) => portClosed(port, deadline))),
(closed) => closed.every(Boolean),
timeoutMs,
);
return true;
} catch {
return false;
}
}
function portClosed(port: number, deadline: number): Promise<boolean> {
return withDeadline(() => new Promise<boolean>((resolveClosed) => {
const socket = createConnection({ host: "127.0.0.1", port });
const finish = (closed: boolean) => { socket.destroy(); resolveClosed(closed); };
socket.once("connect", () => finish(false));
socket.once("error", () => finish(true));
}), deadline);
}
export const profileCleanupScript = String.raw`
$ErrorActionPreference = 'Stop'
$root = [IO.Path]::GetFullPath($env:SCREENER_GATE_PROFILE_TO_REMOVE)
$temp = [IO.Path]::GetFullPath([IO.Path]::GetTempPath()).TrimEnd([IO.Path]::DirectorySeparatorChar)
$parent = [IO.Path]::GetDirectoryName($root).TrimEnd([IO.Path]::DirectorySeparatorChar)
$name = [IO.Path]::GetFileName($root)
if ($parent -ine $temp -or $name -notmatch '^screener-(native-one-viewer|access-privacy)-[A-Za-z0-9_-]{6}$') { exit 31 }
$rootEntries = @([IO.Directory]::EnumerateFileSystemEntries($parent, $name, [IO.SearchOption]::TopDirectoryOnly))
if ($rootEntries.Count -eq 0) { exit 0 }
if ($rootEntries.Count -ne 1) { exit 34 }
$root = $rootEntries[0]
$rootAttributes = [IO.File]::GetAttributes($root)
if (($rootAttributes -band [IO.FileAttributes]::ReparsePoint) -ne 0) { exit 32 }
if (($rootAttributes -band [IO.FileAttributes]::Directory) -eq 0) { exit 34 }
$pending = [Collections.Generic.Stack[string]]::new()
$pending.Push($root)
while ($pending.Count -gt 0) {
$directory = $pending.Pop()
$attributes = [IO.File]::GetAttributes($directory)
if (($attributes -band [IO.FileAttributes]::ReparsePoint) -ne 0) { exit 32 }
foreach ($entry in [IO.Directory]::EnumerateFileSystemEntries($directory)) {
$attributes = [IO.File]::GetAttributes($entry)
if (($attributes -band [IO.FileAttributes]::ReparsePoint) -ne 0) { exit 32 }
if (($attributes -band [IO.FileAttributes]::Directory) -ne 0) { $pending.Push($entry) }
}
}
Remove-Item -LiteralPath $root -Recurse -Force
$rootEntries = @([IO.Directory]::EnumerateFileSystemEntries($parent, $name, [IO.SearchOption]::TopDirectoryOnly))
if ($rootEntries.Count -ne 0) { exit 33 }
`;
async function removeGateProfile(profile: string): Promise<boolean> {
if (!isExactGateProfile(profile, tmpdir())) return false;
if (process.platform !== "win32") {
const pending = [profile];
try {
while (pending.length > 0) {
const current = pending.pop()!;
const stat = await lstat(current);
if (stat.isSymbolicLink()) return false;
if (stat.isDirectory()) {
pending.push(...(await readdir(current)).map((name) => join(current, name)));
}
}
await rm(profile, { recursive: true });
return true;
} catch {
return false;
}
}
return runBounded("powershell.exe", ["-NoProfile", "-NonInteractive", "-Command", profileCleanupScript],
10_000, { ...process.env, SCREENER_GATE_PROFILE_TO_REMOVE: profile });
}
export function animatedSourceUrl(): string {
const html = `<!doctype html><title>Native Gate Source</title><canvas width="1280" height="720"></canvas>
<script>const a=new AudioContext(),o=a.createOscillator(),g=a.createGain();g.gain.value=.02;
@@ -1213,10 +917,6 @@ export function animatedSourceUrl(): string {
return `data:text/html;charset=utf-8,${encodeURIComponent(html)}`;
}
function delay(ms: number): Promise<void> {
return new Promise((resolveDelay) => setTimeout(resolveDelay, ms));
}
if (process.argv[1] && import.meta.url === pathToFileURL(resolve(process.argv[1])).href) {
await main();
}
+50 -11
View File
@@ -287,27 +287,45 @@ class BoundedLog {
}
}
class CdpConnection {
export class CdpConnection {
private nextId = 1;
private readonly pending = new Map<
number,
{ resolve: (value: unknown) => void; reject: (error: Error) => void }
{
resolve: (value: unknown) => void;
reject: (error: Error) => void;
timeout: NodeJS.Timeout;
}
>();
private closed = false;
private constructor(private readonly socket: WebSocket) {
private constructor(
private readonly socket: WebSocket,
private readonly commandTimeoutMs: number,
) {
socket.on("message", (raw) => this.onMessage(raw.toString()));
socket.on("close", () => this.onClose(new Error("CDP connection closed")));
socket.on("error", (error) => this.onClose(error));
}
static async connect(url: string): Promise<CdpConnection> {
static async connect(
url: string,
commandTimeoutMs: number,
): Promise<CdpConnection> {
const socket = new WebSocket(url, { maxPayload: 64 * 1024 * 1024 });
await new Promise<void>((resolveOpen, rejectOpen) => {
socket.once("open", resolveOpen);
socket.once("error", rejectOpen);
const timeout = setTimeout(() => {
socket.terminate();
rejectOpen(new Error("CDP connection timed out"));
}, commandTimeoutMs);
const finish = (callback: () => void) => {
clearTimeout(timeout);
callback();
};
socket.once("open", () => finish(resolveOpen));
socket.once("error", (error) => finish(() => rejectOpen(error)));
});
return new CdpConnection(socket);
return new CdpConnection(socket, commandTimeoutMs);
}
call<T = Record<string, unknown>>(
@@ -320,11 +338,28 @@ class CdpConnection {
}
const id = this.nextId++;
return new Promise<T>((resolveCall, rejectCall) => {
const timeout = setTimeout(() => {
this.pending.delete(id);
rejectCall(new Error(`CDP command timed out: ${method}`));
}, this.commandTimeoutMs);
this.pending.set(id, {
resolve: (value) => resolveCall(value as T),
reject: rejectCall,
resolve: (value) => {
clearTimeout(timeout);
resolveCall(value as T);
},
reject: (error) => {
clearTimeout(timeout);
rejectCall(error);
},
timeout,
});
this.socket.send(JSON.stringify({ id, method, params, sessionId }));
try {
this.socket.send(JSON.stringify({ id, method, params, sessionId }));
} catch (error) {
clearTimeout(timeout);
this.pending.delete(id);
rejectCall(error instanceof Error ? error : new Error("CDP send failed"));
}
});
}
@@ -372,6 +407,7 @@ class CdpConnection {
}
this.closed = true;
for (const pending of this.pending.values()) {
clearTimeout(pending.timeout);
pending.reject(error);
}
this.pending.clear();
@@ -3075,7 +3111,10 @@ export async function main(): Promise<number> {
webSocketDebuggerUrl: string;
[key: string]: unknown;
};
cdp = await CdpConnection.connect(version.webSocketDebuggerUrl);
cdp = await CdpConnection.connect(
version.webSocketDebuggerUrl,
config.connectionTimeoutMs,
);
report.chromium = await cdp.call("Browser.getVersion");
const viewerCounts = config.canaryMode === "viewer-mbb" ? [3] : config.viewerCounts;
+7 -6
View File
@@ -1,11 +1,9 @@
import {
PERSISTENT_NATIVE_EDGE_DEGRADED_WINDOWS,
VIEWER_QUALITY_EVIDENCE_EXPIRY_MS,
} from "../../shared/protocol";
import { PERSISTENT_NATIVE_EDGE_DEGRADED_WINDOWS } from "../../shared/protocol";
import type { ConnectionMetrics } from "../types";
interface DeliveredVideoSample {
timestampMs: number;
windowMs: number;
pixels: number;
framesPerSecond: number;
bitrateKbps: number | null;
@@ -46,6 +44,7 @@ function deliveredVideoSample(
}
return {
timestampMs: sampleTimestampMs,
windowMs: sampleWindowMs,
pixels: frameWidth * frameHeight,
framesPerSecond: Math.round(framesPerSecond),
bitrateKbps:
@@ -66,8 +65,10 @@ function comparableSamples(
if (
!currentVideo ||
!candidateVideo ||
Math.abs(candidateVideo.timestampMs - currentVideo.timestampMs) >
VIEWER_QUALITY_EVIDENCE_EXPIRY_MS ||
Math.max(
currentVideo.timestampMs - currentVideo.windowMs,
candidateVideo.timestampMs - candidateVideo.windowMs,
) >= Math.min(currentVideo.timestampMs, candidateVideo.timestampMs) ||
candidate.intervalFreezeCount === null ||
candidate.intervalFreezeDurationMs === null ||
candidate.intervalPauseCount === null ||
+1 -61
View File
@@ -11,11 +11,7 @@ import {
} from "../sfu/publisher";
import type { ConnectionMetrics } from "../types";
import type { BrowserVideoCodec } from "../webrtc/video-codec";
import type {
AudioSenderParameterReadback,
QualityProfile,
VideoSenderParameterReadback,
} from "./quality";
import type { QualityProfile } from "./quality";
import { qualitySettingsEqual } from "./quality";
import {
MediaRouteTransition,
@@ -36,8 +32,6 @@ interface HostPublisherTransport {
updateProfile(profile: QualityProfile): Promise<boolean>;
getQualityWarning?(): string | null;
getFailureStage?(): SfuPublisherFailureStage | null;
getSenderParameters?(): VideoSenderParameterReadback | null;
getAudioSenderParameters?(): AudioSenderParameterReadback | null;
disconnect(): Promise<void>;
}
@@ -50,12 +44,6 @@ interface HostPublisherSlot {
failed: boolean;
}
export interface HostSfuPublisherSnapshot {
metrics: ConnectionMetrics;
senderParameters: VideoSenderParameterReadback | null;
audioSenderParameters: AudioSenderParameterReadback | null;
}
interface HostSfuRouteEvents {
getStream: () => MediaStream | null;
getProfile: () => QualityProfile;
@@ -67,7 +55,6 @@ interface HostSfuRouteEvents {
revision: number,
publicationGeneration: string,
) => void;
onPublisherUpdate?: (snapshot: HostSfuPublisherSnapshot | null) => void;
createPublisher?: (
onDisconnected: () => void,
onStats: (metrics: ConnectionMetrics | null) => void,
@@ -171,7 +158,6 @@ export class HostSfuRoute {
const active = this.active;
this.pending = null;
this.active = null;
this.events.onPublisherUpdate?.(null);
if (pending) {
pending.failed = true;
}
@@ -332,12 +318,6 @@ export class HostSfuRoute {
: null;
}
getSenderParameters(): VideoSenderParameterReadback | null {
return this.active?.active
? (this.active.publisher.getSenderParameters?.() ?? null)
: null;
}
replaceStream(stream: MediaStream): Promise<boolean> {
const slots = this.publishingSlots();
return slots.length === 0
@@ -362,7 +342,6 @@ export class HostSfuRoute {
if (!slot) {
return;
}
this.events.onPublisherUpdate?.(null);
this.active = null;
await disconnectPublisher(slot.publisher);
this.handleFailure(slot);
@@ -383,7 +362,6 @@ export class HostSfuRoute {
const active = this.active;
this.pending = null;
this.active = null;
this.events.onPublisherUpdate?.(null);
if (active?.active) {
await active.publisher.deactivate().catch(() => false);
}
@@ -543,22 +521,6 @@ export class HostSfuRoute {
);
}
private currentActiveToken(
slot: HostPublisherSlot,
): RouteOperationToken | null {
const token = this.route.token();
const assignment = this.route.getActiveAssignment();
if (
this.active !== slot ||
!token ||
this.route.getPhase() !== "active" ||
assignment?.sfuPublicationGeneration !== slot.publicationGeneration
) {
return null;
}
return token;
}
private commitMedia(token: RouteOperationToken): void {
if (!this.route.markMediaActive(token)) {
return;
@@ -581,7 +543,6 @@ export class HostSfuRoute {
return;
}
const active = this.active;
this.events.onPublisherUpdate?.(null);
this.active = null;
await this.disconnectRetiredPublisher(active);
}
@@ -597,7 +558,6 @@ export class HostSfuRoute {
}
const active = this.active;
this.active = null;
this.events.onPublisherUpdate?.(null);
return active;
}
@@ -631,7 +591,6 @@ export class HostSfuRoute {
this.pending = null;
}
if (wasActive) {
this.events.onPublisherUpdate?.(null);
this.active = null;
}
@@ -694,25 +653,6 @@ export class HostSfuRoute {
slot.publicationGeneration,
);
}
if (
metrics === null ||
this.closed ||
slot.failed ||
!slot.active ||
this.active !== slot ||
!this.currentActiveToken(slot)
) {
if (metrics === null && this.active === slot) {
this.events.onPublisherUpdate?.(null);
}
return;
}
this.events.onPublisherUpdate?.({
metrics: { ...metrics },
senderParameters: slot.publisher.getSenderParameters?.() ?? null,
audioSenderParameters:
slot.publisher.getAudioSenderParameters?.() ?? null,
});
}
}
-10
View File
@@ -1535,16 +1535,6 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
}
return;
}
if (message.type === "media-assignment") {
if (peerAssistedRef.current && !hostSfuRouteRef.current) {
discardPreparedHostChild();
reconcileHostChildren(
message.mediaAssignment.childPeerIds,
generation,
);
}
return;
}
if (message.type === "peer-joined") {
if (peerAssistedRef.current) {
return;
+1 -7
View File
@@ -19,7 +19,6 @@ import {
MAX_VIEWER_PASSWORD_LENGTH,
viewerPasswordSchema,
type IceConfig,
type MediaAssignment,
type ParticipantPresenceEntry,
type ParticipantRouteAssignment,
type ServerMessage,
@@ -90,6 +89,7 @@ import type {
import {
limitMediaAssignment,
MAX_ENDPOINT_MEDIA_CHILDREN,
type MediaAssignment,
viewerRestartMessage,
viewerSignalMessage,
} from "../webrtc/media-assignment";
@@ -1582,12 +1582,6 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) {
}
return;
}
if (message.type === "media-assignment") {
if (peerAssisted && !viewerSfuRoute) {
applyMediaAssignment(message.mediaAssignment);
}
return;
}
if (message.type === "quality-settings") {
if (peerAssisted) {
currentQualitySettings = message.qualitySettings;
+5 -1
View File
@@ -1,10 +1,14 @@
import type {
ClientMessage,
MediaAssignment,
SignalPayload,
} from "../../shared/protocol";
import { MAX_ENDPOINT_MEDIA_COPY_CAPACITY } from "../../shared/media-copy-accounting";
export interface MediaAssignment {
parentPeerId: string | null;
childPeerIds: string[];
}
export const MAX_ENDPOINT_MEDIA_CHILDREN = MAX_ENDPOINT_MEDIA_COPY_CAPACITY;
export function limitMediaAssignment(
+1 -19
View File
@@ -3,7 +3,6 @@ import { debuglog } from "node:util";
import type {
ClientMessage,
MediaAssignment,
MediaRouteUpstream,
ParticipantRouteAssignment,
PreparedRouteCandidate,
@@ -148,7 +147,6 @@ export interface HybridMediaRouterOptions {
export interface HybridAuthenticationState {
routeRevision: number;
routeAssignment: ParticipantRouteAssignment;
mediaAssignment: MediaAssignment;
}
export interface AuthenticatedRouteParticipant {
@@ -232,7 +230,6 @@ export class HybridMediaRouter {
return {
routeRevision: snapshot?.revision ?? 0,
routeAssignment: assignment,
mediaAssignment: mediaAssignment(assignment),
};
}
@@ -935,6 +932,7 @@ export class HybridMediaRouter {
removeViewer(roomId: string, peerId: string): void {
const room = this.rooms.get(roomId);
room?.advertisedCapacityByViewer.delete(peerId);
room?.debugOrdinalByPeerId.delete(peerId);
this.options.deleteConnectionId(roomId, peerId);
if (room?.controller?.confirmDeparture(peerId, this.now())) {
this.requestPump(roomId);
@@ -1257,7 +1255,6 @@ export class HybridMediaRouter {
peerId: operation.childPeerId,
shareGeneration,
publicationGeneration: publication.generation,
allowlistedRootPeerIds: [operation.childPeerId],
});
return {
kind: "ready",
@@ -1335,7 +1332,6 @@ export class HybridMediaRouter {
peerId: host.peerId,
shareGeneration,
publicationGeneration,
allowlistedRootPeerIds: [operation.childPeerId],
}),
fallback.tokenIssuer.issueToken({
roomId,
@@ -1343,7 +1339,6 @@ export class HybridMediaRouter {
peerId: operation.childPeerId,
shareGeneration,
publicationGeneration,
allowlistedRootPeerIds: [operation.childPeerId],
}),
]);
return {
@@ -1773,11 +1768,6 @@ export class HybridMediaRouter {
peerId: participant.peerId,
shareGeneration,
publicationGeneration: publication.generation,
allowlistedRootPeerIds: isHost
? [...snapshot.upstreamByViewer]
.filter(([, candidate]) => candidate.kind === "sfu" && candidate.physicalActive)
.map(([peerId]) => peerId)
: [participant.peerId],
});
const current = room.controller?.snapshot();
if (
@@ -2038,14 +2028,6 @@ function emptyAssignment(): ParticipantRouteAssignment {
};
}
function mediaAssignment(assignment: ParticipantRouteAssignment): MediaAssignment {
return {
parentPeerId:
assignment.upstream.kind === "peer" ? assignment.upstream.peerId : null,
childPeerIds: [...assignment.childPeerIds],
};
}
function preparedRouteCandidate(
operation: OperationSnapshot,
tuple: CandidateTuple,
+2 -22
View File
@@ -19,7 +19,6 @@ export interface SfuTokenRequest {
peerId: string;
shareGeneration: string;
publicationGeneration: string;
allowlistedRootPeerIds: readonly string[];
}
export interface SfuTokenIssuer {
@@ -47,7 +46,7 @@ export class LiveKitTokenIssuer implements SfuTokenIssuer {
}
async issueToken(request: SfuTokenRequest): Promise<string> {
validateTokenRequest(request, this.options.maxViewersPerRoom);
validateTokenRequest(request);
const isHost = request.role === "host";
const room = managedSfuRoomName(request);
@@ -74,10 +73,7 @@ export class LiveKitTokenIssuer implements SfuTokenIssuer {
}
}
function validateTokenRequest(
request: SfuTokenRequest,
maxViewersPerRoom: number,
): void {
function validateTokenRequest(request: SfuTokenRequest): void {
if (!ROOM_ID_PATTERN.test(request.roomId)) {
throw new Error("LiveKit room ID is invalid");
}
@@ -88,20 +84,4 @@ function validateTokenRequest(
) {
throw new Error("LiveKit participant identity is invalid");
}
if (
request.allowlistedRootPeerIds.length > maxViewersPerRoom ||
request.allowlistedRootPeerIds.some(
(peerId) => !OPAQUE_ID_PATTERN.test(peerId),
)
) {
throw new Error("LiveKit SFU root allowlist is invalid");
}
const roots = new Set(request.allowlistedRootPeerIds);
if (roots.size !== request.allowlistedRootPeerIds.length) {
throw new Error("LiveKit SFU root allowlist contains duplicates");
}
if (request.role === "viewer" && !roots.has(request.peerId)) {
throw new Error("LiveKit viewer is not an allowlisted SFU root");
}
}
-4
View File
@@ -9,7 +9,6 @@ import type {
const MANAGED_ROOM_PREFIX = "screener-v1.";
const ROOM_ID_PATTERN = /^[1-9]\d{0,11}$/;
const OPAQUE_ID_PATTERN = /^[A-Za-z0-9_-]{8,128}$/;
const LEGACY_MANAGED_ROOM_PATTERN = /^screener-[1-9]\d{0,11}-[A-Za-z0-9_-]{8,128}$/;
const ROOM_SERVICE_TIMEOUT_SECONDS = 5;
interface LiveKitRoomRecord {
@@ -206,9 +205,6 @@ function managedSfuViewerIdentity(peerId: string): string {
}
export function isManagedSfuRoomName(roomName: string): boolean {
if (LEGACY_MANAGED_ROOM_PATTERN.test(roomName)) {
return true;
}
if (!roomName.startsWith(MANAGED_ROOM_PREFIX)) {
return false;
}
-1
View File
@@ -713,7 +713,6 @@ export class SignalingServer {
mediaMode: "peer-assisted",
shareGeneration:
this.shareGenerationsByRoom.get(participant.roomId) ?? null,
mediaAssignment: hybridState.mediaAssignment,
routeRevision: hybridState.routeRevision,
routeAssignment: hybridState.routeAssignment,
qualitySettings:
-17
View File
@@ -336,16 +336,6 @@ export type PreparedRouteCandidate = z.infer<
typeof preparedRouteCandidateSchema
>;
export const mediaAssignmentSchema = z
.object({
parentPeerId: opaqueIdSchema.nullable(),
childPeerIds: z
.array(opaqueIdSchema)
.max(MAX_ENDPOINT_MEDIA_COPY_CAPACITY),
})
.strict();
export type MediaAssignment = z.infer<typeof mediaAssignmentSchema>;
export const mediaRouteRevisionSchema = z
.number()
.int()
@@ -943,7 +933,6 @@ const authenticatedViewerMessageShape = {
const peerAssistedAuthenticatedShape = {
mediaMode: z.literal("peer-assisted"),
shareGeneration: opaqueIdSchema.nullable(),
mediaAssignment: mediaAssignmentSchema,
routeRevision: mediaRouteRevisionSchema,
routeAssignment: participantRouteAssignmentSchema,
qualitySettings: qualitySettingsSchema,
@@ -1009,12 +998,6 @@ export const serverMessageSchema = z.union([
rebuild: z.boolean(),
})
.strict(),
z
.object({
type: z.literal("media-assignment"),
mediaAssignment: mediaAssignmentSchema,
})
.strict(),
z.discriminatedUnion("phase", [
z
.object({
+44
View File
@@ -116,6 +116,50 @@ describe("SFU quality probe", () => {
).toBe(false);
});
it("keeps nearby but non-overlapping samples pending", () => {
const probe = new P2pQualityProbe();
const current = (timestampMs: number) =>
metrics(timestampMs, {
sampleWindowMs: 1_000,
frameWidth: 320,
frameHeight: 180,
framesPerSecond: 10,
});
const candidate = (timestampMs: number) =>
metrics(timestampMs, {
sampleWindowMs: 1_000,
frameWidth: 1_280,
frameHeight: 720,
framesPerSecond: 30,
});
expect(
p2pCandidateStrictlyImproves(current(10_000), candidate(11_100)),
).toBeNull();
for (const timestampMs of [10_000, 13_000, 16_000]) {
expect(
probe.observe(current(timestampMs), candidate(timestampMs + 1_100)),
).toBe("pending");
}
});
it("keeps the original comparison for overlapping samples", () => {
const current = metrics(10_000, {
sampleWindowMs: 1_000,
frameWidth: 320,
frameHeight: 180,
framesPerSecond: 10,
});
const candidate = metrics(10_900, {
sampleWindowMs: 1_000,
frameWidth: 1_280,
frameHeight: 720,
framesPerSecond: 30,
});
expect(p2pCandidateStrictlyImproves(current, candidate)).toBe(true);
});
it("requires three consecutive strict P2P improvements", () => {
const probe = new P2pQualityProbe();
const current = (timestampMs: number) =>
-2
View File
@@ -1058,7 +1058,6 @@ describe("client signaling recovery policy", () => {
viewerAuthorizationGeneration: "viewer_generation_12345678",
mediaMode: "peer-assisted",
shareGeneration,
mediaAssignment: { parentPeerId: null, childPeerIds: [] },
routeRevision: 1,
routeAssignment: {
upstream: { kind: "none" },
@@ -1312,7 +1311,6 @@ describe("client signaling recovery policy", () => {
viewerAuthorizationGeneration: "viewer_generation_12345678",
mediaMode: "peer-assisted",
shareGeneration: "share_generation_12345678",
mediaAssignment: { parentPeerId: null, childPeerIds: [] },
routeRevision: 1,
routeAssignment: {
upstream: { kind: "none" },
+1 -87
View File
@@ -8,8 +8,6 @@ const apiSecret = "s".repeat(32);
const publicationGeneration = "publication_12345678";
const shareGeneration = "share_generation_12345678";
const rootPeerId = "viewer_root_12345678";
const secondRootPeerId = "viewer_root_23456789";
const thirdRootPeerId = "viewer_root_34567890";
function issuer(maxViewersPerRoom = 8): LiveKitTokenIssuer {
return new LiveKitTokenIssuer({
@@ -20,44 +18,6 @@ function issuer(maxViewersPerRoom = 8): LiveKitTokenIssuer {
}
describe("LiveKitTokenIssuer", () => {
it("allows every admitted root within the room Viewer bound", async () => {
const tokenIssuer = issuer();
await expect(
tokenIssuer.issueToken({
roomId: "42",
role: "host",
peerId: "host_peer_12345678",
shareGeneration,
publicationGeneration,
allowlistedRootPeerIds: [rootPeerId, secondRootPeerId],
}),
).resolves.toEqual(expect.any(String));
await expect(
tokenIssuer.issueToken({
roomId: "42",
role: "viewer",
peerId: secondRootPeerId,
shareGeneration,
publicationGeneration,
allowlistedRootPeerIds: [rootPeerId, secondRootPeerId],
}),
).resolves.toEqual(expect.any(String));
await expect(
tokenIssuer.issueToken({
roomId: "42",
role: "host",
peerId: "host_peer_12345678",
shareGeneration,
publicationGeneration,
allowlistedRootPeerIds: [
rootPeerId,
secondRootPeerId,
thirdRootPeerId,
],
}),
).resolves.toEqual(expect.any(String));
});
it("issues a generation-bound host token limited to screen sharing", async () => {
const claims = await new TokenVerifier(apiKey, apiSecret).verify(
await issuer().issueToken({
@@ -66,7 +26,6 @@ describe("LiveKitTokenIssuer", () => {
peerId: "host_peer_12345678",
shareGeneration,
publicationGeneration,
allowlistedRootPeerIds: [rootPeerId],
}),
);
@@ -84,7 +43,7 @@ describe("LiveKitTokenIssuer", () => {
expect(claims.roomConfig).toBeUndefined();
});
it("issues a subscribe-only token to a current fallback root", async () => {
it("issues a generation-bound subscribe-only Viewer token", async () => {
const claims = await new TokenVerifier(apiKey, apiSecret).verify(
await issuer(3).issueToken({
roomId: "7",
@@ -92,7 +51,6 @@ describe("LiveKitTokenIssuer", () => {
peerId: rootPeerId,
shareGeneration,
publicationGeneration,
allowlistedRootPeerIds: [rootPeerId],
}),
);
@@ -109,49 +67,6 @@ describe("LiveKitTokenIssuer", () => {
expect(claims.roomConfig).toBeUndefined();
});
it("does not issue viewer tokens outside the current root allowlist", async () => {
await expect(
issuer().issueToken({
roomId: "7",
role: "viewer",
peerId: "viewer_other_12345678",
shareGeneration,
publicationGeneration,
allowlistedRootPeerIds: [rootPeerId],
}),
).rejects.toThrow("viewer is not an allowlisted SFU root");
});
it("rejects duplicate, excessive, and malformed root allowlists", async () => {
const tokenIssuer = issuer(1);
const request = {
roomId: "7",
role: "host" as const,
peerId: "host_peer_12345678",
shareGeneration,
publicationGeneration,
};
await expect(
tokenIssuer.issueToken({
...request,
allowlistedRootPeerIds: [rootPeerId, rootPeerId],
}),
).rejects.toThrow("root allowlist is invalid");
await expect(
issuer().issueToken({
...request,
allowlistedRootPeerIds: [rootPeerId, rootPeerId],
}),
).rejects.toThrow("root allowlist contains duplicates");
await expect(
issuer().issueToken({
...request,
allowlistedRootPeerIds: ["not valid"],
}),
).rejects.toThrow("root allowlist is invalid");
});
it("isolates publication generations in separate LiveKit rooms", async () => {
const tokenIssuer = issuer();
const verifier = new TokenVerifier(apiKey, apiSecret);
@@ -160,7 +75,6 @@ describe("LiveKitTokenIssuer", () => {
role: "host" as const,
peerId: "host_peer_12345678",
shareGeneration,
allowlistedRootPeerIds: [rootPeerId],
};
const first = await verifier.verify(
await tokenIssuer.issueToken({
+7 -5
View File
@@ -4,24 +4,26 @@ import { runInNewContext } from "node:vm";
import { describe, expect, it, vi } from "vitest";
import {
captureOneViewerIdentity,
fetchJsonBefore,
isExactGateProfile,
profileCleanupScript,
waitForSample,
type CleanupResult,
} from "../scripts/browser-gate-harness";
import {
captureOneViewerIdentity,
finalizeGate,
hasNoCriticalSenderErrors,
hasOneViewerPionProof,
isExactGateProfile,
retainsFirstBridgeSend,
retainsOneViewerIdentity,
waitForSample,
verifyFinalSenderEvidence,
type CleanupResult,
type SenderIdentityEvidence,
type ViewerIdentityEvidence,
} from "../scripts/native-one-viewer-gate-core";
import {
animatedSourceUrl,
parseGateMode,
profileCleanupScript,
senderProbe,
} from "../scripts/native-one-viewer-gate";
+24 -7
View File
@@ -1,7 +1,9 @@
import { runInNewContext } from "node:vm";
import { describe, expect, it, vi } from "vitest";
import { WebSocketServer } from "ws";
import {
CdpConnection,
activeVideoEdgeCount,
buildBenchmarkFailureEvidence,
buildBenchmarkInitScript,
@@ -245,6 +247,28 @@ function markSfuMedia(observation: ReturnType<typeof page>): void {
}
describe("peer topology loopback configuration", () => {
it("bounds a CDP command that never receives a response", async () => {
const server = new WebSocketServer({ port: 0 });
await new Promise<void>((resolveListening) => {
server.once("listening", resolveListening);
});
const address = server.address();
if (!address || typeof address === "string") {
throw new Error("Unexpected WebSocket address");
}
const cdp = await CdpConnection.connect(
`ws://127.0.0.1:${address.port}`,
500,
);
await expect(cdp.call("Never.responds")).rejects.toThrow(
"CDP command timed out: Never.responds",
);
await cdp.close();
await new Promise<void>((resolveClose, rejectClose) => {
server.close((error) => error ? rejectClose(error) : resolveClose());
});
});
it("keeps the real-relay canary opt-in", () => {
expect(parseBenchmarkCanaryMode(undefined)).toBe("none");
expect(parseBenchmarkCanaryMode("viewer-mbb")).toBe("viewer-mbb");
@@ -1375,13 +1399,6 @@ describe("peer topology loopback observations", () => {
phase: "active",
assignment: activeRoute,
});
owner.emitMessage({
type: "media-assignment",
mediaAssignment: {
parentPeerId: null,
childPeerIds: ["legacy_child_01"],
},
});
expect(observer.snapshot()).toMatchObject({
routeRevision: 5,
routeAssignment: preparedRoute,
+2 -21
View File
@@ -1404,10 +1404,6 @@ describe("server signaling protocol", () => {
...authenticatedMessage(8),
mediaMode: "peer-assisted",
shareGeneration: "share_generation_12345678",
mediaAssignment: {
parentPeerId: null,
childPeerIds: ["viewer_12345678", "viewer_87654321"],
},
routeRevision: 0,
routeAssignment: {
upstream: { kind: "none" },
@@ -1425,12 +1421,6 @@ describe("server signaling protocol", () => {
sfuStandbyUrl: "wss://sfu.example.com",
}).success,
).toBe(true);
expect(
serverMessageSchema.safeParse({
type: "media-assignment",
mediaAssignment: peerAssisted.mediaAssignment,
}).success,
).toBe(true);
expect(
serverMessageSchema.safeParse({
...authenticatedMessage(8),
@@ -1440,20 +1430,11 @@ describe("server signaling protocol", () => {
qualitySettings,
}).success,
).toBe(false);
expect(
serverMessageSchema.safeParse({
...authenticatedMessage(8),
mediaMode: "peer-assisted",
mediaAssignment: peerAssisted.mediaAssignment,
routeRevision: 0,
routeAssignment: peerAssisted.routeAssignment,
}).success,
).toBe(false);
expect(
serverMessageSchema.safeParse({
...peerAssisted,
mediaAssignment: {
parentPeerId: null,
routeAssignment: {
...peerAssisted.routeAssignment,
childPeerIds: [
"viewer_12345678",
"viewer_87654321",
+2 -7
View File
@@ -103,19 +103,14 @@ function deferred(): { promise: Promise<void>; resolve: () => void } {
}
describe("LiveKitSfuRoomControl", () => {
it("drains current and legacy managed rooms before startup", async () => {
it("drains current managed rooms before startup", async () => {
const roomService = new FakeRoomService();
const current = managedSfuRoomName(fence());
const legacy = "screener-7-publication_12345678";
roomService.rooms.set(current, new Set(["host"]));
roomService.rooms.set(legacy, new Set(["host"]));
await expect(control(roomService).initialize()).resolves.toBeUndefined();
expect(roomService.rooms.size).toBe(0);
expect(roomService.operations).toEqual([
`delete:${current}`,
`delete:${legacy}`,
]);
expect(roomService.operations).toEqual([`delete:${current}`]);
});
it("rejects a shared instance and an unconfirmed startup drain", async () => {
+1 -1
View File
@@ -19,9 +19,9 @@
"include": [
"vite.config.ts",
"vitest.config.ts",
"scripts/access-privacy-gate.ts",
"scripts/native-one-viewer-gate.ts",
"scripts/native-one-viewer-ledger.ts",
"scripts/browser-codec-preflight.ts",
"tests"
]
}