diff --git a/apps/server/src/device/DeviceHubProxy.ts b/apps/server/src/device/DeviceHubProxy.ts index 7c58e6eab076b..8f3070231a37c 100644 --- a/apps/server/src/device/DeviceHubProxy.ts +++ b/apps/server/src/device/DeviceHubProxy.ts @@ -19,6 +19,7 @@ import * as NodeSocket from "@effect/platform-node/NodeSocket"; import { authenticateMediaRequest } from "../auth/http.ts"; import * as DeviceService from "./DeviceService.ts"; +import { repeatStillFrame } from "./mjpegStillFrame.ts"; const ALLOWED_PATHS: ReadonlyArray = [ /^\/api\/devices$/, @@ -131,7 +132,10 @@ const proxyHttp = Effect.fn("DeviceHubProxy.proxyHttp")(function* ( } // Long-lived MJPEG and AVCC responses must not be buffered by compression. headers["cache-control"] = "no-store, no-transform"; - return HttpServerResponse.stream(response.stream, { + const body = headers["content-type"]?.startsWith("multipart/x-mixed-replace") + ? repeatStillFrame(response.stream) + : response.stream; + return HttpServerResponse.stream(body, { status: response.status, headers, ...(headers["content-type"] ? { contentType: headers["content-type"] } : {}), diff --git a/apps/server/src/device/mjpegStillFrame.test.ts b/apps/server/src/device/mjpegStillFrame.test.ts new file mode 100644 index 0000000000000..8867dec3bf5a1 --- /dev/null +++ b/apps/server/src/device/mjpegStillFrame.test.ts @@ -0,0 +1,89 @@ +import { describe, expect, it } from "@effect/vitest"; +import * as Cause from "effect/Cause"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Queue from "effect/Queue"; +import * as Stream from "effect/Stream"; +import { TestClock } from "effect/testing"; + +import { STILL_FRAME_DELAY, repeatStillFrame } from "./mjpegStillFrame.ts"; + +const part = (body: string) => + Buffer.from( + `--frame\r\nContent-Type: image/jpeg\r\nContent-Length: ${body.length}\r\n\r\n${body}\r\n`, + ); + +const viewer = Effect.gen(function* () { + const upstream = yield* Queue.unbounded(); + let received = ""; + const fiber = yield* repeatStillFrame(Stream.fromQueue(upstream)).pipe( + Stream.runForEach((chunk) => + Effect.sync(() => { + received += Buffer.from(chunk).toString(); + }), + ), + Effect.forkChild, + ); + const send = (...chunks: ReadonlyArray) => + Effect.gen(function* () { + for (const chunk of chunks) yield* Queue.offer(upstream, chunk); + yield* Effect.yieldNow; + }); + const finish = Effect.gen(function* () { + yield* Queue.end(upstream); + yield* Fiber.join(fiber); + return received; + }); + return { send, finish }; +}); + +describe("repeatStillFrame", () => { + it.effect("re-sends only the newest complete frame once when the screen stays still", () => + Effect.gen(function* () { + const { send, finish } = yield* viewer; + yield* send(part("first")); + yield* TestClock.adjust(STILL_FRAME_DELAY); + yield* send(Buffer.concat([part("second"), part("third")])); + yield* TestClock.adjust(STILL_FRAME_DELAY); + yield* TestClock.adjust(STILL_FRAME_DELAY); + expect(yield* finish).toBe( + `${part("first")}${part("first")}${part("second")}${part("third")}${part("third")}`, + ); + }), + ); + + it.effect("never repeats into a frame that is still arriving", () => + Effect.gen(function* () { + const { send, finish } = yield* viewer; + const second = part("second"); + // The pause lands inside the second part's headers. + yield* send(part("first"), second.subarray(0, 20)); + yield* TestClock.adjust(STILL_FRAME_DELAY); + yield* send(second.subarray(20)); + yield* TestClock.adjust(STILL_FRAME_DELAY); + expect(yield* finish).toBe(`${part("first")}${second}${second}`); + }), + ); + + it.effect("recognizes a frame whose final CRLF arrives in a later chunk", () => + Effect.gen(function* () { + const { send, finish } = yield* viewer; + const first = part("first"); + yield* send(first.subarray(0, -1), first.subarray(-1)); + yield* TestClock.adjust(STILL_FRAME_DELAY); + expect(yield* finish).toBe(`${first}${first}`); + }), + ); + + it.effect("passes through bodies without the hub's part framing", () => + Effect.gen(function* () { + const { send, finish } = yield* viewer; + const unframed = Buffer.from("--frame\r\nContent-Type: image/jpeg\r\n\r\nbody\r\n"); + yield* send(unframed); + yield* TestClock.adjust(STILL_FRAME_DELAY); + yield* send(part("later")); + yield* TestClock.adjust(STILL_FRAME_DELAY); + expect(yield* finish).toBe(`${unframed}${part("later")}`); + }), + ); +}); diff --git a/apps/server/src/device/mjpegStillFrame.ts b/apps/server/src/device/mjpegStillFrame.ts new file mode 100644 index 0000000000000..4792bf083257d --- /dev/null +++ b/apps/server/src/device/mjpegStillFrame.ts @@ -0,0 +1,121 @@ +/** + * Browsers paint a multipart/x-mixed-replace part only once the next part begins, and the hub + * writes a part only when the screen changes. Left alone, an `` viewer never shows a still + * screen (clients such as the iOS app then give up waiting for the first frame) and always hides + * the last frame of a change. Re-sending the last part after a short quiet period makes it paint. + */ +import * as Arr from "effect/Array"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Option from "effect/Option"; +import * as Stream from "effect/Stream"; + +export const STILL_FRAME_DELAY = Duration.millis(100); + +const HEADER_END = Buffer.from("\r\n\r\n"); +const MAX_HEADER_BYTES = 16 * 1024; +// Simulator frames are far smaller; a larger claim is unknown framing, not something to retain. +const MAX_PART_BYTES = 16 * 1024 * 1024; + +type Part = Arr.NonEmptyReadonlyArray; + +const partLength = (headers: string) => { + if (!headers.startsWith("--")) return undefined; + const length = Number(/^content-length:\s*(\d+)\s*$/im.exec(headers)?.[1]); + return length <= MAX_PART_BYTES ? length : undefined; +}; + +const endsWithCrlf = (part: Part) => { + const last = Arr.lastNonEmpty(part); + const cr = last.length > 1 ? last.at(-2) : part.at(-2)?.at(-1); + return cr === 0x0d && last.at(-1) === 0x0a; +}; + +/** + * Follows the hub's part framing (`--frame`, `Content-Length`, CRLF CRLF, body, CRLF; each part + * is one write) and keeps the last complete part as views into the forwarded chunks, so frame + * bytes are never copied. Any other framing turns tracking off and the rest passes through. + */ +const partTracker = () => { + let pieces: Array = []; + let size = 0; + let partEnd: number | null = null; + let tracking = true; + return { + /** The last complete part, while the stream sits at a part boundary. */ + still: null as Part | null, + feed(chunk: Uint8Array) { + if (!tracking) return; + let rest = Buffer.from(chunk.buffer, chunk.byteOffset, chunk.byteLength); + let part: Part | undefined; + while (rest.length > 0) { + if (partEnd === null) { + // A header split across chunks is re-scanned per chunk. Only a header-sized + // prefix of the new chunk joins the scan, so frame bytes are never copied. + const head = + size === 0 + ? rest + : Buffer.concat([...pieces, rest.subarray(0, MAX_HEADER_BYTES + HEADER_END.length)]); + const headerEnd = head.indexOf(HEADER_END); + if (headerEnd < 0 && head.length <= MAX_HEADER_BYTES) { + pieces.push(rest); + size += rest.length; + break; + } + const length = + headerEnd < 0 || headerEnd > MAX_HEADER_BYTES + ? undefined + : partLength(head.subarray(0, headerEnd).toString("latin1")); + if (length === undefined) { + tracking = false; + break; + } + partEnd = headerEnd + HEADER_END.length + length + 2; + } + const taken = rest.subarray(0, partEnd - size); + pieces.push(taken); + size += taken.length; + rest = rest.subarray(taken.length); + if (size < partEnd) break; + const complete = pieces; + if (!Arr.isArrayNonEmpty(complete) || !endsWithCrlf(complete)) { + tracking = false; + break; + } + part = complete; + pieces = []; + size = 0; + partEnd = null; + } + if (!tracking) pieces = []; + this.still = tracking && size === 0 ? (part ?? this.still) : null; + }, + }; +}; + +/** Re-sends the latest frame once when the stream goes quiet between frames. */ +export const repeatStillFrame = (body: Stream.Stream) => + Stream.transformPull(body, (pull, scope) => + Effect.sync(() => { + const tracker = partTracker(); + let pending: Fiber.Fiber, Effect.Error> | null = + null; + const forward = (chunks: Effect.Success) => { + pending = null; + for (const chunk of chunks) tracker.feed(chunk); + return chunks; + }; + return Effect.gen(function* () { + const still = tracker.still; + if (still === null && pending === null) return forward(yield* pull); + pending ??= yield* Effect.forkIn(pull, scope); + if (still === null) return forward(yield* Fiber.join(pending)); + // Timing out the join leaves the pull running, so a frame arriving mid-wait is kept. + const next = yield* Fiber.join(pending).pipe(Effect.timeoutOption(STILL_FRAME_DELAY)); + if (Option.isSome(next)) return forward(next.value); + tracker.still = null; + return still; + }); + }), + ); diff --git a/docs/internals/devices.md b/docs/internals/devices.md index 7a082c17e150d..27c1628691b56 100644 --- a/docs/internals/devices.md +++ b/docs/internals/devices.md @@ -36,8 +36,12 @@ bearer and DPoP clients mint over authenticated HTTP. The ticket is stripped before the request reaches the hub. Stream responses carry `Cache-Control: no-transform`; the compression -middleware would otherwise buffer an MJPEG body that never ends. In browser dev, -the Vite proxy must forward WebSocket upgrades for `/api`, not only `/ws`. +middleware would otherwise buffer an MJPEG body that never ends. Browsers paint an +MJPEG part only once the next part begins and the hub writes one only when the +screen changes, so the proxy +[re-sends the last part](../../apps/server/src/device/mjpegStillFrame.ts) after a +short quiet period. In browser dev, the Vite proxy must forward WebSocket +upgrades for `/api`, not only `/ws`. ## Device settings never go through the hub