From c2182dda4abfbc921c924cb420c832a84b6d6b76 Mon Sep 17 00:00:00 2001 From: Juba Date: Tue, 11 Aug 2026 19:02:43 +0400 Subject: [PATCH 1/2] fix: await capture lifecycle and expose health --- .../Sources/SimNative/CaptureEngine.swift | 22 +- .../__tests__/device-session-health.test.ts | 163 ++++++++++++ .../src/__tests__/session-health.test.ts | 109 ++++++++ packages/serve-sim/src/device-session.ts | 234 +++++++++++++----- packages/serve-sim/src/middleware.ts | 6 +- packages/serve-sim/src/native.ts | 20 +- packages/serve-sim/src/session-health.ts | 163 ++++++++++++ 7 files changed, 641 insertions(+), 76 deletions(-) create mode 100644 packages/serve-sim/src/__tests__/device-session-health.test.ts create mode 100644 packages/serve-sim/src/__tests__/session-health.test.ts create mode 100644 packages/serve-sim/src/session-health.ts diff --git a/packages/serve-sim/Sources/SimNative/CaptureEngine.swift b/packages/serve-sim/Sources/SimNative/CaptureEngine.swift index fa541d89f..904b4ac22 100644 --- a/packages/serve-sim/Sources/SimNative/CaptureEngine.swift +++ b/packages/serve-sim/Sources/SimNative/CaptureEngine.swift @@ -94,8 +94,22 @@ actor CaptureEngine { // drop old frames if there's backpressure bufferingPolicy: .bufferingNewest(1) ) - try await frameCapture.start(deviceUDID: deviceUDID) { pixelBuffer, _ in - frameContinuation.yield(Frame(pixelBuffer: pixelBuffer)) + do { + try await frameCapture.start(deviceUDID: deviceUDID) { pixelBuffer, _ in + frameContinuation.yield(Frame(pixelBuffer: pixelBuffer)) + } + } catch { + frameContinuation.finish() + await frameCapture.stop() + if phase == .starting { phase = .unstarted } + throw error + } + // Actor methods are reentrant across the await above. A concurrent + // stop must win instead of letting a late start resurrect the session. + guard phase == .starting else { + frameContinuation.finish() + await frameCapture.stop() + throw CancellationError() } Task { for await frame in frames { @@ -173,10 +187,10 @@ actor CaptureEngine { } } - func stop() { + func stop() async { if phase == .stopped { return } phase = .stopped - Task { [frameCapture] in await frameCapture.stop() } + await frameCapture.stop() consumers.removeAll() } } diff --git a/packages/serve-sim/src/__tests__/device-session-health.test.ts b/packages/serve-sim/src/__tests__/device-session-health.test.ts new file mode 100644 index 000000000..d3db9af51 --- /dev/null +++ b/packages/serve-sim/src/__tests__/device-session-health.test.ts @@ -0,0 +1,163 @@ +import { afterEach, describe, expect, test } from "bun:test"; +import { createServer, type Server } from "http"; +import type { AddressInfo } from "net"; +import { + DeviceSession, + type DeviceSessionDependencies, +} from "../device-session"; + +const servers: Server[] = []; + +afterEach(async () => { + await Promise.all(servers.splice(0).map((server) => new Promise((resolve) => { + server.close(() => resolve()); + server.closeAllConnections(); + }))); +}); + +describe("DeviceSession lifecycle health", () => { + test("turns an async native start rejection into a 503 response", async () => { + const session = new DeviceSession("TEST-UDID", dependencies({ + start: async () => { throw new Error("Device not booted (state: Shutdown)"); }, + })); + const baseUrl = await serve(session); + + const stream = await fetch(`${baseUrl}/stream.mjpeg`); + expect(stream.status).toBe(503); + expect(await stream.json()).toEqual({ + error: "capture_unavailable", + message: "Device not booted (state: Shutdown)", + }); + + const health = await fetch(`${baseUrl}/health`); + expect(health.status).toBe(503); + expect((await health.json()).status).toBe("failed"); + }); + + test("retains failure diagnostics and allows a later startup retry", async () => { + let startCalls = 0; + const session = new DeviceSession("TEST-UDID", dependencies({ + start: async () => { + startCalls++; + if (startCalls === 1) throw new Error("Device not booted"); + }, + })); + + await expect(session.start()).rejects.toThrow("Device not booted"); + const baseUrl = await serve(session); + const failed = await fetch(`${baseUrl}/health`); + expect(failed.status).toBe(503); + expect((await failed.json()).status).toBe("failed"); + + await expect(session.start()).resolves.toBeUndefined(); + expect(startCalls).toBe(2); + }); + + test("closing during startup cannot resurrect the session", async () => { + let rejectStart!: (error: Error) => void; + let notifyStartCalled!: () => void; + let stopCalls = 0; + const pendingStart = new Promise((_resolve, reject) => { rejectStart = reject; }); + const startCalled = new Promise((resolve) => { notifyStartCalled = resolve; }); + const session = new DeviceSession("TEST-UDID", dependencies({ + start: () => { + notifyStartCalled(); + return pendingStart; + }, + stop: async () => { stopCalls++; }, + })); + + const starting = session.start(); + await startCalled; + const closing = session.close(); + rejectStart(new Error("Device not booted (state: Shutting Down)")); + + await expect(starting).rejects.toThrow("Shutting Down"); + await closing; + expect(stopCalls).toBe(1); + + const baseUrl = await serve(session); + const health = await fetch(`${baseUrl}/health`); + expect(health.status).toBe(503); + expect((await health.json()).status).toBe("stopped"); + }); + + test("becomes ready only after capture produces a frame", async () => { + let sharedFrame!: (frame: { data: Uint8Array; width: number; height: number }) => Promise; + const session = new DeviceSession("TEST-UDID", dependencies({ + subscribeMjpeg: async (callback) => { + sharedFrame ??= callback; + return async () => {}; + }, + })); + await session.start(); + const baseUrl = await serve(session); + + const starting = await fetch(`${baseUrl}/health`); + expect(starting.status).toBe(425); + expect((await starting.json()).status).toBe("starting"); + + await sharedFrame({ + data: new Uint8Array([0xff, 0xd8, 0xff, 0xd9]), + width: 1_206, + height: 2_622, + }); + const ready = await fetch(`${baseUrl}/health`); + expect(ready.status).toBe(200); + expect(await ready.json()).toMatchObject({ + status: "ok", + ready: true, + device: "TEST-UDID", + screen: { width: 1_206, height: 2_622, orientation: "portrait" }, + frames: { mjpeg: 1, avcc: 0, lastCodec: "mjpeg" }, + }); + + await session.close(); + }); + + test("reports native teardown errors without rejecting shutdown", async () => { + const session = new DeviceSession("TEST-UDID", dependencies({ + stop: async () => { throw new Error("native teardown failed"); }, + })); + + await expect(session.close()).resolves.toBeUndefined(); + }); +}); + +function dependencies( + captureOverrides: Partial = {}, +): DeviceSessionDependencies { + return { + capture: { + start: async () => {}, + stop: async () => {}, + subscribeMjpeg: async () => async () => {}, + subscribeAvcc: async () => async () => {}, + ...captureOverrides, + }, + hid: { + touch: async () => {}, + multiTouch: async () => {}, + button: async () => {}, + buttonHid: async () => {}, + key: async () => {}, + scroll: async () => {}, + digitalCrown: async () => {}, + orientation: async () => false, + memoryWarning: async () => {}, + softwareKeyboard: async () => {}, + caDebug: async () => false, + }, + }; +} + +async function serve(session: DeviceSession): Promise { + const server = createServer((req, res) => { + if (req.url === "/health") session.handleHealth(req, res); + else session.handleMjpeg(req, res); + }); + servers.push(server); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const { port } = server.address() as AddressInfo; + return `http://127.0.0.1:${port}`; +} diff --git a/packages/serve-sim/src/__tests__/session-health.test.ts b/packages/serve-sim/src/__tests__/session-health.test.ts new file mode 100644 index 000000000..f806e08b6 --- /dev/null +++ b/packages/serve-sim/src/__tests__/session-health.test.ts @@ -0,0 +1,109 @@ +import { describe, expect, test } from "bun:test"; +import { SessionHealth } from "../session-health"; + +describe("SessionHealth", () => { + test("reports starting until the first frame arrives", () => { + let now = 1_000; + const health = new SessionHealth("TEST-UDID", { + now: () => now, + startupTimeoutMs: 10_000, + staleAfterMs: 3_000, + }); + + health.markRunning(); + now = 2_000; + + expect(health.snapshot({ mjpeg: 0, avcc: 0, hid: 0 })).toEqual({ + status: "starting", + ready: false, + device: "TEST-UDID", + phase: "running", + startedAt: "1970-01-01T00:00:01.000Z", + checkedAt: "1970-01-01T00:00:02.000Z", + uptimeMs: 1_000, + screen: null, + clients: { mjpeg: 0, avcc: 0, hid: 0, total: 0 }, + frames: { + mjpeg: 0, + avcc: 0, + lastAt: null, + lastCodec: null, + staleForMs: null, + }, + error: null, + }); + expect(health.httpStatus()).toBe(425); + }); + + test("reports frame, screen, and client diagnostics when ready", () => { + let now = 10_000; + const health = new SessionHealth("TEST-UDID", { now: () => now }); + health.markRunning(); + + now = 10_250; + health.recordFrame("mjpeg", { width: 1_206, height: 2_622 }); + now = 10_300; + health.recordFrame("avcc", { width: 1_206, height: 2_622 }); + + const snapshot = health.snapshot({ mjpeg: 2, avcc: 1, hid: 3 }, "landscape_left"); + expect(snapshot.status).toBe("ok"); + expect(snapshot.ready).toBe(true); + expect(snapshot.screen).toEqual({ + width: 1_206, + height: 2_622, + orientation: "landscape_left", + }); + expect(snapshot.clients).toEqual({ mjpeg: 2, avcc: 1, hid: 3, total: 6 }); + expect(snapshot.frames).toEqual({ + mjpeg: 1, + avcc: 1, + lastAt: "1970-01-01T00:00:10.300Z", + lastCodec: "avcc", + staleForMs: 0, + }); + expect(health.httpStatus()).toBe(200); + }); + + test("distinguishes a stalled stream from a startup in progress", () => { + let now = 20_000; + const health = new SessionHealth("TEST-UDID", { + now: () => now, + startupTimeoutMs: 10_000, + staleAfterMs: 3_000, + }); + health.markRunning(); + + now = 30_001; + expect(health.snapshot({ mjpeg: 0, avcc: 0, hid: 0 }).status).toBe("stalled"); + expect(health.httpStatus()).toBe(503); + + health.recordFrame("mjpeg", { width: 100, height: 200 }); + expect(health.snapshot({ mjpeg: 0, avcc: 0, hid: 0 }).status).toBe("ok"); + + now = 33_001; + const stale = health.snapshot({ mjpeg: 0, avcc: 0, hid: 0 }); + expect(stale.status).toBe("stalled"); + expect(stale.frames.staleForMs).toBe(3_000); + }); + + test("preserves failure details and stopped state", () => { + let now = 40_000; + const failed = new SessionHealth("TEST-UDID", { now: () => now }); + failed.markRunning(); + now = 40_500; + failed.markFailed(new Error("capture subscription failed")); + + expect(failed.snapshot({ mjpeg: 0, avcc: 0, hid: 0 }).error).toEqual({ + message: "capture subscription failed", + at: "1970-01-01T00:00:40.500Z", + }); + expect(failed.snapshot({ mjpeg: 0, avcc: 0, hid: 0 }).status).toBe("failed"); + expect(failed.httpStatus()).toBe(503); + + const stopped = new SessionHealth("TEST-UDID", { now: () => now }); + stopped.markRunning(); + stopped.markStopped(); + expect(stopped.snapshot({ mjpeg: 0, avcc: 0, hid: 0 }).status).toBe("stopped"); + expect(stopped.httpStatus()).toBe(503); + }); +}); diff --git a/packages/serve-sim/src/device-session.ts b/packages/serve-sim/src/device-session.ts index 17f569dd5..bf0b8c996 100644 --- a/packages/serve-sim/src/device-session.ts +++ b/packages/serve-sim/src/device-session.ts @@ -7,7 +7,7 @@ * /stream.avcc length-prefixed AVCC envelopes (seed + decoder config replay) * /ws binary HID input protocol ([tag][JSON]) → NativeHid * /config { width, height, orientation } - * /health { status: "ok" } + * /health capture readiness + frame/client diagnostics * /ax axe-shaped accessibility JSON (one-shot) * /foreground { bundleId, pid } * @@ -24,6 +24,7 @@ import { type MjpegFrame, } from "./native"; import { eventLogEventForHidMessage, formatEventLogPoint, recordEventLogEvent, updateEventLogEvent } from "./event-log"; +import { SessionHealth } from "./session-health"; /** * Minimal WebSocket surface the HID input channel needs. Satisfied by both the @@ -123,11 +124,41 @@ function waitForDrain(res: ServerResponse): Promise { }); } +type CaptureTransport = Pick< + NativeCapture, + "start" | "stop" | "subscribeMjpeg" | "subscribeAvcc" +>; +type HidTransport = Pick< + NativeHid, + | "touch" + | "multiTouch" + | "button" + | "buttonHid" + | "key" + | "scroll" + | "digitalCrown" + | "orientation" + | "memoryWarning" + | "softwareKeyboard" + | "caDebug" +>; + +export interface DeviceSessionDependencies { + capture: CaptureTransport; + hid: HidTransport; + health?: SessionHealth; +} + +type Unsubscribe = () => void | Promise; + export class DeviceSession { - private readonly capture: NativeCapture; - private readonly hid: NativeHid; - private unsubscribeMjpeg?: () => void; - private phase: "unstarted" | "running" | "stopped" = "unstarted"; + private readonly capture: CaptureTransport; + private readonly hid: HidTransport; + private readonly health: SessionHealth; + private unsubscribeMjpeg?: Unsubscribe; + private phase: "unstarted" | "starting" | "running" | "failed" | "stopped" = "unstarted"; + private startPromise?: Promise; + private closePromise?: Promise; private width = 0; private height = 0; @@ -135,42 +166,82 @@ export class DeviceSession { private latestJpegBuffer: Buffer | null = null; private latestJpegLength = 0; + private readonly mjpegClients = new Set(); + private readonly avccClients = new Set(); private readonly hidSockets = new Set(); private touchGestureLog?: TouchGestureLog; - constructor(public readonly udid: string) { - this.hid = new NativeHid(udid); - this.capture = new NativeCapture(udid); + constructor(public readonly udid: string, dependencies?: DeviceSessionDependencies) { + this.hid = dependencies?.hid ?? new NativeHid(udid); + this.capture = dependencies?.capture ?? new NativeCapture(udid); + this.health = dependencies?.health ?? new SessionHealth(udid); } - /** Begin capture. Throws if the device isn't booted. Idempotent. */ - start(): void { - if (this.phase !== "unstarted") return; - this.capture.start(); - void (async () => { + /** Begin capture and retain one shared MJPEG subscription. Idempotent. */ + start(): Promise { + if (this.phase === "stopped") { + return Promise.reject(new Error(`Device session ${this.udid} is closed`)); + } + if (this.startPromise) return this.startPromise; + + this.phase = "starting"; + this.health.markStarting(); + const startPromise = (async () => { + await this.capture.start(); + if (this.isStopped()) throw new Error(`Device session ${this.udid} closed while starting`); + const unsubscribe = await this.capture.subscribeMjpeg((frame) => this.onSharedMjpegFrame(frame)); - if (this.phase === "running") { // only if someone hasn't already stopped the capture - this.unsubscribeMjpeg = unsubscribe; - } else { - unsubscribe(); + if (this.isStopped()) { + await unsubscribe(); + throw new Error(`Device session ${this.udid} closed while starting`); } - })(); - this.phase = "running"; + this.unsubscribeMjpeg = unsubscribe; + this.phase = "running"; + this.health.markRunning(); + })().catch((error) => { + if (this.phase === "starting") this.phase = "failed"; + this.health.markFailed(error); + if (this.startPromise === startPromise) this.startPromise = undefined; + throw error; + }); + // The registry starts sessions eagerly. Observe the rejection immediately + // so a shutdown race cannot become a process-fatal unhandled Promise. + void startPromise.catch(() => {}); + this.startPromise = startPromise; + return startPromise; } - close(): void { - if (this.phase !== "running") return; + private isStopped(): boolean { + return this.phase === "stopped"; + } + + close(): Promise { + if (this.closePromise) return this.closePromise; + this.phase = "stopped"; + this.health.markStopped(); for (const ws of this.hidSockets) ws.close(); - this.unsubscribeMjpeg?.(); + this.mjpegClients.clear(); + this.avccClients.clear(); this.hidSockets.clear(); - this.capture.stop(); - this.phase = "stopped"; + + const closePromise = (async () => { + if (this.unsubscribeMjpeg) await this.unsubscribeMjpeg(); + await this.capture.stop(); + })().catch((error) => { + console.error( + `[capture] teardown failed for ${this.udid}:`, + error instanceof Error ? error.message : error, + ); + }); + this.closePromise = closePromise; + return closePromise; } // ── Frame handling ─────────────────────────────────────────────────────── private async onSharedMjpegFrame(frame: MjpegFrame): Promise { const { width, height, data: jpeg } = frame; + this.health.recordFrame("mjpeg", { width, height }); if (width !== this.width || height !== this.height) { this.width = width; @@ -201,6 +272,12 @@ export class DeviceSession { // ── HTTP handlers ──────────────────────────────────────────────────────── handleMjpeg(req: IncomingMessage, res: ServerResponse): void { + void this.serveMjpeg(req, res).catch((error) => this.handleStreamFailure(res, error)); + } + + private async serveMjpeg(req: IncomingMessage, res: ServerResponse): Promise { + await this.start(); + if (res.writableEnded || res.destroyed) return; const raw = new URL(req.url ?? "", "http://x").searchParams.get("raw") === "1"; res.writeHead(200, { "Content-Type": raw ? "application/octet-stream" : "multipart/x-mixed-replace; boundary=frame", @@ -208,42 +285,53 @@ export class DeviceSession { Connection: "keep-alive", ...CORS, }); + this.trackStreamClient(this.mjpegClients, res); - void (async () => { - const latestJpeg = this.latestJpeg(); - if (latestJpeg) this.writeMjpegFrame(res, latestJpeg); // paint immediately - const unsubscribe = await this.capture.subscribeMjpeg(async (frame) => { - await waitForDrain(res); - this.writeMjpegFrame(res, frame.data); - }); - if (res.writableEnded) unsubscribe(); - res.on("close", unsubscribe); - res.on("error", unsubscribe); - })(); + const latestJpeg = this.latestJpeg(); + if (latestJpeg) this.writeMjpegFrame(res, latestJpeg); // paint immediately + const unsubscribe = await this.capture.subscribeMjpeg(async (frame) => { + await waitForDrain(res); + this.writeMjpegFrame(res, frame.data); + }); + if (res.writableEnded) { + await unsubscribe(); + return; + } + res.on("close", unsubscribe); + res.on("error", unsubscribe); } handleAvcc(_req: IncomingMessage, res: ServerResponse): void { + void this.serveAvcc(res).catch((error) => this.handleStreamFailure(res, error)); + } + + private async serveAvcc(res: ServerResponse): Promise { + await this.start(); + if (res.writableEnded || res.destroyed) return; res.writeHead(200, { "Content-Type": "application/octet-stream", "Cache-Control": "no-cache, no-store", Connection: "keep-alive", ...CORS, }); + this.trackStreamClient(this.avccClients, res); - void (async () => { - // Seed with the current screen; the per-client native AVCC subscription - // starts with its own decoder config and keyframe. - const latestJpeg = this.latestJpeg(); - if (latestJpeg) res.write(avccSeed(latestJpeg)); + // Seed with the current screen; the per-client native AVCC subscription + // starts with its own decoder config and keyframe. + const latestJpeg = this.latestJpeg(); + if (latestJpeg) res.write(avccSeed(latestJpeg)); - const unsubscribe = await this.capture.subscribeAvcc(async (frame) => { - await waitForDrain(res); - res.write(frame.data); - }); - if (res.writableEnded) unsubscribe(); - res.on("close", unsubscribe); - res.on("error", unsubscribe); - })(); + const unsubscribe = await this.capture.subscribeAvcc(async (frame) => { + this.health.recordFrame("avcc", frame); + await waitForDrain(res); + res.write(frame.data); + }); + if (res.writableEnded) { + await unsubscribe(); + return; + } + res.on("close", unsubscribe); + res.on("error", unsubscribe); } handleConfig(_req: IncomingMessage, res: ServerResponse): void { @@ -251,7 +339,12 @@ export class DeviceSession { } handleHealth(_req: IncomingMessage, res: ServerResponse): void { - this.sendJson(res, 200, { status: "ok" }); + const snapshot = this.health.snapshot({ + mjpeg: this.mjpegClients.size, + avcc: this.avccClients.size, + hid: this.hidSockets.size, + }, this.orientation); + this.sendJson(res, this.health.httpStatus(snapshot.status), snapshot); } handleAx(_req: IncomingMessage, res: ServerResponse): Promise { @@ -524,6 +617,30 @@ export class DeviceSession { for (const ws of this.hidSockets) ws.send(frame); } + private trackStreamClient(clients: Set, res: ServerResponse): void { + clients.add(res); + const release = () => { + clients.delete(res); + res.off("close", release); + res.off("error", release); + }; + res.once("close", release); + res.once("error", release); + } + + private handleStreamFailure(res: ServerResponse, error: unknown): void { + this.health.markFailed(error); + if (res.writableEnded || res.destroyed) return; + if (res.headersSent) { + res.destroy(error instanceof Error ? error : new Error(String(error))); + return; + } + this.sendJson(res, 503, { + error: "capture_unavailable", + message: error instanceof Error ? error.message : String(error), + }); + } + private sendJson(res: ServerResponse, status: number, body: unknown): void { this.sendJsonString(res, status, JSON.stringify(body)); } @@ -551,22 +668,21 @@ const sessions = new Map(); export function getDeviceSession(udid: string): DeviceSession { let session = sessions.get(udid); if (!session) { - session = new DeviceSession(udid); - try { - session.start(); - } catch (err) { - session.close(); - throw err; - } - sessions.set(udid, session); + const created = new DeviceSession(udid); + session = created; + sessions.set(udid, created); + // Preserve failed sessions so /health can report the native error. A later + // stream request retries start() on the same capture after the simulator is + // booted; closeDeviceSession still disposes it during an explicit shutdown. + void created.start().catch(() => {}); } return session; } -export function closeDeviceSession(udid: string): void { +export async function closeDeviceSession(udid: string): Promise { const session = sessions.get(udid); if (session) { - session.close(); sessions.delete(udid); + await session.close(); } } diff --git a/packages/serve-sim/src/middleware.ts b/packages/serve-sim/src/middleware.ts index 0ef6fbdc3..4afe9a1bb 100644 --- a/packages/serve-sim/src/middleware.ts +++ b/packages/serve-sim/src/middleware.ts @@ -345,7 +345,7 @@ export async function readServeSimStates(): Promise { state.device, state.pid, ); - closeDeviceSession(state.device); + await closeDeviceSession(state.device); } else { debugMw( "recycling stale helper pid=%d (device %s no longer booted)", @@ -1532,7 +1532,7 @@ export function simMiddleware(options?: SimMiddlewareOptions): SimMiddleware { req.on("data", (chunk: Buffer | string) => { body += typeof chunk === "string" ? chunk : chunk.toString(); }); - req.on("end", () => { + req.on("end", async () => { let udid = ""; try { udid = (JSON.parse(body) as ShutdownRequestBody).udid ?? ""; } catch {} if (!/^[0-9A-F]{8}-[0-9A-F]{4}-[0-9A-F]{4}-[0-9A-F]{4}-[0-9A-F]{12}$/i.test(udid)) { @@ -1543,7 +1543,7 @@ export function simMiddleware(options?: SimMiddlewareOptions): SimMiddleware { // Stop our own in-process capture for this device first (no-op if it // isn't streamed here). This frees the native session immediately // rather than waiting for the next poll's reaper to notice. - closeDeviceSession(udid); + await closeDeviceSession(udid); // Drop the snapshot so the next /grid/api call re-queries simctl // and prunes any helper bound to this now-shutdown device. bootedSnapshot = { at: 0, booted: null }; diff --git a/packages/serve-sim/src/native.ts b/packages/serve-sim/src/native.ts index ec4d60efa..338eac9a9 100644 --- a/packages/serve-sim/src/native.ts +++ b/packages/serve-sim/src/native.ts @@ -33,9 +33,9 @@ interface SimHIDHandle { } interface SimCaptureHandle { - start(): void; - stop(): void; - subscribe(codec: number, onFrame: RawFrameCallback): Promise<() => void>; + start(): Promise; + stop(): Promise; + subscribe(codec: number, onFrame: RawFrameCallback): Promise<() => void | Promise>; } interface NativeAddon { @@ -196,18 +196,18 @@ export class NativeCapture { this.handle = new (load().SimCapture)(udid); } - /** Begin capturing. Throws if the device isn't booted. */ - start(): void { - this.handle.start(); + /** Begin capturing. Rejects if the device isn't booted. */ + start(): Promise { + return this.handle.start(); } - subscribeMjpeg(onFrame: (frame: MjpegFrame) => Promise): Promise<() => void> { + subscribeMjpeg(onFrame: (frame: MjpegFrame) => Promise): Promise<() => void | Promise> { return this.handle.subscribe(CODEC_MJPEG, (data, width, height, _flags) => { return onFrame({ data, width, height }); }); } - subscribeAvcc(onFrame: (frame: AvccFrame) => Promise): Promise<() => void> { + subscribeAvcc(onFrame: (frame: AvccFrame) => Promise): Promise<() => void | Promise> { return this.handle.subscribe(CODEC_AVCC, (data, width, height, flags) => { return onFrame({ data, @@ -220,8 +220,8 @@ export class NativeCapture { } /** Halt frame production. Full teardown happens when this object is GC'd. */ - stop(): void { - this.handle.stop(); + stop(): Promise { + return this.handle.stop(); } } diff --git a/packages/serve-sim/src/session-health.ts b/packages/serve-sim/src/session-health.ts new file mode 100644 index 000000000..a7cc4faa6 --- /dev/null +++ b/packages/serve-sim/src/session-health.ts @@ -0,0 +1,163 @@ +export type SessionPhase = "unstarted" | "starting" | "running" | "failed" | "stopped"; +export type SessionHealthStatus = "starting" | "ok" | "stalled" | "failed" | "stopped"; +export type StreamCodec = "mjpeg" | "avcc"; + +export type SessionClientCounts = { + mjpeg: number; + avcc: number; + hid: number; +}; + +export type SessionHealthSnapshot = { + status: SessionHealthStatus; + ready: boolean; + device: string; + phase: SessionPhase; + startedAt: string | null; + checkedAt: string; + uptimeMs: number | null; + screen: { width: number; height: number; orientation: string } | null; + clients: SessionClientCounts & { total: number }; + frames: { + mjpeg: number; + avcc: number; + lastAt: string | null; + lastCodec: StreamCodec | null; + staleForMs: number | null; + }; + error: { message: string; at: string } | null; +}; + +type SessionHealthOptions = { + now?: () => number; + startupTimeoutMs?: number; + staleAfterMs?: number; +}; + +const DEFAULT_STARTUP_TIMEOUT_MS = 10_000; +const DEFAULT_STALE_AFTER_MS = 3_000; + +/** + * Small, deterministic state tracker for the native capture pipeline. + * + * The native framebuffer has a 5fps idle floor, so a running session that has + * not produced a frame within the stale window is genuinely unhealthy even + * when the simulator screen is static. + */ +export class SessionHealth { + private readonly now: () => number; + private readonly startupTimeoutMs: number; + private readonly staleAfterMs: number; + + private phase: SessionPhase = "unstarted"; + private startedAt: number | null = null; + private lastFrameAt: number | null = null; + private lastCodec: StreamCodec | null = null; + private width = 0; + private height = 0; + private mjpegFrames = 0; + private avccFrames = 0; + private error: { message: string; at: number } | null = null; + + constructor( + private readonly device: string, + options: SessionHealthOptions = {}, + ) { + this.now = options.now ?? Date.now; + this.startupTimeoutMs = options.startupTimeoutMs ?? DEFAULT_STARTUP_TIMEOUT_MS; + this.staleAfterMs = options.staleAfterMs ?? DEFAULT_STALE_AFTER_MS; + } + + markStarting(): void { + if (this.phase === "stopped") return; + if (this.startedAt === null) this.startedAt = this.now(); + this.phase = "starting"; + this.error = null; + } + + markRunning(): void { + if (this.phase === "stopped") return; + if (this.startedAt === null) this.startedAt = this.now(); + this.phase = "running"; + this.error = null; + } + + recordFrame(codec: StreamCodec, screen: { width: number; height: number }): void { + if (this.phase !== "starting" && this.phase !== "running") return; + if (codec === "mjpeg") this.mjpegFrames++; + else this.avccFrames++; + this.lastFrameAt = this.now(); + this.lastCodec = codec; + this.width = screen.width; + this.height = screen.height; + } + + markFailed(error: unknown): void { + if (this.phase === "stopped") return; + const at = this.now(); + this.phase = "failed"; + this.error = { + message: error instanceof Error ? error.message : String(error), + at, + }; + } + + markStopped(): void { + this.phase = "stopped"; + } + + httpStatus(status = this.currentStatus(this.now())): number { + switch (status) { + case "ok": return 200; + case "starting": return 425; + case "stalled": + case "failed": + case "stopped": return 503; + } + } + + snapshot(clients: SessionClientCounts, orientation = "portrait"): SessionHealthSnapshot { + const now = this.now(); + const status = this.currentStatus(now); + const staleForMs = this.lastFrameAt === null ? null : Math.max(0, now - this.lastFrameAt); + return { + status, + ready: status === "ok", + device: this.device, + phase: this.phase, + startedAt: this.startedAt === null ? null : new Date(this.startedAt).toISOString(), + checkedAt: new Date(now).toISOString(), + uptimeMs: this.startedAt === null ? null : Math.max(0, now - this.startedAt), + screen: this.width > 0 && this.height > 0 + ? { width: this.width, height: this.height, orientation } + : null, + clients: { + ...clients, + total: clients.mjpeg + clients.avcc + clients.hid, + }, + frames: { + mjpeg: this.mjpegFrames, + avcc: this.avccFrames, + lastAt: this.lastFrameAt === null ? null : new Date(this.lastFrameAt).toISOString(), + lastCodec: this.lastCodec, + staleForMs, + }, + error: this.error + ? { message: this.error.message, at: new Date(this.error.at).toISOString() } + : null, + }; + } + + private currentStatus(now: number): SessionHealthStatus { + if (this.phase === "failed") return "failed"; + if (this.phase === "stopped") return "stopped"; + if (this.phase === "unstarted") return "starting"; + + if (this.lastFrameAt === null) { + const startingForMs = this.startedAt === null ? 0 : now - this.startedAt; + return startingForMs > this.startupTimeoutMs ? "stalled" : "starting"; + } + if (this.phase === "starting") return "starting"; + return now - this.lastFrameAt >= this.staleAfterMs ? "stalled" : "ok"; + } +} From fbb950dc327309810552a981f4be2b248ed355c8 Mon Sep 17 00:00:00 2001 From: Juba Date: Wed, 12 Aug 2026 13:47:47 +0400 Subject: [PATCH 2/2] fix: harden stream cleanup and retry health --- .../__tests__/device-session-health.test.ts | 69 ++++++++++++++++++- .../src/__tests__/session-health.test.ts | 28 ++++++++ packages/serve-sim/src/device-session.ts | 41 ++++++++--- packages/serve-sim/src/session-health.ts | 8 ++- 4 files changed, 136 insertions(+), 10 deletions(-) diff --git a/packages/serve-sim/src/__tests__/device-session-health.test.ts b/packages/serve-sim/src/__tests__/device-session-health.test.ts index d3db9af51..5a646c332 100644 --- a/packages/serve-sim/src/__tests__/device-session-health.test.ts +++ b/packages/serve-sim/src/__tests__/device-session-health.test.ts @@ -1,5 +1,5 @@ import { afterEach, describe, expect, test } from "bun:test"; -import { createServer, type Server } from "http"; +import { createServer, get, type Server } from "http"; import type { AddressInfo } from "net"; import { DeviceSession, @@ -122,6 +122,50 @@ describe("DeviceSession lifecycle health", () => { await expect(session.close()).resolves.toBeUndefined(); }); + + test("closes stream clients and observes each async unsubscribe once", async () => { + let sharedFrame!: (frame: { data: Uint8Array; width: number; height: number }) => Promise; + let mjpegSubscriptions = 0; + let mjpegClientUnsubscribes = 0; + let avccClientUnsubscribes = 0; + const session = new DeviceSession("TEST-UDID", dependencies({ + subscribeMjpeg: async (callback) => { + mjpegSubscriptions++; + if (mjpegSubscriptions === 1) { + sharedFrame = callback; + return async () => {}; + } + return async () => { + mjpegClientUnsubscribes++; + throw new Error("mjpeg unsubscribe failed"); + }; + }, + subscribeAvcc: async () => async () => { + avccClientUnsubscribes++; + throw new Error("avcc unsubscribe failed"); + }, + })); + await session.start(); + await sharedFrame({ + data: new Uint8Array([0xff, 0xd8, 0xff, 0xd9]), + width: 1_206, + height: 2_622, + }); + const baseUrl = await serve(session); + const mjpeg = observeStream(`${baseUrl}/stream.mjpeg`); + const avcc = observeStream(`${baseUrl}/stream.avcc`); + await Promise.all([mjpeg.opened, avcc.opened]); + + const health = await fetch(`${baseUrl}/health`); + expect(await health.json()).toMatchObject({ + clients: { mjpeg: 1, avcc: 1, total: 2 }, + }); + + await session.close(); + expect(await resolvesWithin(Promise.all([mjpeg.closed, avcc.closed]), 1_000)).toBe(true); + expect(mjpegClientUnsubscribes).toBe(1); + expect(avccClientUnsubscribes).toBe(1); + }); }); function dependencies( @@ -154,6 +198,7 @@ function dependencies( async function serve(session: DeviceSession): Promise { const server = createServer((req, res) => { if (req.url === "/health") session.handleHealth(req, res); + else if (req.url === "/stream.avcc") session.handleAvcc(req, res); else session.handleMjpeg(req, res); }); servers.push(server); @@ -161,3 +206,25 @@ async function serve(session: DeviceSession): Promise { const { port } = server.address() as AddressInfo; return `http://127.0.0.1:${port}`; } + +function observeStream(url: string): { opened: Promise; closed: Promise } { + let resolveClosed!: () => void; + const closed = new Promise((resolve) => { resolveClosed = resolve; }); + const opened = new Promise((resolve, reject) => { + const request = get(url, (response) => { + response.on("error", () => {}); + response.once("close", resolveClosed); + response.resume(); + resolve(); + }); + request.once("error", reject); + }); + return { opened, closed }; +} + +async function resolvesWithin(promise: Promise, timeoutMs: number): Promise { + return Promise.race([ + promise.then(() => true), + Bun.sleep(timeoutMs).then(() => false), + ]); +} diff --git a/packages/serve-sim/src/__tests__/session-health.test.ts b/packages/serve-sim/src/__tests__/session-health.test.ts index f806e08b6..bcb23a1e9 100644 --- a/packages/serve-sim/src/__tests__/session-health.test.ts +++ b/packages/serve-sim/src/__tests__/session-health.test.ts @@ -106,4 +106,32 @@ describe("SessionHealth", () => { expect(stopped.snapshot({ mjpeg: 0, avcc: 0, hid: 0 }).status).toBe("stopped"); expect(stopped.httpStatus()).toBe(503); }); + + test("restarts startup timing and clears stale frame state on retry", () => { + let now = 50_000; + const health = new SessionHealth("TEST-UDID", { + now: () => now, + startupTimeoutMs: 10_000, + }); + health.markRunning(); + now = 50_500; + health.recordFrame("mjpeg", { width: 100, height: 200 }); + now = 51_000; + health.markFailed(new Error("capture failed")); + + now = 70_000; + health.markStarting(); + const retrying = health.snapshot({ mjpeg: 0, avcc: 0, hid: 0 }); + expect(retrying.status).toBe("starting"); + expect(retrying.startedAt).toBe("1970-01-01T00:01:10.000Z"); + expect(retrying.frames).toMatchObject({ + mjpeg: 1, + lastAt: null, + lastCodec: null, + staleForMs: null, + }); + + now = 80_001; + expect(health.snapshot({ mjpeg: 0, avcc: 0, hid: 0 }).status).toBe("stalled"); + }); }); diff --git a/packages/serve-sim/src/device-session.ts b/packages/serve-sim/src/device-session.ts index bf0b8c996..be4ce24f4 100644 --- a/packages/serve-sim/src/device-session.ts +++ b/packages/serve-sim/src/device-session.ts @@ -220,6 +220,9 @@ export class DeviceSession { this.phase = "stopped"; this.health.markStopped(); for (const ws of this.hidSockets) ws.close(); + for (const res of [...this.mjpegClients, ...this.avccClients]) { + res.destroy(); + } this.mjpegClients.clear(); this.avccClients.clear(); this.hidSockets.clear(); @@ -293,12 +296,11 @@ export class DeviceSession { await waitForDrain(res); this.writeMjpegFrame(res, frame.data); }); - if (res.writableEnded) { - await unsubscribe(); + if (res.writableEnded || res.destroyed) { + await this.runStreamUnsubscribe(unsubscribe); return; } - res.on("close", unsubscribe); - res.on("error", unsubscribe); + this.releaseStreamSubscription(res, unsubscribe); } handleAvcc(_req: IncomingMessage, res: ServerResponse): void { @@ -326,12 +328,11 @@ export class DeviceSession { await waitForDrain(res); res.write(frame.data); }); - if (res.writableEnded) { - await unsubscribe(); + if (res.writableEnded || res.destroyed) { + await this.runStreamUnsubscribe(unsubscribe); return; } - res.on("close", unsubscribe); - res.on("error", unsubscribe); + this.releaseStreamSubscription(res, unsubscribe); } handleConfig(_req: IncomingMessage, res: ServerResponse): void { @@ -628,6 +629,30 @@ export class DeviceSession { res.once("error", release); } + private releaseStreamSubscription(res: ServerResponse, unsubscribe: Unsubscribe): void { + let released = false; + const release = () => { + if (released) return; + released = true; + res.off("close", release); + res.off("error", release); + void this.runStreamUnsubscribe(unsubscribe); + }; + res.once("close", release); + res.once("error", release); + } + + private async runStreamUnsubscribe(unsubscribe: Unsubscribe): Promise { + try { + await unsubscribe(); + } catch (error) { + console.error( + `[capture] stream unsubscribe failed for ${this.udid}:`, + error instanceof Error ? error.message : error, + ); + } + } + private handleStreamFailure(res: ServerResponse, error: unknown): void { this.health.markFailed(error); if (res.writableEnded || res.destroyed) return; diff --git a/packages/serve-sim/src/session-health.ts b/packages/serve-sim/src/session-health.ts index a7cc4faa6..351cc3d36 100644 --- a/packages/serve-sim/src/session-health.ts +++ b/packages/serve-sim/src/session-health.ts @@ -70,7 +70,13 @@ export class SessionHealth { markStarting(): void { if (this.phase === "stopped") return; - if (this.startedAt === null) this.startedAt = this.now(); + if (this.phase === "failed") { + this.startedAt = this.now(); + this.lastFrameAt = null; + this.lastCodec = null; + } else if (this.startedAt === null) { + this.startedAt = this.now(); + } this.phase = "starting"; this.error = null; }