fix(server): iOS device viewer no longer times out on a still simulator screen

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<RegExp> = [
   /^\/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"] } : {}),

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<Uint8Array, Cause.Done>();
+  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<Uint8Array>) =>
+    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));
+    }),
+  );
+});

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 `<img>` 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<Uint8Array>;
+
+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<Uint8Array> = [];
+  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 = <E, R>(body: Stream.Stream<Uint8Array, E, R>) =>
+  Stream.transformPull(body, (pull, scope) =>
+    Effect.sync(() => {
+      const track = partTracker();
+      let pending: Fiber.Fiber<Effect.Success<typeof pull>, Effect.Error<typeof pull>> | null =
+        null;
+      let still: Part | null = null;
+      const forward = (chunks: Effect.Success<typeof pull>) => {
+        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));
+      });
+    }),
+  );

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