From cff0e5b93d9f5cd4ee74d4a04b21a1642396b67a Mon Sep 17 00:00:00 2001 From: Dimitri Kennedy Date: Mon, 3 Aug 2026 15:24:56 -0400 Subject: [PATCH 1/3] fix(daemon): bound Docker event refresh work --- docs/architecture.md | 5 + docs/gateway-api.md | 12 ++ src/control-plane/sdk/gateway-client.ts | 12 ++ src/daemon/docker-events.ts | 41 +++++++ src/daemon/refresh-scheduler.ts | 134 +++++++++++++++++++++ src/daemon/runtime-cache.ts | 113 +++++++++++++++--- src/daemon/server.ts | 69 ++++++++--- src/lib/runtime-projects.ts | 97 +++++++++++++++- tests/docker-events.test.ts | 50 ++++++++ tests/refresh-scheduler.test.ts | 72 ++++++++++++ tests/runtime-cache.test.ts | 118 ++++++++++++++++++- tests/runtime-projects.test.ts | 147 +++++++++++++++++++++++- 12 files changed, 820 insertions(+), 50 deletions(-) create mode 100644 src/daemon/refresh-scheduler.ts create mode 100644 tests/docker-events.test.ts create mode 100644 tests/refresh-scheduler.test.ts diff --git a/docs/architecture.md b/docs/architecture.md index 383b0b09..75b88bd8 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -208,6 +208,11 @@ containers. It serves a small local API over a Unix socket at `~/.hack/daemon/ha - `hack ps --json` - streaming consumers (TUI/MCP) +The event watcher ignores container actions such as health-check `exec_*` events that cannot change +the cached runtime view. Relevant event bursts are debounced and rate-bounded. Event refreshes reuse +inspect data for unchanged container IDs, while startup, watcher recovery, and the 30-second interval +perform full inspection so missed events and mutable network data remain eventually consistent. + If the daemon is not running (or version-mismatched), the CLI falls back to direct Docker calls. Runtime health: diff --git a/docs/gateway-api.md b/docs/gateway-api.md index 39840ccc..acb11460 100644 --- a/docs/gateway-api.md +++ b/docs/gateway-api.md @@ -320,9 +320,21 @@ Response: | `cache_age_ms` | number or null | Cache age in milliseconds | | `last_refresh_at` | string or null | Last refresh attempt | | `refresh_count` | number | Refresh count | +| `refresh_requests` | number | Scheduled refresh requests after event filtering | +| `refresh_requests_coalesced` | number | Refresh requests merged into pending work | | `refresh_failures` | number | Refresh failures | +| `refresh_in_flight` | boolean | Whether the runtime cache is currently refreshing | +| `last_refresh_duration_ms` | number or null | Duration of the latest completed refresh | +| `max_refresh_duration_ms` | number or null | Longest refresh duration since daemon startup | | `last_event_at` | string or null | Last docker event timestamp | | `events_seen` | number | Docker events seen | +| `events_relevant` | number | Docker events classified as runtime-affecting | +| `events_ignored` | number | Known non-state-changing Docker events ignored | +| `inspect_calls` | number | Docker inspect invocations by the daemon cache | +| `inspect_ids` | number | Container IDs requested across inspect invocations | +| `inspect_cache_hits` | number | Container IDs served from the inspect cache | +| `inspect_cache_misses` | number | Container IDs absent from the inspect cache | +| `inspect_full_refreshes` | number | Forced full inspect reconciliations | | `streams_active` | number | Active WS streams | | `runtime_ok` | boolean | Docker runtime availability | | `runtime_error` | string or null | Runtime error details (when unavailable) | diff --git a/src/control-plane/sdk/gateway-client.ts b/src/control-plane/sdk/gateway-client.ts index adae8fea..29335ba3 100644 --- a/src/control-plane/sdk/gateway-client.ts +++ b/src/control-plane/sdk/gateway-client.ts @@ -45,9 +45,21 @@ export type GatewayMetrics = { readonly cache_age_ms: number | null; readonly last_refresh_at: string | null; readonly refresh_count: number; + readonly refresh_requests: number; + readonly refresh_requests_coalesced: number; readonly refresh_failures: number; + readonly refresh_in_flight: boolean; + readonly last_refresh_duration_ms: number | null; + readonly max_refresh_duration_ms: number | null; readonly last_event_at: string | null; readonly events_seen: number; + readonly events_relevant: number; + readonly events_ignored: number; + readonly inspect_calls: number; + readonly inspect_ids: number; + readonly inspect_cache_hits: number; + readonly inspect_cache_misses: number; + readonly inspect_full_refreshes: number; readonly streams_active: number; }; diff --git a/src/daemon/docker-events.ts b/src/daemon/docker-events.ts index cd9aca2d..b6df357e 100644 --- a/src/daemon/docker-events.ts +++ b/src/daemon/docker-events.ts @@ -7,6 +7,36 @@ export interface DockerEventWatcher { stop(): void; } +const ignoredContainerActions = new Set([ + "attach", + "commit", + "copy", + "detach", + "exec_create", + "exec_detach", + "exec_die", + "exec_start", + "export", + "resize", + "top", +]); + +/** + * Returns false only for known container actions that cannot change Hack's + * cached topology, state, ports, labels, mounts, or networks. Unknown actions + * fail open so the periodic reconciliation is not the only freshness path for + * new Docker event types. + */ +export function shouldRefreshForDockerEvent(opts: { + readonly event: DockerEvent; +}): boolean { + const action = readDockerEventAction({ event: opts.event }); + if (!action) { + return true; + } + return !ignoredContainerActions.has(action); +} + export function startDockerEventWatcher(opts: { readonly onEvent: (event: DockerEvent) => void; readonly onError: (message: string) => void; @@ -94,6 +124,17 @@ function parseDockerEvent(opts: { readonly line: string }): DockerEvent | null { } } +function readDockerEventAction(opts: { + readonly event: DockerEvent; +}): string | null { + const rawAction = opts.event.Action ?? opts.event.status; + if (typeof rawAction !== "string") { + return null; + } + const action = rawAction.split(":", 1)[0]?.trim().toLowerCase() ?? ""; + return action.length > 0 ? action : null; +} + function sleep(opts: { readonly ms: number }): Promise { return new Promise((resolve) => setTimeout(resolve, opts.ms)); } diff --git a/src/daemon/refresh-scheduler.ts b/src/daemon/refresh-scheduler.ts new file mode 100644 index 00000000..7f2415c1 --- /dev/null +++ b/src/daemon/refresh-scheduler.ts @@ -0,0 +1,134 @@ +export type RefreshUrgency = "debounced" | "immediate"; + +export interface RefreshScheduler { + request(opts: { + readonly reason: string; + readonly urgency: RefreshUrgency; + }): void; + stop(): void; +} + +type Timer = ReturnType; + +type PendingRefresh = { + readonly reason: string; + readonly requestedAtMs: number; + readonly urgency: RefreshUrgency; +}; + +export function createRefreshScheduler(opts: { + readonly refresh: (opts: { + readonly reason: string; + readonly forceInspect: boolean; + }) => Promise; + readonly debounceMs?: number; + readonly minIntervalMs?: number; + readonly maxWaitMs?: number; + readonly now?: () => number; + readonly onRequest?: (opts: { readonly coalesced: boolean }) => void; + readonly onRefreshStart?: () => void; + readonly onRefreshFinish?: (opts: { + readonly durationMs: number; + readonly error: unknown | null; + }) => void; +}): RefreshScheduler { + const debounceMs = opts.debounceMs ?? 250; + const minIntervalMs = opts.minIntervalMs ?? 1000; + const maxWaitMs = opts.maxWaitMs ?? 2000; + const now = opts.now ?? Date.now; + + let active = false; + let lastCompletedAtMs: number | null = null; + let pending: PendingRefresh | null = null; + let stopped = false; + let timer: Timer | null = null; + + function clearTimer(): void { + if (!timer) { + return; + } + clearTimeout(timer); + timer = null; + } + + function schedulePending(): void { + if (stopped || active || !pending) { + return; + } + + clearTimer(); + const currentTimeMs = now(); + const earliestByIntervalMs = + lastCompletedAtMs === null + ? currentTimeMs + : lastCompletedAtMs + minIntervalMs; + const dueAtMs = + pending.urgency === "immediate" + ? currentTimeMs + : Math.min( + pending.requestedAtMs + maxWaitMs, + Math.max(currentTimeMs + debounceMs, earliestByIntervalMs) + ); + const delayMs = Math.max(0, dueAtMs - currentTimeMs); + timer = setTimeout(() => { + timer = null; + void drain(); + }, delayMs); + } + + async function drain(): Promise { + if (stopped || active || !pending) { + return; + } + + const request = pending; + pending = null; + active = true; + const startedAtMs = now(); + opts.onRefreshStart?.(); + let error: unknown | null = null; + try { + await opts.refresh({ + reason: request.reason, + forceInspect: request.urgency === "immediate", + }); + } catch (caught: unknown) { + error = caught; + } finally { + const completedAtMs = now(); + lastCompletedAtMs = completedAtMs; + active = false; + opts.onRefreshFinish?.({ + durationMs: Math.max(0, completedAtMs - startedAtMs), + error, + }); + schedulePending(); + } + } + + return { + request({ reason, urgency }) { + if (stopped) { + return; + } + const coalesced = active || pending !== null; + opts.onRequest?.({ coalesced }); + const currentTimeMs = now(); + if (!pending) { + pending = { reason, requestedAtMs: currentTimeMs, urgency }; + } else if (urgency === "immediate") { + pending = { + reason, + requestedAtMs: pending.requestedAtMs, + urgency, + }; + } + schedulePending(); + }, + stop() { + stopped = true; + pending = null; + clearTimer(); + }, + }; +} diff --git a/src/daemon/runtime-cache.ts b/src/daemon/runtime-cache.ts index d78bdf48..6b430554 100644 --- a/src/daemon/runtime-cache.ts +++ b/src/daemon/runtime-cache.ts @@ -9,7 +9,9 @@ import { readProjectsRegistry } from "../lib/projects-registry.ts"; import type { RuntimeProject } from "../lib/runtime-projects.ts"; import { autoRegisterRuntimeHackProjects, + createRuntimeInspectCache, filterRuntimeProjects, + getRuntimeInspectCacheDiagnostics, readRuntimeProjects, } from "../lib/runtime-projects.ts"; import { @@ -98,7 +100,10 @@ export type PsPayload = { }; export interface RuntimeCache { - refresh(opts: { readonly reason: string }): Promise; + refresh(opts: { + readonly reason: string; + readonly forceInspect?: boolean; + }): Promise; getProjectsPayload(opts: { readonly filter: string | null; readonly includeGlobal: boolean; @@ -111,8 +116,20 @@ export interface RuntimeCache { readonly branch: string | null; }): PsPayload; getSnapshot(): RuntimeSnapshot | null; + getDiagnostics(): RuntimeCacheDiagnostics; } +export type RuntimeCacheDiagnostics = { + readonly refreshInFlight: boolean; + readonly lastRefreshDurationMs: number | null; + readonly maxRefreshDurationMs: number | null; + readonly inspectCalls: number; + readonly inspectIds: number; + readonly inspectCacheHits: number; + readonly inspectCacheMisses: number; + readonly inspectFullRefreshes: number; +}; + export function createRuntimeCache(opts: { readonly onRefresh?: (snapshot: RuntimeSnapshot) => void; readonly deps?: { @@ -133,7 +150,13 @@ export function createRuntimeCache(opts: { let snapshot: RuntimeSnapshot | null = null; let refreshTask: Promise | null = null; - let pendingReason: string | null = null; + let pendingRefresh: { + readonly reason: string; + readonly forceInspect: boolean; + } | null = null; + let lastRefreshDurationMs: number | null = null; + let maxRefreshDurationMs: number | null = null; + const inspectCache = createRuntimeInspectCache(); let health: RuntimeHealth = { ok: false, error: "runtime_not_checked", @@ -154,19 +177,30 @@ export function createRuntimeCache(opts: { const refresh = async ({ reason, + forceInspect = true, }: { readonly reason: string; + readonly forceInspect?: boolean; }): Promise => { if (refreshTask) { - queueRefresh({ reason: `pending:${reason}`, priority: "normal" }); + queueRefresh({ + forceInspect, + reason: `pending:${reason}`, + priority: "normal", + }); await refreshTask; return; } + const startedAtMs = Date.now(); refreshTask = (async () => { const checkedAtMs = Date.now(); const previousSnapshot = snapshot; - const runtimeResult = await readRuntimeProjects({ includeGlobal: true }); + const runtimeResult = await readRuntimeProjects({ + includeGlobal: true, + inspectCache, + forceInspect, + }); const refreshed = await resolveRefreshResult({ checkedAtMs, currentHealth: health, @@ -174,41 +208,53 @@ export function createRuntimeCache(opts: { reason, runtimeResult, }); - health = refreshed.health; if (refreshed.repairReason) { queueRefresh({ + forceInspect: true, reason: refreshed.repairReason, priority: "repair", }); } + let nextSnapshot: RuntimeSnapshot; if (runtimeResult.ok) { await autoRegisterRuntimeHackProjects({ runtime: runtimeResult.runtime, }); - snapshot = { + nextSnapshot = { runtime: runtimeResult.runtime, updatedAtMs: checkedAtMs, - health, + health: refreshed.health, }; } else { - snapshot = { + nextSnapshot = { runtime: snapshot?.runtime ?? [], updatedAtMs: snapshot?.updatedAtMs ?? null, - health, + health: refreshed.health, }; } - opts.onRefresh?.(snapshot); + health = refreshed.health; + snapshot = nextSnapshot; + opts.onRefresh?.(nextSnapshot); })(); - await refreshTask; - refreshTask = null; + try { + await refreshTask; + } catch (error: unknown) { + pendingRefresh = null; + throw error; + } finally { + const durationMs = Math.max(0, Date.now() - startedAtMs); + lastRefreshDurationMs = durationMs; + maxRefreshDurationMs = Math.max(maxRefreshDurationMs ?? 0, durationMs); + refreshTask = null; + } - if (pendingReason) { - const queuedReason = pendingReason; - pendingReason = null; - await refresh({ reason: queuedReason }); + if (pendingRefresh) { + const queuedRefresh = pendingRefresh; + pendingRefresh = null; + await refresh(queuedRefresh); } }; @@ -329,14 +375,30 @@ export function createRuntimeCache(opts: { }; function queueRefresh(opts: { + readonly forceInspect: boolean; readonly reason: string; readonly priority: "normal" | "repair"; }): void { if (opts.priority === "repair") { - pendingReason = opts.reason; + pendingRefresh = { + forceInspect: true, + reason: opts.reason, + }; + return; + } + if (!pendingRefresh) { + pendingRefresh = { + forceInspect: opts.forceInspect, + reason: opts.reason, + }; return; } - pendingReason ??= opts.reason; + if (opts.forceInspect && !pendingRefresh.forceInspect) { + pendingRefresh = { + forceInspect: true, + reason: opts.reason, + }; + } } return { @@ -344,6 +406,21 @@ export function createRuntimeCache(opts: { getProjectsPayload, getPsPayload, getSnapshot: () => snapshot, + getDiagnostics: () => { + const inspect = getRuntimeInspectCacheDiagnostics({ + cache: inspectCache, + }); + return { + refreshInFlight: refreshTask !== null, + lastRefreshDurationMs, + maxRefreshDurationMs, + inspectCalls: inspect.inspectCalls, + inspectIds: inspect.inspectIds, + inspectCacheHits: inspect.cacheHits, + inspectCacheMisses: inspect.cacheMisses, + inspectFullRefreshes: inspect.fullRefreshes, + }; + }, }; } diff --git a/src/daemon/server.ts b/src/daemon/server.ts index 1e5863e8..ee686f21 100644 --- a/src/daemon/server.ts +++ b/src/daemon/server.ts @@ -26,11 +26,13 @@ import { } from "./control-plane-route-validation.ts"; import { type DockerEventWatcher, + shouldRefreshForDockerEvent, startDockerEventWatcher, } from "./docker-events.ts"; import { createDaemonLogger } from "./logger.ts"; import type { DaemonPaths } from "./paths.ts"; import { removeFileIfExists, writeDaemonPid } from "./process.ts"; +import { createRefreshScheduler } from "./refresh-scheduler.ts"; import { type RequestTargetProxy, startRequestTargetProxy, @@ -52,9 +54,13 @@ type DaemonMetrics = { readonly startedAtMs: number; lastEventAtMs: number | null; eventsSeen: number; + eventsIgnored: number; + eventsRelevant: number; streamsActive: number; lastRefreshAtMs: number | null; refreshCount: number; + refreshRequests: number; + refreshRequestsCoalesced: number; refreshFailures: number; }; @@ -87,9 +93,13 @@ export async function runDaemon({ startedAtMs: Date.now(), lastEventAtMs: null, eventsSeen: 0, + eventsIgnored: 0, + eventsRelevant: 0, streamsActive: 0, lastRefreshAtMs: null, refreshCount: 0, + refreshRequests: 0, + refreshRequestsCoalesced: 0, refreshFailures: 0, }; @@ -111,20 +121,22 @@ export async function runDaemon({ logger.warn({ message: warning }); } - let refreshTimer: ReturnType | null = null; - const scheduleRefresh = ({ reason }: { readonly reason: string }) => { - if (refreshTimer) { - return; - } - refreshTimer = setTimeout(async () => { - refreshTimer = null; - try { - await cache.refresh({ reason }); - } catch { + const refreshScheduler = createRefreshScheduler({ + refresh: async ({ reason, forceInspect }) => { + await cache.refresh({ reason, forceInspect }); + }, + onRequest: ({ coalesced }) => { + metrics.refreshRequests += 1; + if (coalesced) { + metrics.refreshRequestsCoalesced += 1; + } + }, + onRefreshFinish: ({ error }) => { + if (error) { metrics.refreshFailures += 1; } - }, 250); - }; + }, + }); const dockerEventsDisabled = parseBoolean({ value: process.env.HACK_DAEMON_DISABLE_DOCKER_EVENTS ?? null, @@ -132,16 +144,27 @@ export async function runDaemon({ const watcher: DockerEventWatcher = dockerEventsDisabled ? createNoopDockerEventWatcher() : startDockerEventWatcher({ - onEvent: () => { + onEvent: (event) => { metrics.eventsSeen += 1; metrics.lastEventAtMs = Date.now(); - scheduleRefresh({ reason: "event" }); + if (!shouldRefreshForDockerEvent({ event })) { + metrics.eventsIgnored += 1; + return; + } + metrics.eventsRelevant += 1; + refreshScheduler.request({ + reason: "event", + urgency: "debounced", + }); }, onError: (message) => logger.warn({ message: `docker events: ${message}` }), onExit: (exitCode) => { logger.warn({ message: `docker events exited (${exitCode})` }); - scheduleRefresh({ reason: "events-exit" }); + refreshScheduler.request({ + reason: "events-exit", + urgency: "immediate", + }); }, }); if (dockerEventsDisabled) { @@ -153,7 +176,7 @@ export async function runDaemon({ const refreshIntervalMs = 30_000; setInterval(() => { - void cache.refresh({ reason: "interval" }); + refreshScheduler.request({ reason: "interval", urgency: "immediate" }); }, refreshIntervalMs); const requestContext = { @@ -303,6 +326,7 @@ export async function runDaemon({ const shutdown = async ({ reason }: { readonly reason: string }) => { logger.warn({ message: `Shutting down hackd (${reason})` }); + refreshScheduler.stop(); watcher.stop(); await daemonProxy?.close(); await gatewayProxy?.close(); @@ -374,6 +398,7 @@ async function handleRequest({ if (url.pathname === "/v1/metrics") { const snapshot = cache.getSnapshot(); + const diagnostics = cache.getDiagnostics(); const cacheUpdatedAtMs = snapshot?.updatedAtMs ?? null; const runtimeHealth = formatRuntimeHealth({ health: snapshot?.health ?? null, @@ -390,11 +415,23 @@ async function handleRequest({ ? new Date(metrics.lastRefreshAtMs).toISOString() : null, refresh_count: metrics.refreshCount, + refresh_requests: metrics.refreshRequests, + refresh_requests_coalesced: metrics.refreshRequestsCoalesced, refresh_failures: metrics.refreshFailures, + refresh_in_flight: diagnostics.refreshInFlight, + last_refresh_duration_ms: diagnostics.lastRefreshDurationMs, + max_refresh_duration_ms: diagnostics.maxRefreshDurationMs, last_event_at: metrics.lastEventAtMs ? new Date(metrics.lastEventAtMs).toISOString() : null, events_seen: metrics.eventsSeen, + events_relevant: metrics.eventsRelevant, + events_ignored: metrics.eventsIgnored, + inspect_calls: diagnostics.inspectCalls, + inspect_ids: diagnostics.inspectIds, + inspect_cache_hits: diagnostics.inspectCacheHits, + inspect_cache_misses: diagnostics.inspectCacheMisses, + inspect_full_refreshes: diagnostics.inspectFullRefreshes, streams_active: metrics.streamsActive, runtime_ok: runtimeHealth.ok, runtime_error: runtimeHealth.error, diff --git a/src/lib/runtime-projects.ts b/src/lib/runtime-projects.ts index 55714a30..e0f1eeeb 100644 --- a/src/lib/runtime-projects.ts +++ b/src/lib/runtime-projects.ts @@ -69,13 +69,51 @@ export type RuntimeProjectsResult = readonly checkedAtMs: number; }; -type ContainerInspectData = { +export type ContainerInspectData = { readonly labels: Record; readonly image: string | null; readonly mounts: readonly RuntimeContainerMount[]; readonly networks: readonly RuntimeContainerNetwork[]; }; +export type RuntimeInspectCacheDiagnostics = { + readonly inspectCalls: number; + readonly inspectIds: number; + readonly cacheHits: number; + readonly cacheMisses: number; + readonly fullRefreshes: number; +}; + +export interface RuntimeInspectCache { + readonly entries: Map; + readonly diagnostics: { + inspectCalls: number; + inspectIds: number; + cacheHits: number; + cacheMisses: number; + fullRefreshes: number; + }; +} + +export function createRuntimeInspectCache(): RuntimeInspectCache { + return { + entries: new Map(), + diagnostics: { + inspectCalls: 0, + inspectIds: 0, + cacheHits: 0, + cacheMisses: 0, + fullRefreshes: 0, + }, + }; +} + +export function getRuntimeInspectCacheDiagnostics(opts: { + readonly cache: RuntimeInspectCache; +}): RuntimeInspectCacheDiagnostics { + return { ...opts.cache.diagnostics }; +} + export function countRunningServices(runtime: RuntimeProject | null): number { if (!runtime) { return 0; @@ -103,6 +141,8 @@ export function filterRuntimeProjects(opts: { // biome-ignore lint/complexity/noExcessiveCognitiveComplexity: Runtime project discovery merges compose state, daemon state, and lifecycle state in one read path. export async function readRuntimeProjects(opts: { readonly includeGlobal: boolean; + readonly inspectCache?: RuntimeInspectCache; + readonly forceInspect?: boolean; }): Promise { const checkedAtMs = Date.now(); if (!findExecutableInPath("docker")) { @@ -153,7 +193,13 @@ export async function readRuntimeProjects(opts: { const ids = baseRows .map((row) => getString(row, "ID") ?? getString(row, "Id") ?? "") .filter((id) => id.length > 0); - const inspectById = await readContainerInspectData({ ids }); + const inspectById = opts.inspectCache + ? await readCachedContainerInspectData({ + cache: opts.inspectCache, + forceInspect: opts.forceInspect ?? true, + ids, + }) + : await readContainerInspectData({ ids }); const globalRoot = resolveGlobalHackDir(); @@ -548,9 +594,6 @@ async function readContainerInspectData(opts: { const res = await exec(["docker", "inspect", ...opts.ids], { stdin: "ignore", }); - if (res.exitCode !== 0) { - return new Map(); - } let parsed: unknown; try { @@ -594,6 +637,50 @@ async function readContainerInspectData(opts: { return out; } +async function readCachedContainerInspectData(opts: { + readonly cache: RuntimeInspectCache; + readonly forceInspect: boolean; + readonly ids: readonly string[]; +}): Promise> { + const currentIds = new Set(opts.ids); + for (const cachedId of opts.cache.entries.keys()) { + if (!currentIds.has(cachedId)) { + opts.cache.entries.delete(cachedId); + } + } + + if (opts.forceInspect) { + opts.cache.diagnostics.fullRefreshes += 1; + } + const missingIds = opts.ids.filter((id) => !opts.cache.entries.has(id)); + const idsToInspect = opts.forceInspect ? opts.ids : missingIds; + opts.cache.diagnostics.cacheHits += opts.forceInspect + ? 0 + : opts.ids.length - missingIds.length; + opts.cache.diagnostics.cacheMisses += missingIds.length; + + if (idsToInspect.length > 0) { + opts.cache.diagnostics.inspectCalls += 1; + opts.cache.diagnostics.inspectIds += idsToInspect.length; + const inspected = await readContainerInspectData({ ids: idsToInspect }); + for (const id of idsToInspect) { + const detail = inspected.get(id); + if (detail) { + opts.cache.entries.set(id, detail); + } + } + } + + const out = new Map(); + for (const id of opts.ids) { + const detail = opts.cache.entries.get(id); + if (detail) { + out.set(id, detail); + } + } + return out; +} + function resolveProjectDirName(workingDir: string): ".hack" | ".dev" | null { if (workingDir.endsWith("/.hack")) { return ".hack"; diff --git a/tests/docker-events.test.ts b/tests/docker-events.test.ts new file mode 100644 index 00000000..65465c76 --- /dev/null +++ b/tests/docker-events.test.ts @@ -0,0 +1,50 @@ +import { expect, test } from "bun:test"; + +import { shouldRefreshForDockerEvent } from "../src/daemon/docker-events.ts"; + +test("docker event filtering ignores actions that cannot change runtime state", () => { + const ignoredActions = [ + "attach", + "commit", + "copy", + "detach", + "exec_create: /bin/sh -c curl --fail http://localhost/health", + "exec_detach", + "exec_die", + "exec_start: /bin/sh -c curl --fail http://localhost/health", + "export", + "resize", + "top", + ]; + + for (const Action of ignoredActions) { + expect(shouldRefreshForDockerEvent({ event: { Action } })).toBe(false); + } +}); + +test("docker event filtering refreshes for state changes and unknown actions", () => { + const relevantActions = [ + "create", + "destroy", + "die", + "health_status: unhealthy", + "kill", + "oom", + "pause", + "rename", + "restart", + "start", + "stop", + "unpause", + "update", + "future_docker_action", + ]; + + for (const Action of relevantActions) { + expect(shouldRefreshForDockerEvent({ event: { Action } })).toBe(true); + } + expect(shouldRefreshForDockerEvent({ event: {} })).toBe(true); + expect(shouldRefreshForDockerEvent({ event: { status: "exec_die" } })).toBe( + false + ); +}); diff --git a/tests/refresh-scheduler.test.ts b/tests/refresh-scheduler.test.ts new file mode 100644 index 00000000..c869d83d --- /dev/null +++ b/tests/refresh-scheduler.test.ts @@ -0,0 +1,72 @@ +import { expect, test } from "bun:test"; + +import { createRefreshScheduler } from "../src/daemon/refresh-scheduler.ts"; + +test("refresh scheduler coalesces an event storm into one bounded follow-up", async () => { + let releaseFirstRefresh = (): void => {}; + const firstRefresh = new Promise((resolve) => { + releaseFirstRefresh = resolve; + }); + const calls: Array<{ reason: string; forceInspect: boolean }> = []; + const scheduler = createRefreshScheduler({ + debounceMs: 2, + minIntervalMs: 2, + maxWaitMs: 10, + refresh: async (request) => { + calls.push(request); + if (calls.length === 1) { + await firstRefresh; + } + }, + }); + + for (let index = 0; index < 100; index += 1) { + scheduler.request({ reason: "event", urgency: "debounced" }); + } + await waitFor({ predicate: () => calls.length === 1 }); + for (let index = 0; index < 100; index += 1) { + scheduler.request({ reason: "event", urgency: "debounced" }); + } + releaseFirstRefresh(); + await waitFor({ predicate: () => calls.length === 2 }); + await Bun.sleep(20); + + expect(calls).toEqual([ + { reason: "event", forceInspect: false }, + { reason: "event", forceInspect: false }, + ]); + scheduler.stop(); +}); + +test("immediate refresh upgrades pending event work and forces inspection", async () => { + const calls: Array<{ reason: string; forceInspect: boolean }> = []; + const scheduler = createRefreshScheduler({ + debounceMs: 20, + minIntervalMs: 20, + maxWaitMs: 40, + refresh: async (request) => { + calls.push(request); + }, + }); + + scheduler.request({ reason: "event", urgency: "debounced" }); + scheduler.request({ reason: "interval", urgency: "immediate" }); + await waitFor({ predicate: () => calls.length === 1 }); + + expect(calls).toEqual([{ reason: "interval", forceInspect: true }]); + scheduler.stop(); +}); + +async function waitFor(opts: { + readonly predicate: () => boolean; + readonly timeoutMs?: number; +}): Promise { + const timeoutMs = opts.timeoutMs ?? 500; + const startedAtMs = Date.now(); + while (!opts.predicate()) { + if (Date.now() - startedAtMs > timeoutMs) { + throw new Error(`Timed out after ${timeoutMs}ms`); + } + await Bun.sleep(1); + } +} diff --git a/tests/runtime-cache.test.ts b/tests/runtime-cache.test.ts index 28796f14..a8ea296b 100644 --- a/tests/runtime-cache.test.ts +++ b/tests/runtime-cache.test.ts @@ -28,21 +28,60 @@ const identityQueue: Array< | { readonly ok: false; readonly error: string } > = []; const autoRegisterCalls: RuntimeProject[][] = []; +const runtimeReadCalls: Array<{ + readonly forceInspect?: boolean; + readonly includeGlobal: boolean; +}> = []; +let runtimeReadOverride: (() => Promise<(typeof runtimeQueue)[number]>) | null = + null; +let autoRegisterError: Error | null = null; const runtimeProjectsMock = await registerScopedModuleMock({ importerPath: import.meta.path, specifier: "../src/lib/runtime-projects.ts", overrides: { - readRuntimeProjects: async () => - runtimeQueue.shift() ?? { - ok: true, - runtime: [], - error: null, - checkedAtMs: Date.now(), + readRuntimeProjects: async (opts: { + readonly forceInspect?: boolean; + readonly includeGlobal: boolean; + }) => { + runtimeReadCalls.push(opts); + if (runtimeReadOverride) { + return await runtimeReadOverride(); + } + return ( + runtimeQueue.shift() ?? { + ok: true, + runtime: [], + error: null, + checkedAtMs: Date.now(), + } + ); + }, + createRuntimeInspectCache: () => ({ + entries: new Map(), + diagnostics: { + inspectCalls: 0, + inspectIds: 0, + cacheHits: 0, + cacheMisses: 0, + fullRefreshes: 0, }, + }), + getRuntimeInspectCacheDiagnostics: () => ({ + inspectCalls: 0, + inspectIds: 0, + cacheHits: 0, + cacheMisses: 0, + fullRefreshes: 0, + }), autoRegisterRuntimeHackProjects: async (opts: { readonly runtime: RuntimeProject[]; }) => { + if (autoRegisterError) { + const error = autoRegisterError; + autoRegisterError = null; + throw error; + } autoRegisterCalls.push(opts.runtime); }, filterRuntimeProjects: (opts: { @@ -88,6 +127,9 @@ beforeEach(() => { runtimeQueue.length = 0; identityQueue.length = 0; autoRegisterCalls.length = 0; + runtimeReadCalls.length = 0; + runtimeReadOverride = null; + autoRegisterError = null; }); afterAll(() => { @@ -556,3 +598,67 @@ test("getPsPayload matches normalized compose project names", async () => { }, ]); }); + +test("runtime cache coalesces concurrent refreshes and preserves forced reconciliation", async () => { + let releaseFirstRead = (): void => {}; + const firstReadGate = new Promise((resolve) => { + releaseFirstRead = resolve; + }); + let readCount = 0; + runtimeReadOverride = async () => { + readCount += 1; + if (readCount === 1) { + await firstReadGate; + } + return { + ok: true, + runtime: [], + error: null, + checkedAtMs: Date.now(), + }; + }; + + const cache = createRuntimeCache({}); + const first = cache.refresh({ reason: "event", forceInspect: false }); + await waitFor({ predicate: () => readCount === 1 }); + const followers = Array.from({ length: 100 }, () => + cache.refresh({ reason: "event", forceInspect: false }) + ); + followers.push(cache.refresh({ reason: "interval", forceInspect: true })); + releaseFirstRead(); + await Promise.all([first, ...followers]); + + expect(readCount).toBe(2); + expect(runtimeReadCalls.map((call) => call.forceInspect)).toEqual([ + false, + true, + ]); +}); + +test("runtime cache clears an unsuccessful refresh task", async () => { + autoRegisterError = new Error("registration failed"); + const cache = createRuntimeCache({}); + + await expect(cache.refresh({ reason: "first" })).rejects.toThrow( + "registration failed" + ); + expect(cache.getDiagnostics().refreshInFlight).toBe(false); + await cache.refresh({ reason: "retry" }); + + expect(runtimeReadCalls).toHaveLength(2); + expect(cache.getSnapshot()?.health.ok).toBe(true); +}); + +async function waitFor(opts: { + readonly predicate: () => boolean; + readonly timeoutMs?: number; +}): Promise { + const timeoutMs = opts.timeoutMs ?? 500; + const startedAtMs = Date.now(); + while (!opts.predicate()) { + if (Date.now() - startedAtMs > timeoutMs) { + throw new Error(`Timed out after ${timeoutMs}ms`); + } + await Bun.sleep(1); + } +} diff --git a/tests/runtime-projects.test.ts b/tests/runtime-projects.test.ts index 6b072cef..3f2d1f48 100644 --- a/tests/runtime-projects.test.ts +++ b/tests/runtime-projects.test.ts @@ -1,25 +1,63 @@ -import { afterAll, beforeAll, expect, test } from "bun:test"; +import { afterAll, beforeAll, beforeEach, expect, test } from "bun:test"; import { registerScopedModuleMock } from "./helpers/scoped-module-mock.ts"; +let dockerAvailable = false; +let currentIds: string[] = []; +let inspectExitCode = 0; +const inspectCalls: string[][] = []; + const shellMock = await registerScopedModuleMock({ importerPath: import.meta.path, specifier: "../src/lib/shell.ts", overrides: { - exec: () => { - throw new Error("exec should not run when docker is unavailable"); + exec: async (command: readonly string[]) => { + if (command[1] === "ps") { + return { + stdout: currentIds + .map((id) => JSON.stringify(makePsRow({ id }))) + .join("\n"), + stderr: "", + exitCode: 0, + }; + } + if (command[1] === "inspect") { + const ids = [...command.slice(2)]; + inspectCalls.push(ids); + const returnedIds = inspectExitCode === 0 ? ids : ids.slice(0, 1); + return { + stdout: JSON.stringify( + returnedIds.map((id) => makeInspectRow({ id })) + ), + stderr: + inspectExitCode === 0 ? "" : "one inspected container disappeared", + exitCode: inspectExitCode, + }; + } + throw new Error(`unexpected command: ${command.join(" ")}`); }, findExecutableInPath: (executableName: string) => - executableName === "docker" ? null : executableName, + executableName === "docker" && !dockerAvailable ? null : executableName, }, }); -const { readRuntimeProjects } = await import("../src/lib/runtime-projects.ts"); +const { + createRuntimeInspectCache, + getRuntimeInspectCacheDiagnostics, + readRuntimeProjects, +} = await import("../src/lib/runtime-projects.ts"); beforeAll(() => { shellMock.activate(); }); +beforeEach(() => { + dockerAvailable = false; + currentIds = []; + inspectExitCode = 0; + inspectCalls.length = 0; +}); + afterAll(() => { shellMock.deactivate(); }); @@ -31,3 +69,102 @@ test("readRuntimeProjects reports docker absence instead of throwing", async () expect(result.runtime).toEqual([]); expect(result.error).toBe("docker is not installed or not on PATH"); }); + +test("runtime inspect cache reuses unchanged IDs and reconciles replacements", async () => { + dockerAvailable = true; + const firstId = "aaaaaaaaaaaa"; + const replacedId = "bbbbbbbbbbbb"; + const replacementId = "cccccccccccc"; + currentIds = [firstId, replacedId]; + const inspectCache = createRuntimeInspectCache(); + + await readRuntimeProjects({ + includeGlobal: true, + inspectCache, + forceInspect: true, + }); + await readRuntimeProjects({ + includeGlobal: true, + inspectCache, + forceInspect: false, + }); + currentIds = [firstId, replacementId]; + await readRuntimeProjects({ + includeGlobal: true, + inspectCache, + forceInspect: false, + }); + await readRuntimeProjects({ + includeGlobal: true, + inspectCache, + forceInspect: true, + }); + + expect(inspectCalls).toEqual([ + [firstId, replacedId], + [replacementId], + [firstId, replacementId], + ]); + expect([...inspectCache.entries.keys()]).toEqual([firstId, replacementId]); + expect(getRuntimeInspectCacheDiagnostics({ cache: inspectCache })).toEqual({ + inspectCalls: 3, + inspectIds: 5, + cacheHits: 3, + cacheMisses: 3, + fullRefreshes: 2, + }); +}); + +test("runtime inspection keeps valid stdout when another container disappears", async () => { + dockerAvailable = true; + currentIds = ["aaaaaaaaaaaa", "missing00000"]; + inspectExitCode = 1; + + const result = await readRuntimeProjects({ includeGlobal: true }); + + expect(result.ok).toBe(true); + const app = result.runtime[0]?.services.get("service-aaaaaaaaaaaa"); + expect(app?.containers[0]?.image).toBe("image:aaaaaaaaaaaa"); + expect(app?.containers[0]?.networks[0]?.name).toBe("hack-dev"); +}); + +function makePsRow(opts: { readonly id: string }): Record { + return { + ID: opts.id, + State: "running", + Status: "Up 10 seconds", + Names: `container-${opts.id}`, + Ports: "3000/tcp", + Labels: [ + "com.docker.compose.project=alpha", + `com.docker.compose.service=service-${opts.id}`, + "com.docker.compose.project.working_dir=/tmp/alpha/.hack", + ].join(","), + }; +} + +function makeInspectRow(opts: { + readonly id: string; +}): Record { + return { + Id: `${opts.id}${"0".repeat(52)}`, + Config: { + Image: `image:${opts.id}`, + Labels: { + "com.docker.compose.project": "alpha", + "com.docker.compose.service": `service-${opts.id}`, + "com.docker.compose.project.working_dir": "/tmp/alpha/.hack", + }, + }, + Mounts: [], + NetworkSettings: { + Networks: { + "hack-dev": { + IPAddress: "172.20.0.2", + Gateway: "172.20.0.1", + Aliases: [`container-${opts.id}`], + }, + }, + }, + }; +} From b8d0b6789b54b0870472d1324d9c00b29bbe1934 Mon Sep 17 00:00:00 2001 From: Dimitri Kennedy Date: Mon, 3 Aug 2026 15:33:28 -0400 Subject: [PATCH 2/3] fix(daemon): enforce refresh cooldown --- src/daemon/refresh-scheduler.ts | 9 +++++--- tests/refresh-scheduler.test.ts | 38 +++++++++++++++++++++++++++++++++ 2 files changed, 44 insertions(+), 3 deletions(-) diff --git a/src/daemon/refresh-scheduler.ts b/src/daemon/refresh-scheduler.ts index 7f2415c1..4c2c914e 100644 --- a/src/daemon/refresh-scheduler.ts +++ b/src/daemon/refresh-scheduler.ts @@ -65,9 +65,12 @@ export function createRefreshScheduler(opts: { const dueAtMs = pending.urgency === "immediate" ? currentTimeMs - : Math.min( - pending.requestedAtMs + maxWaitMs, - Math.max(currentTimeMs + debounceMs, earliestByIntervalMs) + : Math.max( + earliestByIntervalMs, + Math.min( + pending.requestedAtMs + maxWaitMs, + currentTimeMs + debounceMs + ) ); const delayMs = Math.max(0, dueAtMs - currentTimeMs); timer = setTimeout(() => { diff --git a/tests/refresh-scheduler.test.ts b/tests/refresh-scheduler.test.ts index c869d83d..ab46d744 100644 --- a/tests/refresh-scheduler.test.ts +++ b/tests/refresh-scheduler.test.ts @@ -38,6 +38,44 @@ test("refresh scheduler coalesces an event storm into one bounded follow-up", as scheduler.stop(); }); +test("refresh scheduler enforces the minimum interval after a slow refresh", async () => { + const minimumIntervalMs = 30; + let releaseFirstRefresh = (): void => {}; + const firstRefresh = new Promise((resolve) => { + releaseFirstRefresh = resolve; + }); + const startedAtMs: number[] = []; + let firstCompletedAtMs = 0; + const scheduler = createRefreshScheduler({ + debounceMs: 1, + minIntervalMs: minimumIntervalMs, + maxWaitMs: 5, + refresh: async () => { + startedAtMs.push(Date.now()); + if (startedAtMs.length === 1) { + await firstRefresh; + firstCompletedAtMs = Date.now(); + } + }, + }); + + scheduler.request({ reason: "event", urgency: "debounced" }); + await waitFor({ predicate: () => startedAtMs.length === 1 }); + scheduler.request({ reason: "event", urgency: "debounced" }); + await Bun.sleep(10); + releaseFirstRefresh(); + await waitFor({ predicate: () => startedAtMs.length === 2 }); + + const secondStartedAtMs = startedAtMs[1]; + if (secondStartedAtMs === undefined) { + throw new Error("Expected the pending refresh to start"); + } + expect(secondStartedAtMs - firstCompletedAtMs).toBeGreaterThanOrEqual( + minimumIntervalMs - 2 + ); + scheduler.stop(); +}); + test("immediate refresh upgrades pending event work and forces inspection", async () => { const calls: Array<{ reason: string; forceInspect: boolean }> = []; const scheduler = createRefreshScheduler({ From f9169d524c1a192200d4f3fe0e3ce1b401314fca Mon Sep 17 00:00:00 2001 From: Dimitri Kennedy Date: Mon, 3 Aug 2026 15:44:09 -0400 Subject: [PATCH 3/3] fix(daemon): await coalesced refreshes --- src/daemon/runtime-cache.ts | 56 ++++++++++++++++++----------- tests/runtime-cache.test.ts | 70 ++++++++++++++++++++++++++++++++++++- 2 files changed, 105 insertions(+), 21 deletions(-) diff --git a/src/daemon/runtime-cache.ts b/src/daemon/runtime-cache.ts index 6b430554..3fd5c0af 100644 --- a/src/daemon/runtime-cache.ts +++ b/src/daemon/runtime-cache.ts @@ -130,6 +130,11 @@ export type RuntimeCacheDiagnostics = { readonly inspectFullRefreshes: number; }; +type QueuedRefresh = { + readonly reason: string; + readonly forceInspect: boolean; +}; + export function createRuntimeCache(opts: { readonly onRefresh?: (snapshot: RuntimeSnapshot) => void; readonly deps?: { @@ -150,10 +155,7 @@ export function createRuntimeCache(opts: { let snapshot: RuntimeSnapshot | null = null; let refreshTask: Promise | null = null; - let pendingRefresh: { - readonly reason: string; - readonly forceInspect: boolean; - } | null = null; + let pendingRefresh: QueuedRefresh | null = null; let lastRefreshDurationMs: number | null = null; let maxRefreshDurationMs: number | null = null; const inspectCache = createRuntimeInspectCache(); @@ -192,8 +194,36 @@ export function createRuntimeCache(opts: { return; } + refreshTask = drainRefreshes({ + initialRefresh: { reason, forceInspect }, + }); + await refreshTask; + }; + + async function drainRefreshes(opts: { + readonly initialRefresh: QueuedRefresh; + }): Promise { + let nextRefresh: QueuedRefresh | null = opts.initialRefresh; + try { + while (nextRefresh) { + await runRefresh(nextRefresh); + nextRefresh = pendingRefresh; + pendingRefresh = null; + } + } catch (error: unknown) { + pendingRefresh = null; + throw error; + } finally { + refreshTask = null; + } + } + + async function runRefresh({ + reason, + forceInspect, + }: QueuedRefresh): Promise { const startedAtMs = Date.now(); - refreshTask = (async () => { + try { const checkedAtMs = Date.now(); const previousSnapshot = snapshot; const runtimeResult = await readRuntimeProjects({ @@ -237,26 +267,12 @@ export function createRuntimeCache(opts: { health = refreshed.health; snapshot = nextSnapshot; opts.onRefresh?.(nextSnapshot); - })(); - - try { - await refreshTask; - } catch (error: unknown) { - pendingRefresh = null; - throw error; } finally { const durationMs = Math.max(0, Date.now() - startedAtMs); lastRefreshDurationMs = durationMs; maxRefreshDurationMs = Math.max(maxRefreshDurationMs ?? 0, durationMs); - refreshTask = null; - } - - if (pendingRefresh) { - const queuedRefresh = pendingRefresh; - pendingRefresh = null; - await refresh(queuedRefresh); } - }; + } const getProjectsPayload = async ({ filter, diff --git a/tests/runtime-cache.test.ts b/tests/runtime-cache.test.ts index a8ea296b..fc0649b1 100644 --- a/tests/runtime-cache.test.ts +++ b/tests/runtime-cache.test.ts @@ -604,11 +604,17 @@ test("runtime cache coalesces concurrent refreshes and preserves forced reconcil const firstReadGate = new Promise((resolve) => { releaseFirstRead = resolve; }); + let releaseSecondRead = (): void => {}; + const secondReadGate = new Promise((resolve) => { + releaseSecondRead = resolve; + }); let readCount = 0; runtimeReadOverride = async () => { readCount += 1; if (readCount === 1) { await firstReadGate; + } else if (readCount === 2) { + await secondReadGate; } return { ok: true, @@ -624,8 +630,21 @@ test("runtime cache coalesces concurrent refreshes and preserves forced reconcil const followers = Array.from({ length: 100 }, () => cache.refresh({ reason: "event", forceInspect: false }) ); - followers.push(cache.refresh({ reason: "interval", forceInspect: true })); + const forcedFollower = cache.refresh({ + reason: "interval", + forceInspect: true, + }); + let forcedFollowerSettled = false; + void forcedFollower.finally(() => { + forcedFollowerSettled = true; + }); releaseFirstRead(); + await waitFor({ predicate: () => readCount === 2 }); + await Bun.sleep(0); + + expect(forcedFollowerSettled).toBe(false); + releaseSecondRead(); + await forcedFollower; await Promise.all([first, ...followers]); expect(readCount).toBe(2); @@ -635,6 +654,55 @@ test("runtime cache coalesces concurrent refreshes and preserves forced reconcil ]); }); +test("runtime cache reports queued refresh failures to coalesced callers", async () => { + let releaseFirstRead = (): void => {}; + const firstReadGate = new Promise((resolve) => { + releaseFirstRead = resolve; + }); + let readCount = 0; + runtimeReadOverride = async () => { + readCount += 1; + if (readCount === 1) { + await firstReadGate; + return { + ok: true, + runtime: [], + error: null, + checkedAtMs: Date.now(), + }; + } + throw new Error("forced refresh failed"); + }; + + const cache = createRuntimeCache({}); + const first = cache.refresh({ reason: "event", forceInspect: false }); + await waitFor({ predicate: () => readCount === 1 }); + const forcedFollower = cache.refresh({ + reason: "interval", + forceInspect: true, + }); + releaseFirstRead(); + const [firstResult, followerResult] = await Promise.allSettled([ + first, + forcedFollower, + ]); + + expect(firstResult.status).toBe("rejected"); + if ( + followerResult.status !== "rejected" || + !(followerResult.reason instanceof Error) + ) { + throw new Error( + "Expected the coalesced caller to receive the refresh error" + ); + } + expect(followerResult.reason.message).toBe("forced refresh failed"); + expect(runtimeReadCalls.map((call) => call.forceInspect)).toEqual([ + false, + true, + ]); +}); + test("runtime cache clears an unsuccessful refresh task", async () => { autoRegisterError = new Error("registration failed"); const cache = createRuntimeCache({});