Skip to content

Commit 34d9855

Browse files
authored
Merge pull request #40 from streamwizard/feat/node-cache
perf: stop polling Supabase for data this node already knows
2 parents d22288b + 41811cb commit 34d9855

19 files changed

Lines changed: 947 additions & 41 deletions

.env.example

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,26 @@ OBS_IMAGE_TAG=
3232
# node's per-node obs_command key, whose hash is fetched from GET /api/nodes/me
3333
# at boot (no env var needed — it rides the existing NODE_API_KEY channel).
3434

35+
# How long this node's own row (capacity + hardware specs, from GET
36+
# /api/nodes/me) is held in memory. Defaults to 1h — the row is write-once in
37+
# practice, and everything that needs a guaranteed-fresh read (capacity gate,
38+
# command-key rotation) bypasses the TTL anyway. Leave unset unless you are
39+
# debugging staleness — lowering it costs rest-api requests on every metrics
40+
# tick, which is exactly what this cache exists to avoid.
41+
NODE_CACHE_TTL_MS=
42+
43+
# How long to keep serving the cached row after a failed refresh before trying
44+
# again. Defaults to 10s. Stops a rest-api outage turning the 3s dashboard
45+
# ticks into a retry storm.
46+
NODE_CACHE_ERROR_BACKOFF_MS=
47+
48+
# This node's instance list, cached in memory. Kept correct by invalidation:
49+
# every local mutation busts it, and the 60s OBS-watcher sync's live read
50+
# refreshes it. The TTL is only a failsafe for a dead sync loop — defaults to
51+
# 90s; the backoff (default 10s) matches NODE_CACHE_ERROR_BACKOFF_MS's job.
52+
INSTANCE_CACHE_TTL_MS=
53+
INSTANCE_CACHE_ERROR_BACKOFF_MS=
54+
3555
# ── ws-server lifecycle broadcast (optional) ──────────────────────────────────
3656
# Pushes container start/stop/delete/crash events to the owning user's browser
3757
# clients via the main ws-server's /internal/broadcast fan-out. Both must be set

package.json

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
"scripts": {
77
"dev": "bun --hot run src/index.ts",
88
"start": "bun run src/index.ts",
9-
"test": "bun test"
9+
"test": "bash scripts/test.sh"
1010
},
1111
"devDependencies": {
1212
"@types/bun": "latest",

scripts/test.sh

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
#!/usr/bin/env bash
2+
# Runs each test file in its own bun process.
3+
#
4+
# `bun test` normally runs every file in one process, and mock.module() writes
5+
# into a process-global module registry that neither mock.restore() nor a
6+
# cache-busting import specifier can undo. So a mock registered by one file
7+
# leaks into every file discovered after it: routes/obs.test.ts stubs
8+
# services/command-key, which meant services/command-key.test.ts was silently
9+
# asserting against that stub instead of the real module.
10+
#
11+
# Per-file isolation costs a few seconds and makes the mocks mean what they say.
12+
set -euo pipefail
13+
14+
failed=0
15+
16+
while IFS= read -r file; do
17+
echo "── $file"
18+
if ! bun test "$file"; then
19+
failed=1
20+
fi
21+
done < <(find src -name '*.test.ts' | sort)
22+
23+
exit "$failed"

src/clients/docker.ts

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,8 @@
11
import Docker from "dockerode";
22
import { debug, log } from "../utils/logger";
33
import { PLUGINS_LOCAL_DIR } from "../services/plugins";
4-
import { getInstanceByIdAdmin, listNodeInstances, updateInstance, updateInstanceByContainerId } from "./supabase";
4+
import { getInstanceByIdAdmin, updateInstance, updateInstanceByContainerId } from "./supabase";
5+
import { refreshNodeInstances } from "../services/instance-cache";
56
import { trackInstanceEvent } from "../services/influx-metrics";
67
import { StreamwizardApi } from "./streamwizard-api";
78
import { broadcastLifecycle } from "./ws-server";
@@ -273,7 +274,9 @@ const RESTART_STAGGER_MS = 3000;
273274
// so a node reboot or manager crash doesn't strand every paying customer's
274275
// session until an operator notices.
275276
export async function reconcileContainers(nodeId: string): Promise<void> {
276-
const nodeInstances = await listNodeInstances(nodeId);
277+
// Live read (and boot-time cache warm-up): the process just started, so its
278+
// memory is empty by definition — reconcile must see the database's truth.
279+
const nodeInstances = await refreshNodeInstances();
277280
const knownContainerIds = new Set(
278281
nodeInstances.map((i) => i.container_id).filter((id): id is string => !!id)
279282
);

src/clients/streamwizard-api.ts

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,8 +9,15 @@ if (!apiUrl || !nodeApiKey) {
99
throw new Error("REST_API_URL and NODE_API_KEY must be set");
1010
}
1111

12+
// Everything on this instance is small JSON — media uploads go straight to S3 —
13+
// so no request has a reason to run long. The timeout matters because callers
14+
// now share responses through node-cache's single-flight: without it one hung
15+
// socket would wedge every waiting consumer instead of just its own call.
16+
const REQUEST_TIMEOUT_MS = 15_000;
17+
1218
export const StreamwizardApi = axios.create({
1319
baseURL: apiUrl,
20+
timeout: REQUEST_TIMEOUT_MS,
1421
headers: {
1522
"Content-Type": "application/json",
1623
Authorization: `Bearer ${nodeApiKey}`,

src/clients/supabase.ts

Lines changed: 22 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,10 @@ import {
1313
apiUpdateInstanceByContainerId,
1414
} from "./streamwizard-api";
1515
import type { CloudObsPlanLimits, Instance, Node } from "../types";
16+
// Import cycle with services/instance-cache (it reads listNodeInstances from
17+
// this module) — harmless: both sides only call across the boundary at
18+
// runtime, never during module evaluation.
19+
import { invalidateNodeInstances } from "../services/instance-cache";
1620

1721
export async function isAdmin(userId: string): Promise<boolean> {
1822
return apiIsAdmin(userId);
@@ -38,22 +42,36 @@ export async function getInstanceByIdAdmin(instanceId: string): Promise<Instance
3842
return apiGetInstanceByIdAdmin(instanceId);
3943
}
4044

45+
// The four instance mutations below are the single choke point through which
46+
// every change to this node's instances flows (provision, start/stop, the
47+
// crash watchdog, delete). Each one invalidates the cached instance list on
48+
// success, which is what makes services/instance-cache correct rather than
49+
// merely fresh-ish — see the header comment there before adding a mutation
50+
// path that bypasses this file.
51+
4152
export async function insertInstance(
4253
instance: Omit<Instance, "created_at" | "storage_quota_mb" | "used_storage_bytes">,
4354
): Promise<Instance> {
44-
return apiInsertInstance(instance);
55+
const created = await apiInsertInstance(instance);
56+
invalidateNodeInstances();
57+
return created;
4558
}
4659

4760
export async function updateInstance(instanceId: string, fields: Partial<Instance>): Promise<Instance> {
48-
return apiUpdateInstance(instanceId, fields);
61+
const updated = await apiUpdateInstance(instanceId, fields);
62+
invalidateNodeInstances();
63+
return updated;
4964
}
5065

5166
export async function updateInstanceByContainerId(containerId: string, fields: Partial<Instance>): Promise<Instance | null> {
52-
return apiUpdateInstanceByContainerId(containerId, fields);
67+
const updated = await apiUpdateInstanceByContainerId(containerId, fields);
68+
if (updated) invalidateNodeInstances();
69+
return updated;
5370
}
5471

5572
export async function deleteInstance(instanceId: string): Promise<void> {
56-
return apiDeleteInstance(instanceId);
73+
await apiDeleteInstance(instanceId);
74+
invalidateNodeInstances();
5775
}
5876

5977
export async function countActiveInstances(_nodeId: string): Promise<number> {

src/index.ts

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@ import metrics from "./routes/metrics";
88
import { websocket } from "./utils/ws";
99
import { debug, log } from "./utils/logger";
1010
import { reconcileContainers, registerConfigHandlers, registerInstanceLifecycleHandlers, registerObsEventHandlers, startEventListener } from "./clients/docker";
11-
import { listNodeInstances } from "./clients/supabase";
11+
import { getCachedNodeInstances } from "./services/instance-cache";
1212
import { pushObsConfig, removeLocalConfig } from "./services/obs-config";
1313
import { restartInstance } from "./services/instance-lifecycle";
1414
import { checkS3 } from "./clients/s3";
@@ -117,7 +117,9 @@ setInterval(() => {
117117
startMetricsPersistence(NODE_ID);
118118

119119
// Load this node's obs_command key hash so the /obs route can authenticate the
120-
// obs-auto-switcher, then keep it fresh so key rotations propagate live.
120+
// obs-auto-switcher. The retry loop only fires while no hash is held (failed
121+
// boot fetch); rotations propagate through verifyCommandKey's mismatch path,
122+
// not polling.
121123
await loadCommandKeyHash();
122124
startCommandKeyRefresh();
123125

@@ -132,7 +134,7 @@ setInterval(() => {
132134
}, CONFIG_AUTOSAVE_INTERVAL_MS);
133135

134136
async function autosaveRunningInstanceConfigs(): Promise<void> {
135-
const nodeInstances = await listNodeInstances(NODE_ID);
137+
const nodeInstances = await getCachedNodeInstances();
136138
const running = nodeInstances.filter((i) => i.status === "running" && i.container_id);
137139

138140
await Promise.all(

src/routes/admin.ts

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,10 @@
11
import { randomUUID } from "node:crypto";
22
import { Hono, type Context, type Next } from "hono";
3-
import { deleteInstance, getInstanceByIdAdmin, isAdmin, listNodeInstances, updateInstance } from "../clients/supabase";
3+
import { deleteInstance, getInstanceByIdAdmin, isAdmin, updateInstance } from "../clients/supabase";
4+
import { getCachedNodeInstances } from "../services/instance-cache";
45
import { getAllMetrics } from "../services/metrics";
56
import { clearApiStopping, markApiStopping, NOVNC_PORT_INTERNAL, OBS_WS_PORT_INTERNAL, removeContainer, stopContainer } from "../clients/docker";
67
import { broadcastLifecycle } from "../clients/ws-server";
7-
import { NODE_ID } from "../utils/node";
88
import { authMiddleware } from "../middleware/auth";
99
import { withInstanceLock } from "../utils/instance-lock";
1010
import { upgradeWebSocket } from "../utils/ws";
@@ -85,7 +85,10 @@ admin.get(
8585
};
8686

8787
const sendMetrics = async (ws: MetricsSocket) => {
88-
const nodeInstances = await listNodeInstances(NODE_ID);
88+
// Cached: admin start/stop/delete invalidate it, so a state change this
89+
// process made is visible on the next 3s tick — same as when this
90+
// fetched live, minus the rest-api round trip per tick per socket.
91+
const nodeInstances = await getCachedNodeInstances();
8992
const payload = await getAllMetrics(nodeInstances);
9093
send(ws, { type: "notification", payload });
9194
debug("ws", `metrics/stream sent payload for ${nodeInstances.length} instance(s)`);

src/routes/instances.ts

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5,12 +5,12 @@ import {
55
countActiveInstances,
66
deleteInstance,
77
getInstanceById,
8-
getNode,
98
getSubscriptionLimits,
109
insertInstance,
1110
listUserInstances,
1211
updateInstance,
1312
} from "../clients/supabase";
13+
import { refreshNode } from "../services/node-cache";
1414
import {
1515
createContainer,
1616
clearApiStopping,
@@ -261,8 +261,13 @@ instances.post("/", async (c) => {
261261
return c.json({ error: "obs_ws_password is required" }, 400);
262262
}
263263

264+
// refreshNode, not the cached read: max_instances/max_encoder_sessions gate
265+
// authorization here rather than feeding telemetry, so an admin raising
266+
// capacity to unblock someone must take effect on the very next create — not
267+
// whenever the TTL happens to lapse. Creates are rare and rate-limited, so
268+
// the live read costs nothing.
264269
const [node, planLimits] = await Promise.all([
265-
getNode(NODE_ID),
270+
refreshNode(),
266271
getSubscriptionLimits(subscriptionId),
267272
]);
268273

src/services/command-key.test.ts

Lines changed: 166 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,166 @@
1+
import { createHash } from "crypto";
2+
import { describe, expect, it, mock } from "bun:test";
3+
import type { Node } from "../types";
4+
5+
process.env.S3_ENDPOINT ??= "http://127.0.0.1:1";
6+
process.env.S3_ACCESS_KEY ??= "test";
7+
process.env.S3_SECRET_KEY ??= "test";
8+
process.env.TOKEN_ENCRYPTION_KEY ??= "0".repeat(64);
9+
process.env.NODE_ID ??= "test-node";
10+
process.env.REST_API_URL ??= "http://127.0.0.1:1";
11+
process.env.NODE_API_KEY ??= "test";
12+
13+
const sha = (s: string) => createHash("sha256").update(s).digest("hex");
14+
15+
const KEY_A = "command-key-a";
16+
const KEY_B = "command-key-b";
17+
18+
function makeNode(commandKeyHash: string | null): Node {
19+
return {
20+
id: "test-node",
21+
name: "obs-node-1",
22+
max_instances: 4,
23+
memory_mb: 32_000,
24+
cpu_quota: 4,
25+
vram_mb: 8_000,
26+
total_vram_mb: 24_000,
27+
shm_size: "2g",
28+
gpu_bus_id: "00000000:01:00.0",
29+
max_encoder_sessions: 5,
30+
command_key_hash: commandKeyHash,
31+
created_at: "2026-07-21T00:00:00Z",
32+
} as Node;
33+
}
34+
35+
let refreshCalls = 0;
36+
let cachedCalls = 0;
37+
let nextNode: Node = makeNode(sha(KEY_A));
38+
let failWith: Error | null = null;
39+
40+
// Mock the cache, not the client — the point of these tests is WHICH cache
41+
// entry point command-key reaches for.
42+
mock.module("./node-cache", () => ({
43+
refreshNode: async () => {
44+
refreshCalls++;
45+
if (failWith) throw failWith;
46+
return nextNode;
47+
},
48+
getCachedNode: async () => {
49+
cachedCalls++;
50+
return nextNode;
51+
},
52+
peekCachedNode: () => nextNode,
53+
resetNodeCache: () => {},
54+
}));
55+
56+
const { loadCommandKeyHash, verifyCommandKey, hasCommandKey, retryCommandKeyLoadIfMissing } =
57+
await import("./command-key");
58+
59+
/**
60+
* These tests share module state (`commandKeyHash`, `lastFetchAt`) and run in
61+
* declaration order, which is load-bearing: the mismatch-refresh path is gated
62+
* on a 30s cooldown measured from the last SUCCESSFUL fetch, so the only way to
63+
* exercise it without sleeping 30s is to go first, while `lastFetchAt` is still
64+
* 0 from a failed load. Do not reorder these blocks.
65+
*/
66+
67+
describe("command-key, before any successful load", () => {
68+
it("reports no key and rejects everything", async () => {
69+
expect(hasCommandKey()).toBe(false);
70+
});
71+
72+
it("a failed load leaves lastFetchAt untouched, so the cooldown stays open", async () => {
73+
failWith = new Error("rest-api down");
74+
await loadCommandKeyHash();
75+
76+
expect(refreshCalls).toBe(1);
77+
expect(hasCommandKey()).toBe(false);
78+
});
79+
80+
it("picks up a rotated key through the mismatch path", async () => {
81+
// lastFetchAt is still 0 (the load above failed), so the cooldown has
82+
// elapsed by definition and a mismatch is allowed to re-fetch.
83+
failWith = null;
84+
nextNode = makeNode(sha(KEY_B));
85+
const before = refreshCalls;
86+
87+
const ok = await verifyCommandKey(KEY_B);
88+
89+
expect(ok).toBe(true);
90+
expect(refreshCalls).toBe(before + 1);
91+
expect(hasCommandKey()).toBe(true);
92+
});
93+
});
94+
95+
describe("command-key, with a warm hash", () => {
96+
it("accepts the matching key without touching the network", async () => {
97+
nextNode = makeNode(sha(KEY_B));
98+
await loadCommandKeyHash();
99+
const before = refreshCalls;
100+
101+
expect(await verifyCommandKey(KEY_B)).toBe(true);
102+
expect(refreshCalls).toBe(before);
103+
});
104+
105+
it("rejects a wrong key", async () => {
106+
expect(await verifyCommandKey("nope")).toBe(false);
107+
});
108+
109+
it("does not re-fetch on repeated mismatches inside the cooldown", async () => {
110+
// The /obs route is public, so unauthenticated callers can reach this path.
111+
// The cooldown is what stops them amplifying into rest-api load.
112+
await loadCommandKeyHash();
113+
const before = refreshCalls;
114+
115+
await verifyCommandKey("bad-1");
116+
await verifyCommandKey("bad-2");
117+
await verifyCommandKey("bad-3");
118+
119+
expect(refreshCalls).toBe(before);
120+
});
121+
});
122+
123+
describe("the rotation invariant", () => {
124+
it("loadCommandKeyHash forces a live read and never accepts a cached one", async () => {
125+
const refreshBefore = refreshCalls;
126+
const cachedBefore = cachedCalls;
127+
128+
await loadCommandKeyHash();
129+
130+
// If this ever flips to getCachedNode, a rotated obs_command key can never
131+
// reach this node and /obs fails closed permanently.
132+
expect(refreshCalls).toBe(refreshBefore + 1);
133+
expect(cachedCalls).toBe(cachedBefore);
134+
});
135+
136+
it("survives a node with no command key provisioned", async () => {
137+
nextNode = makeNode(null);
138+
await loadCommandKeyHash();
139+
140+
expect(hasCommandKey()).toBe(false);
141+
expect(await verifyCommandKey(KEY_B)).toBe(false);
142+
});
143+
});
144+
145+
describe("the retry loop", () => {
146+
it("keeps fetching while no hash is held", async () => {
147+
// Carries state from the block above: the last load found no key, so the
148+
// loop must still be trying — this is the failed-boot-fetch self-heal.
149+
nextNode = makeNode(sha(KEY_A));
150+
const before = refreshCalls;
151+
152+
await retryCommandKeyLoadIfMissing();
153+
154+
expect(refreshCalls).toBe(before + 1);
155+
expect(hasCommandKey()).toBe(true);
156+
});
157+
158+
it("goes quiet once a hash is held — rotation rides the mismatch path, not polling", async () => {
159+
const before = refreshCalls;
160+
161+
await retryCommandKeyLoadIfMissing();
162+
163+
expect(refreshCalls).toBe(before);
164+
expect(hasCommandKey()).toBe(true);
165+
});
166+
});

0 commit comments

Comments
 (0)