diff --git a/compute/core/src/control-plane/sse.ts b/compute/core/src/control-plane/sse.ts index 712f6af..dbd17f1 100644 --- a/compute/core/src/control-plane/sse.ts +++ b/compute/core/src/control-plane/sse.ts @@ -3,6 +3,8 @@ export interface SseFrameInput { id?: string | number; data?: T; comment?: string; + /** Reconnection delay (ms) sent to the client EventSource as a `retry:` line. */ + retry?: number; } export function encodeSseFrame(input: SseFrameInput): string { @@ -10,6 +12,9 @@ export function encodeSseFrame(input: SseFrameInput): string { if (typeof input.comment === 'string') { lines.push(`: ${input.comment}`); } + if (typeof input.retry === 'number' && Number.isFinite(input.retry)) { + lines.push(`retry: ${Math.max(0, Math.floor(input.retry))}`); + } if (typeof input.id !== 'undefined') { lines.push(`id: ${String(input.id)}`); } diff --git a/compute/core/tests/control-plane/sse.test.ts b/compute/core/tests/control-plane/sse.test.ts index e207950..9b8e5b9 100644 --- a/compute/core/tests/control-plane/sse.test.ts +++ b/compute/core/tests/control-plane/sse.test.ts @@ -23,4 +23,14 @@ describe('sse codec', () => { expect(parseSseEventId(frame)).toBe(5); expect(parseSsePayload(frame)).toBe('line1\nline2'); }); + + test('emits a retry directive when provided', () => { + const frame = encodeSseFrame({ retry: 120_000 }); + expect(frame).toContain('retry: 120000'); + }); + + test('omits retry when not finite and floors fractional values', () => { + expect(encodeSseFrame({ retry: Number.NaN })).not.toContain('retry:'); + expect(encodeSseFrame({ retry: 1500.9 })).toContain('retry: 1500'); + }); }); diff --git a/compute/worker/src/runtime.ts b/compute/worker/src/runtime.ts index 442df53..040e215 100644 --- a/compute/worker/src/runtime.ts +++ b/compute/worker/src/runtime.ts @@ -83,6 +83,11 @@ const COMPUTE_STATE_TTL_MS = 24 * 60 * 60 * 1000; const LOOP_ERROR_BACKOFF_MS = 500; const RUNNING_HEARTBEAT_MS = 5000; const OP_EVENTS_KEEPALIVE_MS = 15_000; +// Reconnection delay handed to the browser EventSource via the SSE `retry:` +// directive. When a silent stream is torn down for idle sleep, this keeps the +// client from immediately reconnecting and re-waking the worker; instead it +// reconnects on a slow cadence so the container stays asleep most of the time. +const OP_EVENTS_RECONNECT_HINT_MS = 120_000; const DOCUMENT_ID_REGEX = /^[a-f0-9]{64}$/i; const SAFE_NAMESPACE_REGEX = /^[a-zA-Z0-9._-]{1,128}$/; const WHISPER_MAX_DELIVER = 1; @@ -588,7 +593,11 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti if (idleTimer) return; idleTimer = setInterval(() => { if (!session || stopping) return; - if (inFlightHttp > 0 || activeSse > 0 || inFlightJobs > 0) return; + // Hard work in flight always blocks idle. An open SSE no longer blocks just + // by existing — only by delivering events, which refresh lastActivityAt via + // markActivity() in the stream's onEvent. So a silent/stuck-op stream lets + // the idle window elapse and is torn down by disconnect() below. + if (inFlightHttp > 0 || inFlightJobs > 0) return; if (Date.now() - lastActivityAt < IDLE_DISCONNECT_MS) return; void disconnect('idle'); }, IDLE_CHECK_INTERVAL_MS); @@ -612,6 +621,15 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti async function disconnect(reason: string): Promise { const current = session; if (!current) return; + // Snapshot what's still attached so a dropped connection is visible in logs + // (e.g. SSE streams torn down because we went idle while they were open). + app.log.info({ + reason, + activeSse, + inFlightHttp, + inFlightJobs, + idleForMs: Date.now() - lastActivityAt, + }, 'nats dropping connection'); // Clear synchronously (before any await) so concurrent requests reconnect a // fresh session instead of using the connection we're about to close. session = null; @@ -1032,6 +1050,10 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti reply.raw.setHeader('Cache-Control', 'no-cache, no-transform'); reply.raw.setHeader('Connection', 'keep-alive'); reply.raw.setHeader('X-Accel-Buffering', 'no'); + // Tell the browser EventSource to back off before reconnecting. If this stream + // is torn down because the worker went idle (NATS dropped), we don't want the + // client to reconnect immediately and re-wake the container. + reply.raw.write(encodeSseFrame({ retry: OP_EVENTS_RECONNECT_HINT_MS })); let closed = false; let unsubscribe: (() => void) | null = null; @@ -1095,6 +1117,9 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti if (nextSignature !== signature) { current = event.snapshot; signature = nextSignature; + // A real event is progress: refresh the idle window so an actively + // streaming op keeps the worker awake. A silent stream does not. + markActivity(); writeSnapshot(current, event.eventId); } if (isTerminalStatus(event.snapshot.status)) {