diff --git a/docs/internal/DESIGN.md b/docs/internal/DESIGN.md index 2f7b0d8c6..6fde08476 100644 --- a/docs/internal/DESIGN.md +++ b/docs/internal/DESIGN.md @@ -162,7 +162,7 @@ put together from them and draws nothing of its own that one of them already dra device row opens the drawer. Choosing a lease, a device, a request or an event opens a **drawer**: a 480px panel (full width under 768px) at the right edge over the page, with a dim layer over everything else, the sidebar included. It has a title (a lease's holder, a device's model and runtime, a request's requester, an event's name), a 40px square close button - (`IconButton`) and, for a lease, tabs. Escape, the close button and the dim layer close it. It slides in once, when it opens, + (`IconButton`) and, for a lease or a device, tabs: a device's are Details and Screen, a lease's Details and Events. Escape, the close button and the dim layer close it. It slides in once, when it opens, and switching its tabs does not replay that. A lease's drawer uses no status colour of its own; a device's shows the device's state as a status, word and square. - **Menu.** A table row ends in a **⋯** `IconButton` that opens a small list; a right-click on the diff --git a/e2e/console/fixtures.ts b/e2e/console/fixtures.ts index d279368b6..e7d575a1a 100644 --- a/e2e/console/fixtures.ts +++ b/e2e/console/fixtures.ts @@ -435,15 +435,24 @@ export interface Fleet { * leases through the gateway or reads a card's capacity right away would see that gap. */ export async function startFleet( - workers: readonly { readonly label: string; readonly driverScript?: FakeDriverScript }[], + workers: readonly { + readonly label: string; + readonly driverScript?: FakeDriverScript; + readonly config?: Readonly>; + }[], + options: { readonly gatewayConfig?: Readonly> } = {}, ): Promise { - const gateway = await startDaemon("gateway"); + const gateway = await startDaemon( + "gateway", + options.gatewayConfig === undefined ? {} : { config: options.gatewayConfig }, + ); const started: RunningDaemon[] = []; try { for (const worker of workers) { started.push( await startDaemon("worker", { config: { + ...worker.config, gateway: { label: worker.label, token: gateway.tokens.worker, diff --git a/e2e/console/leases.spec.ts b/e2e/console/leases.spec.ts index 84303aa3a..236ab98c4 100644 --- a/e2e/console/leases.spec.ts +++ b/e2e/console/leases.spec.ts @@ -335,7 +335,7 @@ test.describe("the leases page", () => { await leaseRow(page, lease.id).click(); await expect(page.getByRole("dialog")).toBeVisible(); - for (const name of ["Screen", "Details", "Screen", "Details"]) { + for (const name of ["Events", "Details", "Events", "Details"]) { await page.getByRole("tab", { name }).click(); } await page.waitForTimeout(400); @@ -649,8 +649,7 @@ test.describe("the leases page", () => { await leaseRow(page, granted.id).click(); await expect(inspectorTitle(page, /ci-runner-3/)).toBeVisible(); await expect(fact(page, "Device", "UDID")).toHaveText(granted.udid); - await page.getByRole("tab", { name: "Screen" }).click(); - await expect(page.getByRole("button", { name: "Show screen" })).toBeVisible(); + await expect(page.getByRole("dialog").getByRole("tab")).toHaveText(["Details", "Events"]); expect(await hasHorizontalScroll(page)).toBe(false); } finally { await host.dispose(); diff --git a/e2e/console/screen.spec.ts b/e2e/console/screen.spec.ts index 746514f0b..5724042de 100644 --- a/e2e/console/screen.spec.ts +++ b/e2e/console/screen.spec.ts @@ -4,7 +4,15 @@ import { join } from "node:path"; import type { Page } from "@playwright/test"; -import { expect, type RunningDaemon, signIn, startDaemon, test } from "./fixtures.js"; +import { + disposeFleet, + expect, + type RunningDaemon, + signIn, + startDaemon, + startFleet, + test, +} from "./fixtures.js"; import { captureLog, writeCaptureProgram } from "./screen-capture.js"; // why: the capture program this lane writes is its witness as well as its picture source -- it records @@ -19,10 +27,11 @@ interface StreamingHost { readonly command: string; readonly dir: string; readonly host: RunningDaemon; - readonly lease: string; } -async function streamingHost(): Promise { +async function streamingHost( + extraConfig: Readonly> = {}, +): Promise { const dir = await mkdtemp(join(tmpdir(), "simlock-screen-")); const command = await writeCaptureProgram(dir); let host: RunningDaemon | undefined; @@ -30,17 +39,11 @@ async function streamingHost(): Promise { host = await startDaemon("worker", { config: { capabilities: { stream: { enabled: true, fps: 5 } }, + ...extraConfig, }, driverScript: { ios: { ...IOS, screenCaptureCommand: command } }, }); - const granted = await host.cli([...LEASE, "--agent-id", "console-screen", "--detach"]); - expect(granted.code, granted.stderr).toBe(0); - return { - command, - dir, - host, - lease: (granted.json as { lease: { id: string } }).lease.id, - }; + return { command, dir, host }; } catch (error) { await rm(dir, { force: true, recursive: true }); await host?.dispose(); @@ -53,19 +56,85 @@ async function cleanup(started: StreamingHost): Promise { await rm(started.dir, { force: true, recursive: true }); } -async function openLeasePage(page: Page, host: RunningDaemon, lease: string): Promise { - await page.goto(new URL(`/leases/${encodeURIComponent(lease)}`, host.url).toString()); - await signIn(page, host.tokens.operator); - await page.getByRole("tab", { name: "Screen" }).click(); - await expect(page.getByRole("heading", { level: 3, name: "Screen" })).toBeVisible(); +interface Target { + readonly deviceId: string; + readonly origin: string; + readonly token: string; + readonly workerId: string; } -function showButton(page: Page) { - return page.getByRole("button", { name: /^(Show screen|Stop)$/ }); +async function workerIdOn(reader: RunningDaemon): Promise { + const listed = await reader.cli(["worker", "list", "--json"]); + expect(listed.code, listed.stderr).toBe(0); + const [view] = (listed.json as { workers: { id: string }[] }).workers; + return view?.id ?? ""; +} + +async function deviceIds(host: RunningDaemon): Promise { + const result = await host.cli(["status", "--json"]); + expect(result.code, result.stderr).toBe(0); + return (result.json as { devices: { id: string }[] }).devices.map((device) => device.id); +} + +// why: a lease then a release leaves one warm device that nobody holds, which is the device a viewer +// can watch without a lease. +async function warmDevice(host: RunningDaemon): Promise { + const granted = await host.cli([...LEASE, "--agent-id", "console-screen", "--detach"]); + expect(granted.code, granted.stderr).toBe(0); + const released = await host.cli(["release", "--all", "--yes"]); + expect(released.code, released.stderr).toBe(0); + let ready = ""; + await expect + .poll(async () => { + const listed = await host.cli(["status", "--json"]); + const [device] = (listed.json as { devices: { id: string; state: string }[] }).devices; + ready = device?.state === "ready" ? device.id : ""; + return ready; + }) + .not.toBe(""); + return ready; +} + +// why: with the warm pool off a released device is left shut down, which is the one state a viewer +// can reach a device in without a lease changing under them. +async function shutdownDevice(host: RunningDaemon): Promise { + const granted = await host.cli([...LEASE, "--agent-id", "console-screen", "--detach"]); + expect(granted.code, granted.stderr).toBe(0); + const released = await host.cli(["release", "--all", "--yes"]); + expect(released.code, released.stderr).toBe(0); + let id = ""; + await expect + .poll(async () => { + const listed = await host.cli(["status", "--json"]); + const [device] = (listed.json as { devices: { id: string; state: string }[] }).devices; + id = device?.state === "shutdown" ? device.id : ""; + return id; + }) + .not.toBe(""); + return id; +} + +async function openDevice(page: Page, target: Target, search = "?tab=devices"): Promise { + const path = `/fleet/${encodeURIComponent(target.workerId)}/${encodeURIComponent(target.deviceId)}${search}`; + await page.goto(new URL(path, target.origin).toString()); + await signIn(page, target.token); + await expect(page.getByRole("dialog")).toBeVisible(); +} + +// why: the page is already signed in on this origin, so the token form is not there to fill. +async function reopen(page: Page, target: Target): Promise { + const path = `/fleet/${encodeURIComponent(target.workerId)}/${encodeURIComponent(target.deviceId)}?tab=devices&drawer=screen`; + await page.goto(new URL(path, target.origin).toString()); + await expect(page.getByRole("dialog")).toBeVisible(); +} + +// why: the stream numbers' popover is a dialog too, so the drawer is the one that holds the tabs. +function drawer(page: Page) { + return page.getByRole("dialog").filter({ has: page.getByRole("tablist") }); } function screen(page: Page) { - return page.getByLabel("The leased device's screen"); + return page.getByLabel("The device's screen"); } // why: the mean colour of what the canvas holds, so "the picture changed" is a number rather than a @@ -91,189 +160,276 @@ async function averageColour(page: Page): Promise { return rgb.join(","); } -async function fact(page: Page, label: string): Promise { - return page - .getByRole("region", { exact: true, name: "Screen" }) - .locator(".facts > div") - .filter({ has: page.getByText(label, { exact: true }) }) - .first() - .getByRole("definition") - .innerText(); -} - -async function frames(page: Page): Promise { - const shown = await fact(page, "Frames"); - const parsed = Number.parseInt(shown, 10); - return Number.isNaN(parsed) ? 0 : parsed; -} - -async function streamCapability(host: RunningDaemon): Promise { - const listed = await host.cli(["worker", "list", "--json"]); - expect(listed.code, listed.stderr).toBe(0); - const views = (listed.json as { workers: { capabilities?: { name: string; state: string }[] }[] }) - .workers; - return views[0]?.capabilities?.find((entry) => entry.name === "stream")?.state; -} - async function captureLogLines(dir: string, word: string): Promise { const log = await captureLog(dir); return log.split("\n").filter((line) => line.startsWith(word)).length; } -test.describe("the Screen pane", () => { - test("shows nothing and starts no capture until asked", async ({ page }) => { - const started = await streamingHost(); - try { - await openLeasePage(page, started.host, started.lease); - await expect(screen(page)).toHaveCount(0); - await expect(page.getByText("No picture while this is off.")).toBeVisible(); - await expect.poll(() => fact(page, "Picture")).toBe("—"); - await expect.poll(() => fact(page, "State")).toBe("off"); - expect(await captureLogLines(started.dir, "opened")).toBe(0); - expect(await streamCapability(started.host)).toBe("available"); - } finally { - await cleanup(started); - } - }); - - test("Show screen starts the capture, paints it, and reports what it is showing", async ({ +test.describe("the device drawer's Screen tab", () => { + test("opens on Details with no capture, starts one and paints it on Screen, closes it on Details or closing the drawer, and keeps it when the device is leased", async ({ page, }) => { const started = await streamingHost(); + const fleetDir = await mkdtemp(join(tmpdir(), "simlock-screen-")); + let fleet: Awaited> | undefined; try { - await openLeasePage(page, started.host, started.lease); - await showButton(page).click(); - await expect(screen(page)).toBeVisible(); - await logReached(started.dir, "opened", 1); + const gatewayCommand = await writeCaptureProgram(fleetDir); + fleet = await startFleet( + [ + { + config: { capabilities: { stream: { enabled: true, fps: 5 } } }, + driverScript: { ios: { ...IOS, screenCaptureCommand: gatewayCommand } }, + label: "worker-a", + }, + ], + { gatewayConfig: { capabilities: { stream: { enabled: true } } } }, + ); + const worker = fleet.workers[0]; + if (worker === undefined) throw new Error("the fleet has no worker"); + const cases = [ + { dir: started.dir, host: started.host, reader: started.host }, + { dir: fleetDir, host: worker, reader: fleet.gateway }, + ]; + for (const { dir, host, reader } of cases) { + const workerId = await workerIdOn(reader); + const watched = await warmDevice(host); + const target = { + deviceId: watched, + origin: reader.url, + token: reader.tokens.operator, + workerId, + }; + const streams: string[] = []; + const track = (request: { url(): string }) => { + if (request.url().includes("/v1/stream/")) streams.push(request.url()); + }; + page.on("requestfinished", track); + page.on("requestfailed", track); - // why: these numbers are the capture program's own, so the pane sizing itself from them proves it - // read the meta frame rather than guessing a size from whatever arrived first. - await expect.poll(() => fact(page, "Picture"), { timeout: 10_000 }).toBe("120 × 240 mjpeg"); - await expect.poll(async () => (await frames(page)) > 0, { timeout: 10_000 }).toBe(true); - await expect.poll(() => fact(page, "State"), { timeout: 10_000 }).toBe("live"); + await openDevice(page, target); + await expect(drawer(page).getByRole("tab", { name: "Details" })).toHaveAttribute( + "aria-selected", + "true", + ); + await expect(screen(page)).toHaveCount(0); + expect(await captureLogLines(dir, "opened")).toBe(0); - // why: three different colours arrive in turn, so readings that keep differing prove frames are - // still arriving rather than one picture painted once and left on the canvas. - const seen = new Set(); - await expect - .poll( - async () => { - seen.add(await averageColour(page)); - return seen.size; - }, - { timeout: 10_000 }, - ) - .toBeGreaterThanOrEqual(2); - } finally { - await cleanup(started); - } - }); + await drawer(page).getByRole("tab", { name: "Screen" }).click(); + await expect(page).toHaveURL(/[?&]drawer=screen(&|$)/); + await expect(screen(page)).toBeVisible(); + await logReached(dir, "opened", 1); + const seen = new Set(); + await expect + .poll( + async () => { + seen.add(await averageColour(page)); + return seen.size; + }, + { timeout: 10_000 }, + ) + .toBeGreaterThanOrEqual(2); - test("Stop stops the capture, and so does leaving the page", async ({ page }) => { - const started = await streamingHost(); - try { - await openLeasePage(page, started.host, started.lease); - await showButton(page).click(); - await logReached(started.dir, "opened", 1); - await expect.poll(async () => (await frames(page)) > 0, { timeout: 10_000 }).toBe(true); + const leased = await host.cli([...LEASE, "--agent-id", "console-screen", "--detach"]); + expect(leased.code, leased.stderr).toBe(0); + await page.waitForTimeout(500); + expect(await captureLogLines(dir, "opened")).toBe(1); + expect(await captureLogLines(dir, "closed")).toBe(0); + expect(streams).toEqual([]); - await showButton(page).click(); - await logReached(started.dir, "closed", 1); - await expect(screen(page)).toHaveCount(0); - await expect(showButton(page)).toHaveText("Show screen"); + await drawer(page).getByRole("tab", { name: "Details" }).click(); + await logReached(dir, "closed", 1); + await expect(screen(page)).toHaveCount(0); - await showButton(page).click(); - await logReached(started.dir, "opened", 2); - await page.goto(new URL("/leases", started.host.url).toString()); - await logReached(started.dir, "closed", 2); + await drawer(page).getByRole("tab", { name: "Screen" }).click(); + await logReached(dir, "opened", 2); + // why: Escape comes once the picture is up, so the test closes a capture that is streaming, + // as the Details click above does, and not one still starting. + await expect.poll(() => averageColour(page), { timeout: 10_000 }).not.toBe("0,0,0"); + await page.keyboard.press("Escape"); + await logReached(dir, "closed", 2); + await expect(drawer(page)).toBeHidden(); + await expect(page).not.toHaveURL(/drawer=/); - // why: arriving back at the lease page must not resume a picture nobody asked for -- a third - // "opened" here is a capture running for a viewer that only opened a page. - await page.goto( - new URL(`/leases/${encodeURIComponent(started.lease)}`, started.host.url).toString(), - ); - await page.getByRole("tab", { name: "Screen" }).click(); - await expect(showButton(page)).toHaveText("Show screen"); - expect(await captureLogLines(started.dir, "opened")).toBe(2); + const second = await host.cli([...LEASE, "--agent-id", "console-screen-2", "--detach"]); + expect(second.code, second.stderr).toBe(0); + const other = (await deviceIds(host)).find((id) => id !== watched) ?? ""; + expect(other).not.toBe(""); + await page + .getByRole("region", { name: "Devices", exact: true }) + .getByRole("row") + .filter({ hasText: other }) + .getByRole("link") + .first() + .click(); + await expect(drawer(page).getByRole("tab", { name: "Details" })).toHaveAttribute( + "aria-selected", + "true", + ); + await page.waitForTimeout(500); + expect(await captureLogLines(dir, "opened")).toBe(2); + page.off("requestfinished", track); + page.off("requestfailed", track); + } } finally { + if (fleet !== undefined) await disposeFleet(fleet); + await rm(fleetDir, { force: true, recursive: true }); await cleanup(started); } }); +}); + +const SHUT_DOWN = "The device is shut down. Its screen shows while it runs."; - test("switching from Screen to Details keeps the capture running, and closing the drawer stops it", async ({ +test.describe("the Screen tab of a device that is not running", () => { + test("says why instead of starting a capture, starts the picture when the device runs, shows the sentence again when it is shut down while watched, shows the daemon's words for a refused stream, and says streaming is off where it is", async ({ page, }) => { - const started = await streamingHost(); + const started = await streamingHost({ warmPool: { enabled: false } }); + const off = await startDaemon("worker", { + config: { warmPool: { enabled: false } }, + driverScript: { ios: IOS }, + }); try { - const ended: string[] = []; - const track = (request: { url(): string }) => { - if (request.url().includes("/v1/stream/leases/")) ended.push(request.url()); - }; - page.on("requestfinished", track); - page.on("requestfailed", track); - await openLeasePage(page, started.host, started.lease); - await showButton(page).click(); - await logReached(started.dir, "opened", 1); - await expect.poll(async () => (await frames(page)) > 0, { timeout: 10_000 }).toBe(true); + const host = started.host; + const workerId = await workerIdOn(host); + const deviceId = await shutdownDevice(host); + const streams: string[] = []; + page.on("request", (request) => { + if (request.url().includes("/v1/stream/")) streams.push(request.url()); + }); - await page.getByRole("tab", { name: "Details" }).click(); - await expect(page.getByRole("tab", { name: "Details" })).toHaveAttribute( - "aria-selected", - "true", + await openDevice( + page, + { deviceId, origin: host.url, token: host.tokens.operator, workerId }, + "?tab=devices&drawer=screen", ); - await page.waitForTimeout(500); - expect(ended).toEqual([]); - expect(await captureLogLines(started.dir, "closed")).toBe(0); + await expect(drawer(page).getByText(SHUT_DOWN)).toBeVisible(); + await expect(screen(page)).toHaveCount(0); + expect(streams).toEqual([]); + expect(await captureLogLines(started.dir, "opened")).toBe(0); - await page.keyboard.press("Escape"); + const leased = await host.cli([...LEASE, "--agent-id", "console-screen", "--detach"]); + expect(leased.code, leased.stderr).toBe(0); + await expect(screen(page)).toBeVisible(); + await logReached(started.dir, "opened", 1); + + const released = await host.cli(["release", "--all", "--yes"]); + expect(released.code, released.stderr).toBe(0); await logReached(started.dir, "closed", 1); - await expect.poll(() => ended.length).toBe(1); + await expect(drawer(page).getByText(SHUT_DOWN)).toBeVisible(); + await expect(screen(page)).toHaveCount(0); + expect(await captureLogLines(started.dir, "opened")).toBe(1); + + // why: the daemon's own words are the only thing a refused stream can tell the reader, so the + // answers are scripted per request; a 409 on a row that says `leased` is the row being behind. + const reasoned = await host.cli([...LEASE, "--agent-id", "console-screen-409", "--detach"]); + expect(reasoned.code, reasoned.stderr).toBe(0); + const refuse = (status: number, code: string, message: string) => + page.route("**/v1/stream/**", (route) => + route.fulfill({ + body: JSON.stringify({ error: { code, message } }), + contentType: "application/json", + status, + }), + ); + let held = ""; + await expect + .poll(async () => { + const listed = await host.cli(["status", "--json"]); + const devices = (listed.json as { devices: { id: string; state: string }[] }).devices; + held = devices.find((device) => device.state === "leased")?.id ?? ""; + return held; + }) + .not.toBe(""); + const target = { deviceId: held, origin: host.url, token: host.tokens.operator, workerId }; + await refuse(409, "DEVICE_NOT_RUNNING", "The row is behind: the device stopped just now."); + await reopen(page, target); + await expect( + drawer(page).getByText("The row is behind: the device stopped just now."), + ).toBeVisible(); + await expect(screen(page)).toHaveCount(0); + await expect(drawer(page).getByRole("alert")).toHaveCount(0); + await page.unroute("**/v1/stream/**"); + await refuse(500, "CAPTURE_FAILED", "The capture program could not start."); + await reopen(page, target); + await expect(drawer(page).getByRole("alert")).toHaveText( + "The capture program could not start.", + ); + await page.unroute("**/v1/stream/**"); + + // why: a leased device is one the daemon would stream, so only the worker's own setting can be + // the reason no request is made. + const heldOff = await off.cli([...LEASE, "--agent-id", "console-screen", "--detach"]); + expect(heldOff.code, heldOff.stderr).toBe(0); + const [silent = ""] = await deviceIds(off); + expect(silent).not.toBe(""); + const quiet: string[] = []; + page.on("request", (request) => { + if (request.url().includes("/v1/stream/")) quiet.push(request.url()); + }); + await openDevice( + page, + { + deviceId: silent, + origin: off.url, + token: off.tokens.operator, + workerId: await workerIdOn(off), + }, + "?tab=devices&drawer=screen", + ); + await expect(drawer(page).getByText(/^Streaming is off on /)).toBeVisible(); + expect(quiet).toEqual([]); } finally { + await off.dispose(); await cleanup(started); } }); +}); - test("a worker with streaming off disables Show screen and says why before any click", async ({ +test.describe("the stream numbers", () => { + test("sit behind the Stream details button, in a popover that Escape or a second click closes while the drawer and its capture stay", async ({ page, }) => { - const dir = await mkdtemp(join(tmpdir(), "simlock-screen-")); - let host: RunningDaemon | undefined; - try { - host = await startDaemon("worker", { driverScript: { ios: IOS } }); - const granted = await host.cli([...LEASE, "--agent-id", "console-screen", "--detach"]); - expect(granted.code, granted.stderr).toBe(0); - const lease = (granted.json as { lease: { id: string } }).lease.id; - await openLeasePage(page, host, lease); - await expect(showButton(page)).toBeDisabled(); - await expect(page.getByText(/^Streaming is off on /)).toBeVisible(); - expect(await streamCapability(host)).toBe("off"); - await expect(screen(page)).toHaveCount(0); - expect(await captureLog(dir)).toBe(""); - } finally { - await host?.dispose(); - await rm(dir, { force: true, recursive: true }); - } - }); - - test("a release takes the pane away and leaves no capture running", async ({ page }) => { const started = await streamingHost(); try { - await openLeasePage(page, started.host, started.lease); - await showButton(page).click(); + const host = started.host; + const target = { + deviceId: await warmDevice(host), + origin: host.url, + token: host.tokens.operator, + workerId: await workerIdOn(host), + }; + await openDevice(page, target, "?tab=devices&drawer=screen"); + await expect(screen(page)).toBeVisible(); await logReached(started.dir, "opened", 1); - await expect.poll(async () => (await frames(page)) > 0, { timeout: 10_000 }).toBe(true); - const released = await started.host.cli(["release", started.lease]); - expect(released.code, released.stderr).toBe(0); + const details = drawer(page).getByRole("button", { name: "Stream details" }); + const numbers = page.getByRole("dialog", { name: "Stream details" }); + await expect(drawer(page).getByText("Picture", { exact: true })).toHaveCount(0); + await expect(numbers).toHaveCount(0); - // why: a lease that is gone has no page to watch, so the pane goes with it and its abort stops the - // capture. What this proves is the promise an operator can see: leaving a lease behind never - // leaves a device being filmed. The guarantee that a lease ending stops a capture whose viewer is - // still connected is not observable from a browser -- it is pinned in - // `src/stream/captures.test.ts`, where a capture can be held open past the lease. - await logReached(started.dir, "closed", 1); - await expect(page.getByText("This lease is not held.")).toBeVisible(); - expect(await captureLogLines(started.dir, "opened")).toBe(1); + await details.click(); + await expect(numbers).toBeVisible(); + const labels = await numbers.locator(".facts dt").allTextContents(); + expect(labels).toEqual(["State", "Picture", "Frames", "Rate", "Bandwidth", "Dropped"]); + await expect(drawer(page).getByText("Picture", { exact: true })).toHaveCount(0); + const picture = numbers + .locator(".facts > div") + .filter({ has: page.getByText("Picture", { exact: true }) }) + .getByRole("definition"); + await expect(picture).toHaveText("120 × 240 mjpeg", { timeout: 10_000 }); + + await page.keyboard.press("Escape"); + await expect(numbers).toHaveCount(0); + await expect(drawer(page)).toBeVisible(); + + await details.click(); + await expect(numbers).toBeVisible(); + await details.click(); + await expect(numbers).toHaveCount(0); + await expect(drawer(page)).toBeVisible(); + await expect(screen(page)).toBeVisible(); + expect(await captureLogLines(started.dir, "closed")).toBe(0); } finally { await cleanup(started); } diff --git a/src/daemon/stream-gateway.e2e.test.ts b/src/daemon/stream-gateway.e2e.test.ts index 52720adc2..30842df48 100644 --- a/src/daemon/stream-gateway.e2e.test.ts +++ b/src/daemon/stream-gateway.e2e.test.ts @@ -143,16 +143,12 @@ describe("a screen stream through a gateway", () => { const deviceId = (granted as { lease: { deviceId: string } }).lease.deviceId; const { workers } = await gateway.dispatch("worker.list", {}, admin); const workerId = workers[0]?.id ?? ""; - const deviceStream = (path = `${workerId}/devices/${deviceId}`) => - fetch(`http://127.0.0.1:${String(port)}/v1/stream/workers/${path}`, { - headers: { authorization: `Bearer ${operator}` }, - }); const stream = (signal?: AbortSignal) => - fetch(`http://127.0.0.1:${String(port)}/v1/stream/leases/${encodeURIComponent(leaseId)}`, { + fetch(`http://127.0.0.1:${String(port)}/v1/stream/workers/${workerId}/devices/${deviceId}`, { headers: { authorization: `Bearer ${operator}` }, ...(signal === undefined ? {} : { signal }), }); - return { deviceStream, gateway, leaseId, screen, stream }; + return { gateway, leaseId, screen, stream }; } it( @@ -181,22 +177,6 @@ describe("a screen stream through a gateway", () => { }, ); - it( - "relays an operator's device stream through the gateway: the worker's frames", - { timeout: 30_000 }, - async () => { - const { deviceStream } = await fleet(true); - const response = await deviceStream(); - expect(response.status).toBe(200); - expect(response.headers.get("simlock-stream-codec")).toBe("h264"); - const reader = response.body?.getReader(); - if (reader === undefined) throw new Error("no body"); - const first = await reader.read(); - expect(first.value?.[4]).toBe(SCREEN_FRAME_TAG.meta); - await reader.cancel(); - }, - ); - it( "a viewer leaving the gateway stops the capture on the worker", { timeout: 30_000 }, diff --git a/src/drivers/ios/AGENTS.md b/src/drivers/ios/AGENTS.md index 9738b66d4..1c7a531fb 100644 --- a/src/drivers/ios/AGENTS.md +++ b/src/drivers/ios/AGENTS.md @@ -132,7 +132,7 @@ record of a device's mode; `driverData` holds the slim signature. - **A screen stream can freeze after a reboot under its lease.** The capture program checks the simulator is booted only when it starts, then repaints its last surface every 200 ms, so after the health monitor reboots a crashed simulator the viewer can see the old picture until they - press Stop and Show screen again. A lease ending stops it as usual. + leave the Screen tab and open it again. A lease ending stops it as usual. - **An iPhone rotated mid-stream keeps the first picture's scale**, as a simulator does today: the scaler is sized once, from the first picture. An iPhone that stays connected and unlocked but stops sending video, or a decoder that stops producing pictures, shows the last picture until diff --git a/src/http/app.test.ts b/src/http/app.test.ts index c0fe5e917..470070b87 100644 --- a/src/http/app.test.ts +++ b/src/http/app.test.ts @@ -3276,9 +3276,10 @@ describe("the stream mount's listed route", () => { const answer = (path: string, method = "GET") => app.request(path, { headers: agentAuth, method }); - const found = await answer(route.path.replace("{id}", "lse_1")); + const listedPath = route.path.replace("{workerId}", "w1").replace("{deviceId}", "dev_1"); + const found = await answer(listedPath); const other = await answer("/v1/stream/x"); - const wrongMethod = await answer(route.path.replace("{id}", "lse_1"), "POST"); + const wrongMethod = await answer(listedPath, "POST"); expect(found.status).toBe(200); expect(found.headers.get("content-type")).toBe("application/vnd.simlock.stream+h264"); @@ -3286,7 +3287,7 @@ describe("the stream mount's listed route", () => { expect(listed).toEqual( expect.arrayContaining([found.status, other.status, wrongMethod.status]), ); - expect(await other.json()).toMatchObject({ error: { code: "UNKNOWN_LEASE" } }); + expect(await other.json()).toMatchObject({ error: { code: "UNKNOWN_DEVICE" } }); await found.body?.cancel(); }); }); diff --git a/src/http/core-routes.test.ts b/src/http/core-routes.test.ts index b877ccdb4..ebeabc822 100644 --- a/src/http/core-routes.test.ts +++ b/src/http/core-routes.test.ts @@ -85,10 +85,7 @@ describe("the routes createHttpApp registers", () => { expect(registered.some((route) => route.path.startsWith("/v1/stream"))).toBe(true); expect(registered.some((route) => route.path.startsWith("/v1/agent-device"))).toBe(true); expect(coverageProblems(core, declared)).toEqual([]); - expect(moduleRoutes).toEqual([ - "GET /v1/stream/leases/{id}", - "GET /v1/stream/workers/{workerId}/devices/{deviceId}", - ]); + expect(moduleRoutes).toEqual(["GET /v1/stream/workers/{workerId}/devices/{deviceId}"]); expect([...documented].sort()).toEqual([...declared, ...moduleRoutes].sort()); expect(coverageProblems([...core, "GET /v1/new"], declared)).toEqual([ "not declared: GET /v1/new", diff --git a/src/http/openapi.test.ts b/src/http/openapi.test.ts index 599ed2488..0903f7aac 100644 --- a/src/http/openapi.test.ts +++ b/src/http/openapi.test.ts @@ -350,12 +350,11 @@ describe("buildOpenApiDocument with a module that carries http", () => { expect(result.valid).toBe(true); }); - it("an integrator reads GET /v1/stream/leases/{id} and GET /v1/stream/workers/{workerId}/devices/{deviceId} under stream, with its headers, binary 200, 405 and refusals, and no agent-device or uplink path", () => { + it("an integrator reads GET /v1/stream/workers/{workerId}/devices/{deviceId} under stream, with its headers, binary 200, 405 and refusals, and no lease stream, agent-device or uplink path", () => { const document = buildOpenApiDocument(optionalModules); - const operation = document.paths?.["/v1/stream/leases/{id}"]?.get; - const ok = (operation?.responses ?? {})["200"] as OkResponse; + const device = document.paths?.["/v1/stream/workers/{workerId}/devices/{deviceId}"]?.get; + const ok = (device?.responses ?? {})["200"] as OkResponse; - expect(operation?.tags).toEqual(["stream"]); expect(document.tags?.find((tag) => tag.name === "stream")?.description).toContain( "capabilities.stream.enabled", ); @@ -363,31 +362,9 @@ describe("buildOpenApiDocument with a module that carries http", () => { "application/vnd.simlock.stream+h264", "application/vnd.simlock.stream+mjpeg", ]); - expect(sortedKeys(ok.headers)).toEqual([ - "simlock-stream-codec", - "simlock-stream-fps", - "simlock-stream-height", - "simlock-stream-width", - "x-simlock-lease", + expect(sortedKeys(document.paths).filter((path) => path.startsWith("/v1/stream"))).toEqual([ + "/v1/stream/workers/{workerId}/devices/{deviceId}", ]); - expect( - Object.fromEntries( - ["401", "403", "404", "405", "415", "500", "502", "503"].map((status) => [ - status, - errorCodes(operation, status), - ]), - ), - ).toEqual({ - "401": ["UNAUTHENTICATED"], - "403": ["FORBIDDEN"], - "404": ["UNKNOWN_LEASE"], - "405": ["BAD_REQUEST"], - "415": ["STREAM_NOT_SUPPORTED"], - "500": ["INTERNAL"], - "502": ["STREAM_UNAVAILABLE"], - "503": ["STREAM_UNAVAILABLE", "WORKER_UNREACHABLE"], - }); - const device = document.paths?.["/v1/stream/workers/{workerId}/devices/{deviceId}"]?.get; expect(device?.tags).toEqual(["stream"]); expect( Object.fromEntries( diff --git a/src/stream/AGENTS.md b/src/stream/AGENTS.md index a9781d91a..f8dd46f8e 100644 --- a/src/stream/AGENTS.md +++ b/src/stream/AGENTS.md @@ -1,6 +1,6 @@ # Stream -Does: serves a running device's live screen to operators at `GET /v1/stream/workers/{workerId}/devices/{deviceId}` (and, until #910, a lease's at `GET /v1/stream/leases/{leaseId}`), on a worker and through a gateway. +Does: serves a running device's live screen to operators at `GET /v1/stream/workers/{workerId}/devices/{deviceId}` on a worker and through a gateway. Entry: `index.ts` → `streamModule`, an optional module (ADR 0030): `worker.ts` `createWorker`, `gateway.ts` `createGateway`, loaded by the module host only when turned on. Uses: core (`ScreenFeed`, frame tags, registry records), modules (the context and hook types), ports. @@ -9,51 +9,48 @@ Off unless `capabilities.stream.enabled` (false). Off: nothing loaded, and the ` ## Files -| File | Role | -| ------------------------- | ------------------------------------------------------------------------------------------ | -| `definition.ts` | `streamModule`: name `stream`, config section, `http` route data, the two loaders | -| `worker.ts`, `gateway.ts` | The module's two sides, built from the host's context | -| `create.ts` | The two factories; the `stream` capability; the four device events that stop captures | -| `worker-link.ts` | Worker: lease or device check, open through the driver, register, re-check after frame 1 | -| `gateway-link.ts` | Gateway: fleet ownership (lease) or known worker (device), relay, refusal and header check | -| `feed.ts` | `readScreenFeed`: meta frame first, start and silence bounds, close on any end | -| `framing.ts` | `MetaFrameScanner`, `checkedMeta`: codec and size a picture may claim | -| `captures.ts` | Captures per lease and per device; the lease-end participant that stops them | -| `service.ts` | One request either door: method, refusal body, response headers, log line | -| `mounts.ts` | The HTTP mount and the relay target | -| `refusal.ts` | `StreamRefusalError`, each refusal's status and code, a relayed refusal checked | -| `caller.ts` | Who asks | -| `test-fixtures.ts` | `TestRegistry`, `ScriptedFeed` for tests | -| `testing.ts` | What other components' tests may import: `createStreamMount` | +| File | Role | +| ------------------------- | ------------------------------------------------------------------------------------- | +| `definition.ts` | `streamModule`: name `stream`, config section, `http` route data, the two loaders | +| `worker.ts`, `gateway.ts` | The module's two sides, built from the host's context | +| `create.ts` | The two factories; the `stream` capability; the four device events that stop captures | +| `worker-link.ts` | Worker: device check, open through the driver, register, re-check after frame 1 | +| `gateway-link.ts` | Gateway: known worker, relay, refusal and header check | +| `feed.ts` | `readScreenFeed`: meta frame first, start and silence bounds, close on any end | +| `framing.ts` | `MetaFrameScanner`, `checkedMeta`: codec and size a picture may claim | +| `captures.ts` | Captures per device; the lease-end participant that stops them | +| `service.ts` | One request either door: method, refusal body, response headers, log line | +| `mounts.ts` | The HTTP mount and the relay target | +| `refusal.ts` | `StreamRefusalError`, each refusal's status and code, a relayed refusal checked | +| `caller.ts` | Who asks | +| `test-fixtures.ts` | `TestRegistry`, `ScriptedFeed` for tests | +| `testing.ts` | What other components' tests may import: `createStreamMount` | ## How it works Worker — `worker-link.ts` `createWorkerStreamLink`: -1. Lease route: the lease from the registry snapshot. Agent: its requester (relayed) or principal - must hold it. Unknown and not yours both answer 404 `UNKNOWN_LEASE`; nothing is opened first. - Device route: operator only (an agent gets 404 `UNKNOWN_DEVICE`, as for any device), the path's - worker id must be this worker's own (404 `UNKNOWN_WORKER`), the device must exist (404 - `UNKNOWN_DEVICE`) and be `ready` or `leased` (409 `DEVICE_NOT_RUNNING`). A relayed device - request is served whatever the tenant: the gateway has already refused non-operators. -2. `captures.begin(leaseId)` or `captures.beginOnDevice(deviceId)` — registered before the capture starts. +1. Operator only (an agent gets 404 `UNKNOWN_DEVICE`, as for any device), the path's worker id + must be this worker's own (404 `UNKNOWN_WORKER`), the device must exist (404 `UNKNOWN_DEVICE`) + and be `ready` or `leased` (409 `DEVICE_NOT_RUNNING`); nothing is opened first. A relayed + request is served whatever the tenant: the gateway has already refused non-operators. Any other + stream path, a lease's included, answers 404 `UNKNOWN_DEVICE` "This path names no stream". +2. `captures.beginOnDevice(deviceId)` — registered before the capture starts. 3. `ctx.screens.openScreen(platform, device, settings)` (the daemon passes its `DriverCatalog`) → the driver's `ScreenFeed`. `ScreenNotSupportedError` → 415 `STREAM_NOT_SUPPORTED`; anything else → 503 `STREAM_UNAVAILABLE`. 4. `readScreenFeed`: the meta frame within the start bound, or 503 and close. -5. Lease re-read: gone, re-granted, or its slot stopped → close, 404. A device slot stopped → close, 409 `DEVICE_NOT_RUNNING`. +5. Its slot stopped meanwhile → close, 409 `DEVICE_NOT_RUNNING`. 6. 200; headers `simlock-stream-{codec,fps,width,height}`; body = the frames, meta first. Ends — each closes the capture once (`closedOnce`): the viewer cancels or aborts; the capture ends or errors; no byte within the silence bound; the lease ends (the `beforeDeviceReclaim` hook, `createStreamLeaseEnd` → -`captures.stop` and `captures.stopOnDevice`, which wait for every close, bounded by the driver and by leasing's signal); +`captures.stopOnDevice`, which waits for every close, bounded by the driver and by leasing's signal); its device stops with no lease end (`device.shutdown`, `device.absence-detected`, `device.quarantined` or `device.deleted` → `captures.stopOnDevice`). A lease granted while streaming ends nothing. -Gateway — `gateway-link.ts`: the fleet lease index names owner, worker and the worker's lease -id. Not the owner (agent) → 404. Relays `GET /stream/leases/` as target `stream` -with tenant `` from the index, never from the caller. A device -request: not an operator → 404 `UNKNOWN_DEVICE`; a worker not in the fleet → 404 `UNKNOWN_WORKER`; -else relays `GET /stream/devices/` with tenant `` and no lease id. A refusal +Gateway — `gateway-link.ts`: not an operator → 404 `UNKNOWN_DEVICE`; a worker not in the fleet → +404 `UNKNOWN_WORKER`; else relays `GET /stream/devices/` with tenant +``. A refusal body is read to 16 KiB; its code and message are forwarded only if the code matches `^[A-Z][A-Z0-9_]{0,63}$`, else the code becomes `STREAM_UNAVAILABLE`. The worker's 4xx or 5xx status is kept either way; a 2xx without a picture → 502. Headers go through `checkedMeta`. @@ -64,7 +61,7 @@ cannot says why in its 503); `unavailable` when none can, with each platform's r ## State and timers -Memory only: running captures per lease and per device (`captures.ts`). +Memory only: running captures per device (`captures.ts`). | What | Default | Key | | ---------------------------- | -------------------- | ------------------------------------------------------------------------------------------------ | @@ -76,11 +73,11 @@ Memory only: running captures per lease and per device (`captures.ts`). ## Rules - No platform name here: a driver offers `screenCapture` or not (`core/screen.ts`). -- Ownership before any work; unknown and not-yours answer the same. -- A capture is registered before it starts and the lease is re-read after its first frame, - so a capture never outlives its lease or its device's running state (`worker-link.test.ts`, +- The caller and the device are checked before any work; an agent and an unknown device get the same answer. +- A capture is registered before it starts and its slot is read again after its first frame, + so a capture never outlives its device's running state (`worker-link.test.ts`, `create.test.ts`). -- A gateway never parses a video byte and never takes a caller's word for the requester. +- A gateway never parses a video byte and never takes a caller's word for the tenant. - No business events: watching changes no lease or device. Each open and refusal is logged. ## Known gaps diff --git a/src/stream/caller.ts b/src/stream/caller.ts index c5f78a84d..22b319115 100644 --- a/src/stream/caller.ts +++ b/src/stream/caller.ts @@ -2,6 +2,5 @@ export type StreamRole = "agent" | "operator"; export interface StreamCaller { readonly principal: string; - readonly requesterId?: string; readonly role: StreamRole; } diff --git a/src/stream/captures.test.ts b/src/stream/captures.test.ts index 2622fa274..147e5ee3a 100644 --- a/src/stream/captures.test.ts +++ b/src/stream/captures.test.ts @@ -6,16 +6,16 @@ async function flush(): Promise { for (let turn = 0; turn < 10; turn += 1) await new Promise((resolve) => setImmediate(resolve)); } -function ended(leaseId: string) { - return { leaseId } as Parameters["leaseEnded"]>[0]; +function ended(deviceId: string) { + return { deviceId } as Parameters["leaseEnded"]>[0]; } describe("the capture registry", () => { - it("a lease that ends while its capture is starting aborts the slot, waits for it to attach, and closes it", async () => { + it("a device that stops while its capture is starting aborts the slot, waits for it to attach, and closes it", async () => { const captures = createCaptureRegistry(); - const slot = captures.begin("lse_1"); + const slot = captures.beginOnDevice("dev_1"); let stopped = false; - const stopping = captures.stop("lse_1").then(() => (stopped = true)); + const stopping = captures.stopOnDevice("dev_1").then(() => (stopped = true)); expect(slot.ended.aborted).toBe(true); await Promise.resolve(); expect(stopped).toBe(false); @@ -27,53 +27,54 @@ describe("the capture registry", () => { expect(closes).toBe(1); }); - it("stops every capture of the lease that ended and leaves another lease's running", async () => { + it("stops every capture of the device that stopped and leaves another device's running", async () => { const captures = createCaptureRegistry(); const closed: string[] = []; - captures.begin("lse_1").attach(async () => void closed.push("a")); - captures.begin("lse_1").attach(async () => void closed.push("b")); - captures.begin("lse_2").attach(async () => void closed.push("c")); - expect(await captures.stop("lse_1")).toBe(2); + captures.beginOnDevice("dev_1").attach(async () => void closed.push("a")); + captures.beginOnDevice("dev_1").attach(async () => void closed.push("b")); + captures.beginOnDevice("dev_2").attach(async () => void closed.push("c")); + expect(await captures.stopOnDevice("dev_1")).toBe(2); expect(closed).toEqual(["a", "b"]); - expect(captures.leases()).toEqual(["lse_2"]); + expect(await captures.stopOnDevice("dev_2")).toBe(1); }); it("closes a capture once however many doors close it", async () => { const captures = createCaptureRegistry(); let closes = 0; - const slot = captures.begin("lse_1"); + const slot = captures.beginOnDevice("dev_1"); slot.attach(async () => { closes += 1; }); await slot.release(); await slot.release(); - await captures.stop("lse_1"); + await captures.stopOnDevice("dev_1"); expect(closes).toBe(1); }); - it("an abandoned open no longer counts as its lease's", async () => { + it("an abandoned open no longer counts as its device's", async () => { const captures = createCaptureRegistry(); - captures.begin("lse_1").abandon(); - expect(captures.leases()).toEqual([]); - expect(await captures.stop("lse_1")).toBe(0); + captures.beginOnDevice("dev_1").abandon(); + expect(await captures.stopOnDevice("dev_1")).toBe(0); }); - it("a released capture counts as its lease's until it is closed", async () => { + it("a released capture counts as its device's until it is closed", async () => { const captures = createCaptureRegistry(); - const slot = captures.begin("lse_1"); + const slot = captures.beginOnDevice("dev_1"); let finish!: () => void; slot.attach(() => new Promise((resolve) => (finish = resolve))); const releasing = slot.release(); await flush(); - expect(captures.leases()).toEqual(["lse_1"]); + const stopping = captures.stopOnDevice("dev_1"); + await flush(); finish(); await releasing; - expect(captures.leases()).toEqual([]); + expect(await stopping).toBe(1); + expect(await captures.stopOnDevice("dev_1")).toBe(0); }); it("a slot released before its capture attached closes that capture once it attaches", async () => { const captures = createCaptureRegistry(); - const slot = captures.begin("lse_1"); + const slot = captures.beginOnDevice("dev_1"); const releasing = slot.release(); let closes = 0; slot.attach(async () => { @@ -81,18 +82,20 @@ describe("the capture registry", () => { }); await releasing; expect(closes).toBe(1); - expect(captures.leases()).toEqual([]); + expect(await captures.stopOnDevice("dev_1")).toBe(0); }); }); describe("the stream's lease-end participant", () => { - it("returns only once the ended lease's captures have closed", async () => { + it("returns only once the ended lease's device's captures have closed", async () => { const captures = createCaptureRegistry(); let finish!: () => void; - captures.begin("lse_1").attach(() => new Promise((resolve) => (finish = resolve))); + captures + .beginOnDevice("dev_1") + .attach(() => new Promise((resolve) => (finish = resolve))); let returned = false; const done = createStreamLeaseEnd(captures) - .leaseEnded(ended("lse_1"), new AbortController().signal) + .leaseEnded(ended("dev_1"), new AbortController().signal) .then(() => { returned = true; }); @@ -105,10 +108,10 @@ describe("the stream's lease-end participant", () => { it("stops waiting when leasing gives up on it", async () => { const captures = createCaptureRegistry(); - captures.begin("lse_1").attach(() => new Promise(() => undefined)); + captures.beginOnDevice("dev_1").attach(() => new Promise(() => undefined)); const abort = new AbortController(); const done = createStreamLeaseEnd(captures) - .leaseEnded(ended("lse_1"), abort.signal) + .leaseEnded(ended("dev_1"), abort.signal) .then(() => "returned"); abort.abort(); const outcome = await Promise.race([ diff --git a/src/stream/captures.ts b/src/stream/captures.ts index 5ab17cd8e..be4cf12d4 100644 --- a/src/stream/captures.ts +++ b/src/stream/captures.ts @@ -1,12 +1,11 @@ -// why: a capture belongs to the lease or the device it films, not to a connection -- the lease is the thing -// that says this machine still owns that device, and the device is the thing that says it still runs. A -// viewer going away stops a capture through its own request; this covers what that cannot, a lease that -// ends somewhere else (released elsewhere, expired, the device lost) while somebody is still watching, or -// while a capture is still starting. +// why: a capture belongs to the device it films, not to a connection -- the device is the thing that says +// it still runs. A viewer going away stops a capture through its own request; this covers what that +// cannot, a device that stops running or whose lease ends (released elsewhere, expired, the device lost) +// while somebody is still watching, or while a capture is still starting. import type { ModuleLeaseEndParticipant } from "../modules/index.js"; export interface CaptureSlot { - // why: aborted when the lease ends, so a capture still being opened learns it has no lease to + // why: aborted when the device stops, so a capture still being opened learns it has nothing to // film before it hands a picture out. readonly ended: AbortSignal; attach(close: () => Promise): void; @@ -19,9 +18,6 @@ export interface CaptureSlot { } export interface CaptureRegistry { - begin(leaseId: string): CaptureSlot; - stop(leaseId: string): Promise; - leases(): string[]; // why: a device's stream belongs to the device, not to a lease, because a device nobody leases streams // too; its captures end when the device stops running or its lease ends, whoever the viewer is. beginOnDevice(deviceId: string): CaptureSlot; @@ -41,7 +37,6 @@ interface Slot { interface SlotTable { begin(key: string): CaptureSlot; stop(key: string): Promise; - keys(): string[]; } function createSlotTable(): SlotTable { @@ -57,17 +52,17 @@ function createSlotTable(): SlotTable { await slot.closing; }; - const forget = (leaseId: string, slot: Slot) => { - const forLease = running.get(leaseId); - forLease?.delete(slot); - if (forLease?.size === 0) running.delete(leaseId); + const forget = (key: string, slot: Slot) => { + const forKey = running.get(key); + forKey?.delete(slot); + if (forKey?.size === 0) running.delete(key); }; return { // why: registered before the capture starts, not after its first frame -- a lease that ends in // between must find something to stop, or the capture would come up filming a device that is // already on its way to the next holder. - begin: (leaseId) => { + begin: (key) => { let settle = () => undefined as void; const settled = new Promise((resolve) => (settle = resolve)); const slot: Slot = { @@ -77,9 +72,9 @@ function createSlotTable(): SlotTable { settle, settled, }; - const forLease = running.get(leaseId) ?? new Set(); - forLease.add(slot); - running.set(leaseId, forLease); + const forKey = running.get(key) ?? new Set(); + forKey.add(slot); + running.set(key, forKey); return { // why: a capture can finish opening after its viewer left or its lease ended; whoever let // go of the slot is waiting on `settled`, and closes it the moment it attaches. @@ -90,36 +85,28 @@ function createSlotTable(): SlotTable { ended: slot.controller.signal, abandon: () => { slot.settle(); - forget(leaseId, slot); + forget(key, slot); }, release: async () => { await closeSlot(slot); - forget(leaseId, slot); + forget(key, slot); }, }; }, - keys: () => [...running.keys()], - stop: async (leaseId) => { - const forLease = running.get(leaseId); - if (forLease === undefined) return 0; - running.delete(leaseId); - for (const slot of forLease) slot.controller.abort(); - await Promise.allSettled([...forLease].map(closeSlot)); - return forLease.size; + stop: async (key) => { + const forKey = running.get(key); + if (forKey === undefined) return 0; + running.delete(key); + for (const slot of forKey) slot.controller.abort(); + await Promise.allSettled([...forKey].map(closeSlot)); + return forKey.size; }, }; } export function createCaptureRegistry(): CaptureRegistry { - const leases = createSlotTable(); const devices = createSlotTable(); - return { - begin: leases.begin, - beginOnDevice: devices.begin, - leases: leases.keys, - stop: leases.stop, - stopOnDevice: devices.stop, - }; + return { beginOnDevice: devices.begin, stopOnDevice: devices.stop }; } export function createStreamLeaseEnd(captures: CaptureRegistry): ModuleLeaseEndParticipant { @@ -129,7 +116,7 @@ export function createStreamLeaseEnd(captures: CaptureRegistry): ModuleLeaseEndP // the wait if leasing gives up first. leaseEnded: async (ended, signal) => { await Promise.race([ - Promise.all([captures.stop(ended.leaseId), captures.stopOnDevice(ended.deviceId)]), + captures.stopOnDevice(ended.deviceId), new Promise((resolve) => { if (signal.aborted) resolve(); else signal.addEventListener("abort", () => resolve(), { once: true }); diff --git a/src/stream/create.test.ts b/src/stream/create.test.ts index cef17808e..7be66bb2e 100644 --- a/src/stream/create.test.ts +++ b/src/stream/create.test.ts @@ -12,19 +12,16 @@ function worker( openGate: Promise = Promise.resolve(), ) { const clock = new FakeClock(); - const registry = new TestRegistry( - [{ deviceId: "dev_1", id: "lse_1", ownerId: "tok_agent", requesterId: "gw:inst:agent-1" }], - [ - { id: "dev_1", platform: "ios" }, - { id: "dev_ready", platform: "ios", state: "ready" }, - { id: "dev_leased", platform: "ios", state: "leased" }, - { id: "dev_shutdown", platform: "ios", state: "shutdown" }, - { id: "dev_provisioning", platform: "ios", state: "provisioning" }, - { id: "dev_reclaiming", platform: "ios", state: "reclaiming" }, - { id: "dev_quarantined", platform: "ios", state: "quarantined" }, - { id: "dev_absent", physical: true, platform: "ios", state: "absent" }, - ], - ); + const registry = new TestRegistry([ + { id: "dev_1", platform: "ios" }, + { id: "dev_ready", platform: "ios", state: "ready" }, + { id: "dev_leased", platform: "ios", state: "leased" }, + { id: "dev_shutdown", platform: "ios", state: "shutdown" }, + { id: "dev_provisioning", platform: "ios", state: "provisioning" }, + { id: "dev_reclaiming", platform: "ios", state: "reclaiming" }, + { id: "dev_quarantined", platform: "ios", state: "quarantined" }, + { id: "dev_absent", physical: true, platform: "ios", state: "absent" }, + ]); const feeds: ScriptedFeed[] = []; let opened = 0; const handlers = new Map void)[]>(); @@ -74,7 +71,7 @@ const OPERATOR = { principal: "tok_op", role: "operator" }; const deviceRequest = (worker: string, device: string) => new Request(`http://daemon/v1/stream/workers/${worker}/devices/${device}`); -const relayed = () => new Request("http://simlock-worker/stream/leases/lse_1"); +const relayed = () => new Request("http://simlock-worker/stream/devices/dev_1"); async function settledWithin(promise: Promise): Promise { for (let flush = 0; flush < 10; flush += 1) await new Promise((resolve) => setImmediate(resolve)); @@ -95,7 +92,7 @@ describe("createWorkerStream", () => { it("gives a capture 30 s to send its first frame", async () => { const { clock, feeds, stream } = worker([], false); - const served = stream.relayTarget(relayed(), "gw:inst:agent-1", "lse_1"); + const served = stream.relayTarget(relayed(), "gw:inst:agent-1"); await settledWithin(served); clock.advance(29_999); expect(await settledWithin(served)).toBe("pending"); @@ -107,7 +104,7 @@ describe("createWorkerStream", () => { it("ends a stream whose capture sends nothing for 30 s", async () => { const { clock, feeds, stream } = worker(); - const served = await stream.relayTarget(relayed(), "gw:inst:agent-1", "lse_1"); + const served = await stream.relayTarget(relayed(), "gw:inst:agent-1"); const reader = served.body!.getReader(); await reader.read(); const next = reader.read().then( @@ -140,26 +137,12 @@ describe("createWorkerStream", () => { }); }); - it("serves a relayed request for the requester the gateway names, checked against this worker's registry", async () => { - const { stream } = worker(); - const request = () => new Request("http://simlock-worker/stream/leases/lse_1"); - const served = await stream.relayTarget(request(), "gw:inst:agent-1", "lse_1"); - expect(served.status).toBe(200); - await served.body?.cancel(); - const refused = await stream.relayTarget(request(), "gw:inst:agent-2", "lse_1"); - expect(refused.status).toBe(404); - }); - it("stops a relayed capture when its lease ends", async () => { const { feeds, stream } = worker(); - const served = await stream.relayTarget( - new Request("http://simlock-worker/stream/leases/lse_1"), - "gw:inst:agent-1", - "lse_1", - ); + const served = await stream.relayTarget(relayed(), "gw:inst:agent-1"); expect(served.status).toBe(200); await stream.leaseEnd.leaseEnded( - { leaseId: "lse_1" } as Parameters[0], + { deviceId: "dev_1", leaseId: "lse_1" } as Parameters[0], new AbortController().signal, ); expect(feeds[0]?.closes).toBe(1); @@ -241,7 +224,6 @@ describe("createWorkerStream", () => { const relays: string[] = []; const gateway = createGatewayStream({ fleet: { - holder: () => undefined, knowsWorker: () => true, relay: async (workerId) => { relays.push(workerId); @@ -282,7 +264,6 @@ describe("createWorkerStream", () => { const relays: { workerId: string; url: string; tenant: string }[] = []; const gateway = createGatewayStream({ fleet: { - holder: () => undefined, knowsWorker: (workerId) => workerId in answers, relay: async (workerId, call) => { relays.push({ tenant: call.tenant, url: call.request.url, workerId }); @@ -338,12 +319,6 @@ describe("createWorkerStream", () => { registry.devices = registry.devices.map((device) => device.id === "dev_ready" ? { ...device, state: "leased" } : device, ); - registry.leases.push({ - deviceId: "dev_ready", - id: "lse_2", - ownerId: "tok_agent", - requesterId: "agent-2", - }); feeds[0]?.keyframe([9]); expect((await reader.read()).done).toBe(false); expect(feeds[0]?.closes).toBe(0); diff --git a/src/stream/create.ts b/src/stream/create.ts index a1c5f5bc6..cfe0049cd 100644 --- a/src/stream/create.ts +++ b/src/stream/create.ts @@ -46,7 +46,7 @@ export interface WorkerStream { } // why: the mount, the relay target and the lease-end participant sit on one link and one registry, so -// a stream asked for on this host and one relayed by a gateway go through the same lease check, and a +// a stream asked for on this host and one relayed by a gateway go through the same device check, and a // lease that ends has one place to stop its captures whichever door the picture came in by. export function createWorkerStream(options: WorkerStreamOptions): WorkerStream { const captures = createCaptureRegistry(); diff --git a/src/stream/definition.ts b/src/stream/definition.ts index 6e8193b91..4094d9c76 100644 --- a/src/stream/definition.ts +++ b/src/stream/definition.ts @@ -22,46 +22,6 @@ export const streamModule: OptionalModuleDefinition = { }, http: { routes: [ - { - method: "get", - path: "/v1/stream/leases/{id}", - description: - "Watch the screen of a leased device. Served only when capabilities.stream.enabled is true. A gateway keeps a worker's refusal with its code.", - statuses: { - 200: { - description: "The picture, as a binary body of frames; its codec and size are headers.", - contentTypes: [ - "application/vnd.simlock.stream+h264", - "application/vnd.simlock.stream+mjpeg", - ], - headers: { - "simlock-stream-codec": "The codec of the body: h264 or mjpeg.", - "simlock-stream-fps": "Frames per second.", - "simlock-stream-height": "Picture height in pixels.", - "simlock-stream-width": "Picture width in pixels.", - "x-simlock-lease": "The lease being watched.", - }, - }, - 404: { - description: "No such lease, or no such stream path.", - errorCodes: ["UNKNOWN_LEASE"], - }, - 405: { description: "A stream is read, not written.", errorCodes: ["BAD_REQUEST"] }, - 415: { - description: "The leased device cannot be streamed.", - errorCodes: ["STREAM_NOT_SUPPORTED"], - }, - 500: { description: "The stream failed unexpectedly.", errorCodes: ["INTERNAL"] }, - 502: { - description: "A gateway's worker answered success with no picture.", - errorCodes: ["STREAM_UNAVAILABLE"], - }, - 503: { - description: "The stream is not available, or a gateway cannot reach the worker.", - errorCodes: ["STREAM_UNAVAILABLE", "WORKER_UNREACHABLE"], - }, - }, - }, { method: "get", path: "/v1/stream/workers/{workerId}/devices/{deviceId}", diff --git a/src/stream/gateway-link.test.ts b/src/stream/gateway-link.test.ts index fab8ed699..65a0aaa3e 100644 --- a/src/stream/gateway-link.test.ts +++ b/src/stream/gateway-link.test.ts @@ -4,14 +4,13 @@ import type { StreamCaller } from "./caller.js"; import { createGatewayStreamLink, type FleetPort } from "./gateway-link.js"; import { StreamRefusalError } from "./refusal.js"; -const AGENT: StreamCaller = { principal: "tok_agent", role: "agent" }; -const STRANGER: StreamCaller = { principal: "tok_other", role: "agent" }; +const OPERATOR: StreamCaller = { principal: "tok_op", role: "operator" }; +const DEVICE = { deviceId: "dev_9", workerId: "w1" }; interface Relayed { readonly workerId: string; readonly url: string; readonly tenant: string; - readonly leaseId: string | undefined; } class CodedError extends Error { @@ -27,18 +26,9 @@ function fleet(answer: () => Response): { port: FleetPort; relayed: Relayed[] } const relayed: Relayed[] = []; return { port: { - holder: (leaseId) => - leaseId === "w1.lse_9" - ? { ownerId: "tok_agent", requesterId: "agent-1", workerId: "w1", workerLeaseId: "lse_9" } - : undefined, knowsWorker: () => true, relay: async (workerId, call) => { - relayed.push({ - leaseId: call.leaseId, - tenant: call.tenant, - url: call.request.url, - workerId, - }); + relayed.push({ tenant: call.tenant, url: call.request.url, workerId }); return answer(); }, requesterPrefix: "gw:inst:", @@ -69,42 +59,19 @@ async function refusalOf(served: Promise): Promise const signal = () => new AbortController().signal; describe("the gateway's stream link", () => { - it("relays to the worker holding the lease, by the worker's own lease id, as the lease's namespaced requester", async () => { + it("relays to the worker named, for the operator's namespaced principal, and describes the picture from its headers", async () => { const { port, relayed } = fleet(() => new Response("bytes", { headers: streamHeaders() })); - const stream = await createGatewayStreamLink({ fleet: port }).serve( - "w1.lse_9", - AGENT, - signal(), - ); + const stream = await createGatewayStreamLink({ fleet: port }).serve(DEVICE, OPERATOR, signal()); expect(relayed).toEqual([ { - leaseId: "lse_9", - tenant: "gw:inst:agent-1", - url: "http://simlock-worker/stream/leases/lse_9", + tenant: "gw:inst:tok_op", + url: "http://simlock-worker/stream/devices/dev_9", workerId: "w1", }, ]); expect(stream.meta).toEqual({ codec: "h264", fps: 30, height: 720, width: 330 }); }); - it("answers a lease the fleet does not hold with 404 UNKNOWN_LEASE, relaying nothing", async () => { - const { port, relayed } = fleet(() => new Response("bytes", { headers: streamHeaders() })); - const refusal = await refusalOf( - createGatewayStreamLink({ fleet: port }).serve("w1.lse_ended", AGENT, signal()), - ); - expect([refusal.status, refusal.code]).toEqual([404, "UNKNOWN_LEASE"]); - expect(relayed).toEqual([]); - }); - - it("answers a caller who does not own the lease exactly as an unknown lease, relaying nothing", async () => { - const { port, relayed } = fleet(() => new Response("bytes", { headers: streamHeaders() })); - const refusal = await refusalOf( - createGatewayStreamLink({ fleet: port }).serve("w1.lse_9", STRANGER, signal()), - ); - expect([refusal.status, refusal.code]).toEqual([404, "UNKNOWN_LEASE"]); - expect(relayed).toEqual([]); - }); - it("forwards the worker's own refusal code and status", async () => { const { port } = fleet(() => Response.json( @@ -113,7 +80,7 @@ describe("the gateway's stream link", () => { ), ); const refusal = await refusalOf( - createGatewayStreamLink({ fleet: port }).serve("w1.lse_9", AGENT, signal()), + createGatewayStreamLink({ fleet: port }).serve(DEVICE, OPERATOR, signal()), ); expect([refusal.status, refusal.code, refusal.message]).toEqual([ 415, @@ -127,7 +94,7 @@ describe("the gateway's stream link", () => { Response.json({ error: { code: "