From 37b1b120ab8d79dddc0dd6681731fc075a4e2942 Mon Sep 17 00:00:00 2001 From: Joaquin Alfian Date: Sat, 17 Oct 2026 16:11:56 -0400 Subject: [PATCH 1/6] fix(server): iOS device viewer no longer times out on a still simulator screen Browsers only paint a multipart/x-mixed-replace part once the next part starts, and the device hub writes a frame only when the screen changes. The iOS app views the simulator over MJPEG, so a still screen never appeared and the viewer failed after 15 seconds with "No video received". The device hub proxy now re-sends the latest complete MJPEG part once after 100 ms of quiet. It follows the hub's Content-Length framing, never injects bytes mid-part, and passes other bodies through untouched. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/device/DeviceHubProxy.ts | 6 +- .../server/src/device/mjpegStillFrame.test.ts | 90 +++++++++++++++ apps/server/src/device/mjpegStillFrame.ts | 108 ++++++++++++++++++ docs/internals/devices.md | 6 +- 4 files changed, 208 insertions(+), 2 deletions(-) create mode 100644 apps/server/src/device/mjpegStillFrame.test.ts create mode 100644 apps/server/src/device/mjpegStillFrame.ts diff --git a/apps/server/src/device/DeviceHubProxy.ts b/apps/server/src/device/DeviceHubProxy.ts index a00c49a424b8..b95447c8447a 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$/, @@ -127,7 +128,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 000000000000..59d0e3bd146a --- /dev/null +++ b/apps/server/src/device/mjpegStillFrame.test.ts @@ -0,0 +1,90 @@ +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 the latest 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* TestClock.adjust(STILL_FRAME_DELAY); + expect(yield* finish).toBe(`${part("first")}${part("first")}`); + }), + ); + + it.effect("never interleaves a repeat into a frame that is still arriving", () => + Effect.gen(function* () { + const { send, finish } = yield* viewer; + const second = part("second"); + 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("adds nothing while frames keep arriving", () => + Effect.gen(function* () { + const { send, finish } = yield* viewer; + yield* send(part("first")); + yield* send(part("second")); + expect(yield* finish).toBe(`${part("first")}${part("second")}`); + }), + ); + + it.effect("repeats only the newest of several frames in one chunk", () => + Effect.gen(function* () { + const { send, finish } = yield* viewer; + yield* send(Buffer.concat([part("first"), part("second")])); + yield* TestClock.adjust(STILL_FRAME_DELAY); + expect(yield* finish).toBe(`${part("first")}${part("second")}${part("second")}`); + }), + ); + + 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); + expect(yield* finish).toBe(String(unframed)); + }), + ); +}); diff --git a/apps/server/src/device/mjpegStillFrame.ts b/apps/server/src/device/mjpegStillFrame.ts new file mode 100644 index 000000000000..16b55e8186f4 --- /dev/null +++ b/apps/server/src/device/mjpegStillFrame.ts @@ -0,0 +1,108 @@ +/** + * Browsers paint a multipart/x-mixed-replace part only after the next part begins, and the hub + * writes a frame only when the screen changes. Without help, an `` viewer shows nothing + * for a still screen and always hides the last frame of a change. Clients that view MJPEG, such + * as the iOS app, then time out waiting for the first frame. + */ +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; + +type Part = Arr.NonEmptyReadonlyArray; + +const endsWithCrlf = (part: Part) => { + const tail = Buffer.concat(part.slice(-2)).subarray(-2); + return tail[0] === 0x0d && tail[1] === 0x0a; +}; + +/** + * Follows the hub's part framing (`--boundary`, headers with Content-Length, body, CRLF) and + * keeps the last complete part as views into the forwarded chunks, so frames are never copied. + * Any other framing turns it off and the stream passes through untouched. + */ +const partTracker = () => { + let pieces: Array = []; + let size = 0; + let partEnd: number | null = null; + let recognized = true; + return (chunk: Uint8Array): { readonly atBoundary: boolean; readonly part?: Part } => { + if (!recognized) return { atBoundary: false }; + let rest = Buffer.from(chunk.buffer, chunk.byteOffset, chunk.byteLength); + let part: Part | undefined; + while (rest.length > 0) { + if (partEnd === null) { + const head = size === 0 ? rest : Buffer.concat([...pieces, rest]); + const headerEnd = head.indexOf(HEADER_END); + if (headerEnd < 0) { + if (head.length > MAX_HEADER_BYTES) recognized = false; + pieces.push(rest); + size += rest.length; + break; + } + const headers = head.subarray(0, headerEnd).toString("latin1"); + const length = /^content-length:\s*(\d+)\s*$/im.exec(headers)?.[1]; + if (!headers.startsWith("--") || length === undefined) { + recognized = false; + break; + } + partEnd = headerEnd + HEADER_END.length + Number(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)) { + recognized = false; + break; + } + part = complete; + pieces = []; + size = 0; + partEnd = null; + } + if (!recognized) pieces = []; + return { atBoundary: recognized && size === 0, ...(part ? { part } : {}) }; + }; +}; + +/** 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 track = partTracker(); + let pending: Fiber.Fiber, Effect.Error> | null = + null; + let still: Part | null = null; + const forward = (chunks: Effect.Success) => { + pending = null; + for (const chunk of chunks) { + const { atBoundary, part } = track(chunk); + if (!atBoundary) still = null; + else if (part) still = part; + } + return chunks; + }; + return Effect.gen(function* () { + pending ??= yield* Effect.forkIn(pull, scope); + if (still) { + const repeat = still; + // Timing out the join leaves the pull running, so the next frame is never lost. + const next = yield* Fiber.join(pending).pipe(Effect.timeoutOption(STILL_FRAME_DELAY)); + if (Option.isSome(next)) return forward(next.value); + still = null; + return repeat; + } + return forward(yield* Fiber.join(pending)); + }); + }), + ); diff --git a/docs/internals/devices.md b/docs/internals/devices.md index 7a082c17e150..d9f29b4f2848 100644 --- a/docs/internals/devices.md +++ b/docs/internals/devices.md @@ -36,7 +36,11 @@ 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, +middleware would otherwise buffer an MJPEG body that never ends. Browsers paint an +MJPEG part only once the next part starts, and the hub writes a frame only when the +screen changes, so the proxy +[re-sends the last frame](../../apps/server/src/device/mjpegStillFrame.ts) after a +short quiet period; without it a still screen never appears. In browser dev, the Vite proxy must forward WebSocket upgrades for `/api`, not only `/ws`. ## Device settings never go through the hub From fb86fb91e4c1a5ba8cba0161c768908723b29808 Mon Sep 17 00:00:00 2001 From: Joaquin Alfian Date: Sat, 17 Oct 2026 16:35:06 -0400 Subject: [PATCH 2/6] fix(server): check the MJPEG part's trailing CRLF without copying the frame Co-Authored-By: Claude Opus 5.5 --- apps/server/src/device/mjpegStillFrame.test.ts | 10 ++++++++++ apps/server/src/device/mjpegStillFrame.ts | 5 +++-- 2 files changed, 13 insertions(+), 2 deletions(-) diff --git a/apps/server/src/device/mjpegStillFrame.test.ts b/apps/server/src/device/mjpegStillFrame.test.ts index 59d0e3bd146a..7f159f05ce61 100644 --- a/apps/server/src/device/mjpegStillFrame.test.ts +++ b/apps/server/src/device/mjpegStillFrame.test.ts @@ -69,6 +69,16 @@ describe("repeatStillFrame", () => { }), ); + it.effect("recognizes a frame whose final CRLF is split across chunks", () => + 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("repeats only the newest of several frames in one chunk", () => Effect.gen(function* () { const { send, finish } = yield* viewer; diff --git a/apps/server/src/device/mjpegStillFrame.ts b/apps/server/src/device/mjpegStillFrame.ts index 16b55e8186f4..1c8be6226e66 100644 --- a/apps/server/src/device/mjpegStillFrame.ts +++ b/apps/server/src/device/mjpegStillFrame.ts @@ -19,8 +19,9 @@ const MAX_HEADER_BYTES = 16 * 1024; type Part = Arr.NonEmptyReadonlyArray; const endsWithCrlf = (part: Part) => { - const tail = Buffer.concat(part.slice(-2)).subarray(-2); - return tail[0] === 0x0d && tail[1] === 0x0a; + 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; }; /** From beeeb1247235295d342292b029da0016f130734e Mon Sep 17 00:00:00 2001 From: Joaquin Alfian Date: Sat, 17 Oct 2026 16:36:09 -0400 Subject: [PATCH 3/6] fix(server): stop tracking MJPEG parts that claim more than 16 MiB Co-Authored-By: Claude Opus 5.5 --- apps/server/src/device/mjpegStillFrame.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/apps/server/src/device/mjpegStillFrame.ts b/apps/server/src/device/mjpegStillFrame.ts index 1c8be6226e66..081ff8764333 100644 --- a/apps/server/src/device/mjpegStillFrame.ts +++ b/apps/server/src/device/mjpegStillFrame.ts @@ -15,6 +15,8 @@ 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 stops tracking instead of retaining it. +const MAX_PART_BYTES = 16 * 1024 * 1024; type Part = Arr.NonEmptyReadonlyArray; @@ -50,7 +52,7 @@ const partTracker = () => { } const headers = head.subarray(0, headerEnd).toString("latin1"); const length = /^content-length:\s*(\d+)\s*$/im.exec(headers)?.[1]; - if (!headers.startsWith("--") || length === undefined) { + if (!headers.startsWith("--") || length === undefined || Number(length) > MAX_PART_BYTES) { recognized = false; break; } From 482230446aa81afd8a5ad2ee647e3d294a018c02 Mon Sep 17 00:00:00 2001 From: Joaquin Alfian Date: Sat, 17 Oct 2026 16:42:50 -0400 Subject: [PATCH 4/6] fix(server): stop tracking MJPEG parts whose headers exceed 16 KiB Co-Authored-By: Claude Opus 5.5 --- apps/server/src/device/mjpegStillFrame.test.ts | 12 ++++++++++++ apps/server/src/device/mjpegStillFrame.ts | 4 ++++ 2 files changed, 16 insertions(+) diff --git a/apps/server/src/device/mjpegStillFrame.test.ts b/apps/server/src/device/mjpegStillFrame.test.ts index 7f159f05ce61..bed6b137e7bc 100644 --- a/apps/server/src/device/mjpegStillFrame.test.ts +++ b/apps/server/src/device/mjpegStillFrame.test.ts @@ -88,6 +88,18 @@ describe("repeatStillFrame", () => { }), ); + it.effect("stops tracking a part whose headers exceed the header budget", () => + Effect.gen(function* () { + const { send, finish } = yield* viewer; + const padded = Buffer.from( + `--frame\r\nX-Padding: ${"a".repeat(32 * 1024)}\r\nContent-Length: 4\r\n\r\nbody\r\n`, + ); + yield* send(padded); + yield* TestClock.adjust(STILL_FRAME_DELAY); + expect(yield* finish).toBe(String(padded)); + }), + ); + it.effect("passes through bodies without the hub's part framing", () => Effect.gen(function* () { const { send, finish } = yield* viewer; diff --git a/apps/server/src/device/mjpegStillFrame.ts b/apps/server/src/device/mjpegStillFrame.ts index 081ff8764333..0c66d7a8d476 100644 --- a/apps/server/src/device/mjpegStillFrame.ts +++ b/apps/server/src/device/mjpegStillFrame.ts @@ -50,6 +50,10 @@ const partTracker = () => { size += rest.length; break; } + if (headerEnd > MAX_HEADER_BYTES) { + recognized = false; + break; + } const headers = head.subarray(0, headerEnd).toString("latin1"); const length = /^content-length:\s*(\d+)\s*$/im.exec(headers)?.[1]; if (!headers.startsWith("--") || length === undefined || Number(length) > MAX_PART_BYTES) { From 454b58ec8c0eac222f42c4842327c58b4eb47f68 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 02:54:55 +0000 Subject: [PATCH 5/6] fix(server): still simulator screens repeat their last MJPEG frame Tightens the part tracker to one give-up path and lets it own the still frame slot; trims the tests to the three behaviors a viewer notices. Co-Authored-By: Claude Fable 5.1 --- .../server/src/device/mjpegStillFrame.test.ts | 55 ++------ apps/server/src/device/mjpegStillFrame.ts | 129 +++++++++--------- docs/internals/devices.md | 8 +- 3 files changed, 80 insertions(+), 112 deletions(-) diff --git a/apps/server/src/device/mjpegStillFrame.test.ts b/apps/server/src/device/mjpegStillFrame.test.ts index bed6b137e7bc..8739ffb3ff83 100644 --- a/apps/server/src/device/mjpegStillFrame.test.ts +++ b/apps/server/src/device/mjpegStillFrame.test.ts @@ -38,20 +38,25 @@ const viewer = Effect.gen(function* () { }); describe("repeatStillFrame", () => { - it.effect("re-sends the latest frame once when the screen stays still", () => + 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); - expect(yield* finish).toBe(`${part("first")}${part("first")}`); + yield* TestClock.adjust(STILL_FRAME_DELAY); + expect(yield* finish).toBe( + `${part("first")}${part("first")}${part("second")}${part("third")}${part("third")}`, + ); }), ); - it.effect("never interleaves a repeat into a frame that is still arriving", () => + 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)); @@ -60,53 +65,15 @@ describe("repeatStillFrame", () => { }), ); - it.effect("adds nothing while frames keep arriving", () => - Effect.gen(function* () { - const { send, finish } = yield* viewer; - yield* send(part("first")); - yield* send(part("second")); - expect(yield* finish).toBe(`${part("first")}${part("second")}`); - }), - ); - - it.effect("recognizes a frame whose final CRLF is split across chunks", () => - 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("repeats only the newest of several frames in one chunk", () => - Effect.gen(function* () { - const { send, finish } = yield* viewer; - yield* send(Buffer.concat([part("first"), part("second")])); - yield* TestClock.adjust(STILL_FRAME_DELAY); - expect(yield* finish).toBe(`${part("first")}${part("second")}${part("second")}`); - }), - ); - - it.effect("stops tracking a part whose headers exceed the header budget", () => - Effect.gen(function* () { - const { send, finish } = yield* viewer; - const padded = Buffer.from( - `--frame\r\nX-Padding: ${"a".repeat(32 * 1024)}\r\nContent-Length: 4\r\n\r\nbody\r\n`, - ); - yield* send(padded); - yield* TestClock.adjust(STILL_FRAME_DELAY); - expect(yield* finish).toBe(String(padded)); - }), - ); - 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); - expect(yield* finish).toBe(String(unframed)); + 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 index 0c66d7a8d476..8faffcd75cf6 100644 --- a/apps/server/src/device/mjpegStillFrame.ts +++ b/apps/server/src/device/mjpegStillFrame.ts @@ -1,8 +1,8 @@ /** - * Browsers paint a multipart/x-mixed-replace part only after the next part begins, and the hub - * writes a frame only when the screen changes. Without help, an `` viewer shows nothing - * for a still screen and always hides the last frame of a change. Clients that view MJPEG, such - * as the iOS app, then time out waiting for the first frame. + * 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"; @@ -15,11 +15,17 @@ 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 stops tracking instead of retaining it. +// 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); @@ -27,58 +33,60 @@ const endsWithCrlf = (part: Part) => { }; /** - * Follows the hub's part framing (`--boundary`, headers with Content-Length, body, CRLF) and - * keeps the last complete part as views into the forwarded chunks, so frames are never copied. - * Any other framing turns it off and the stream passes through untouched. + * 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 recognized = true; - return (chunk: Uint8Array): { readonly atBoundary: boolean; readonly part?: Part } => { - if (!recognized) return { atBoundary: false }; - let rest = Buffer.from(chunk.buffer, chunk.byteOffset, chunk.byteLength); - let part: Part | undefined; - while (rest.length > 0) { - if (partEnd === null) { - const head = size === 0 ? rest : Buffer.concat([...pieces, rest]); - const headerEnd = head.indexOf(HEADER_END); - if (headerEnd < 0) { - if (head.length > MAX_HEADER_BYTES) recognized = false; - pieces.push(rest); - size += rest.length; - break; - } - if (headerEnd > MAX_HEADER_BYTES) { - recognized = false; - break; + 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, bounded by MAX_HEADER_BYTES. + const head = size === 0 ? rest : Buffer.concat([...pieces, rest]); + 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 headers = head.subarray(0, headerEnd).toString("latin1"); - const length = /^content-length:\s*(\d+)\s*$/im.exec(headers)?.[1]; - if (!headers.startsWith("--") || length === undefined || Number(length) > MAX_PART_BYTES) { - recognized = false; + 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; } - partEnd = headerEnd + HEADER_END.length + Number(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)) { - recognized = false; - break; + part = complete; + pieces = []; + size = 0; + partEnd = null; } - part = complete; - pieces = []; - size = 0; - partEnd = null; - } - if (!recognized) pieces = []; - return { atBoundary: recognized && size === 0, ...(part ? { part } : {}) }; + if (!tracking) pieces = []; + this.still = tracking && size === 0 ? (part ?? this.still) : null; + }, }; }; @@ -86,30 +94,23 @@ const partTracker = () => { export const repeatStillFrame = (body: Stream.Stream) => Stream.transformPull(body, (pull, scope) => Effect.sync(() => { - const track = partTracker(); + const tracker = partTracker(); let pending: Fiber.Fiber, Effect.Error> | null = null; - let still: Part | null = null; const forward = (chunks: Effect.Success) => { pending = null; - for (const chunk of chunks) { - const { atBoundary, part } = track(chunk); - if (!atBoundary) still = null; - else if (part) still = part; - } + for (const chunk of chunks) tracker.feed(chunk); return chunks; }; return Effect.gen(function* () { pending ??= yield* Effect.forkIn(pull, scope); - if (still) { - const repeat = still; - // Timing out the join leaves the pull running, so the next frame is never lost. - const next = yield* Fiber.join(pending).pipe(Effect.timeoutOption(STILL_FRAME_DELAY)); - if (Option.isSome(next)) return forward(next.value); - still = null; - return repeat; - } - return forward(yield* Fiber.join(pending)); + const still = tracker.still; + 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 d9f29b4f2848..27c1628691b5 100644 --- a/docs/internals/devices.md +++ b/docs/internals/devices.md @@ -37,11 +37,11 @@ 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. Browsers paint an -MJPEG part only once the next part starts, and the hub writes a frame only when the +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 frame](../../apps/server/src/device/mjpegStillFrame.ts) after a -short quiet period; without it a still screen never appears. In browser dev, -the Vite proxy must forward WebSocket upgrades for `/api`, not only `/ws`. +[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 From 45953d52f13294e394dab7d2b4ccef9f71232dd6 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 03:04:01 +0000 Subject: [PATCH 6/6] fix(server): still-frame tracker scans a bounded header prefix A header split across chunks concatenated the whole next chunk, copying the frame it carried. Only a header-sized prefix joins the scan now. The common path with no still frame pending pulls directly instead of forking. Co-Authored-By: Claude Fable 5.1 --- apps/server/src/device/mjpegStillFrame.test.ts | 10 ++++++++++ apps/server/src/device/mjpegStillFrame.ts | 11 ++++++++--- 2 files changed, 18 insertions(+), 3 deletions(-) diff --git a/apps/server/src/device/mjpegStillFrame.test.ts b/apps/server/src/device/mjpegStillFrame.test.ts index 8739ffb3ff83..8867dec3bf5a 100644 --- a/apps/server/src/device/mjpegStillFrame.test.ts +++ b/apps/server/src/device/mjpegStillFrame.test.ts @@ -65,6 +65,16 @@ describe("repeatStillFrame", () => { }), ); + 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; diff --git a/apps/server/src/device/mjpegStillFrame.ts b/apps/server/src/device/mjpegStillFrame.ts index 8faffcd75cf6..4792bf083257 100644 --- a/apps/server/src/device/mjpegStillFrame.ts +++ b/apps/server/src/device/mjpegStillFrame.ts @@ -51,8 +51,12 @@ const partTracker = () => { let part: Part | undefined; while (rest.length > 0) { if (partEnd === null) { - // A header split across chunks is re-scanned per chunk, bounded by MAX_HEADER_BYTES. - const head = size === 0 ? rest : Buffer.concat([...pieces, rest]); + // 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); @@ -103,8 +107,9 @@ export const repeatStillFrame = (body: Stream.Stream) => return chunks; }; return Effect.gen(function* () { - pending ??= yield* Effect.forkIn(pull, scope); 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));