diff --git a/src/http/executor-server.test.ts b/src/http/executor-server.test.ts index 4151b6f..30182a5 100644 --- a/src/http/executor-server.test.ts +++ b/src/http/executor-server.test.ts @@ -1,3 +1,4 @@ +import type { AddressInfo } from 'node:net'; import { afterEach, describe, expect, it } from 'vitest'; import type { AgentSessions } from '../app/executor/agent.ts'; import type { ComposeExecutor } from '../app/executor/executor.ts'; @@ -178,3 +179,54 @@ describe('repair sessions over HTTP', () => { expect(calls).toEqual([{ op: 'open', request: { project: 'nextcloud', service: 'app' } }]); }); }); + +// Nextcloud 34 took 6 minutes to apply (a 1 GB pull); Node's fetch drops a response that sends no headers for 5 minutes, so the +// driver gave up on an apply that then finished (2026-09-20). The answer now starts at once and is kept alive. +describe('a long operation over HTTP', () => { + async function slowApply(work: () => Promise>) { + const executor: ComposeExecutor = { ...recordingExecutor().executor, apply: work as ComposeExecutor['apply'] }; + const app = await buildExecutorServer({ executor, heartbeatMs: 10 }); + await app.listen({ host: '127.0.0.1', port: 0 }); + closers.push(() => app.close()); + return `http://127.0.0.1:${String((app.server.address() as AddressInfo).port)}`; + } + + it('sends the headers at once and keeps the connection alive while the executor works, and the answer still parses', async () => { + let release = (): void => undefined; + let finished = false; + const url = await slowApply(() => new Promise((resolve) => (release = () => ((finished = true), resolve({ outcome: 'refused', reason: 'not_allowed', detail: 'slow' }))))); + const response = await fetch(`${url}/apply`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify(APPLY) }); + expect(finished).toBe(false); // the apply is still going, yet the response is here + expect(response.status).toBe(200); + await new Promise((resolve) => setTimeout(resolve, 60)); // several heartbeats + release(); + expect(await response.json()).toEqual({ outcome: 'refused', reason: 'not_allowed', detail: 'slow' }); + }); + + it('does not stop the operation when the driver hangs up: an apply half done is worse than an apply that finishes', async () => { + let release = (): void => undefined; + let done = false; + const url = await slowApply(() => new Promise((resolve) => (release = () => ((done = true), resolve({ outcome: 'refused', reason: 'busy', detail: 'x' }))))); + const hangUp = new AbortController(); + const response = await fetch(`${url}/apply`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify(APPLY), signal: hangUp.signal }); + hangUp.abort(); + await response.text().catch(() => undefined); + await new Promise((resolve) => setTimeout(resolve, 40)); + expect(done).toBe(false); + release(); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(done).toBe(true); + }); + + it('an unexpected failure after the headers went out is still an answer the client can read, as a failure of the executor', async () => { + const url = await slowApply(() => Promise.reject(new Error('disk failure at /secret/path'))); + const response = await fetch(`${url}/apply`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify(APPLY) }); + expect(await response.json()).toEqual({ error: 'internal', message: 'unexpected executor error' }); + }); + + it('a request that is not valid is still refused with 400 before anything starts', async () => { + const url = await slowApply(() => Promise.resolve({ outcome: 'refused', reason: 'busy', detail: 'x' })); + const response = await fetch(`${url}/apply`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ project: 'x' }) }); + expect(response.status).toBe(400); + }); +}); diff --git a/src/http/executor-server.ts b/src/http/executor-server.ts index 9e984d1..78ac02d 100644 --- a/src/http/executor-server.ts +++ b/src/http/executor-server.ts @@ -17,6 +17,7 @@ import { SessionOpenRequest, } from '../domain/executor-api.ts'; import type { ExecutorResponse } from '../domain/executor-api.ts'; +import { answerWhenDone } from './keep-alive.ts'; // The executor's HTTP surface on the internal network of the driver, the executor and the model. The driver's routes (one // operation on one service each, and opening/closing a repair session) answer only the driver's address; the model's `hl` @@ -29,6 +30,8 @@ export interface ExecutorServerDeps { /** Whether a remote address is the driver's; without it every address may call the driver's routes (tests, local runs). */ readonly isDriver?: (address: string) => Promise; readonly logger?: boolean; + /** How often a running operation sends a space to keep the connection alive (http/keep-alive.ts). */ + readonly heartbeatMs?: number; } /** Requests are a few hundred bytes; anything much larger is not from the driver. The model writes whole compose files. */ @@ -54,11 +57,19 @@ export async function buildExecutorServer(deps: ExecutorServerDeps): Promise void }, cause: unknown) => { + log.error({ err: cause }, 'request failed'); + return { error: 'internal', message: 'unexpected executor error' }; + }; + const route = (url: string, schema: z.ZodType, run: (request: T) => Promise>) => { app.post(url, async (request, reply) => { const parsed = schema.safeParse(request.body); if (!parsed.success) return badRequest(reply, issues(parsed.error)); - return reply.send(await run(parsed.data)); + // Never stopped when the driver hangs up: an apply half done is worse than an apply that finishes (its lock says so). + await answerWhenDone(reply, heartbeat, () => run(parsed.data), (cause) => internalFailure(request.log, cause)); }); }; @@ -88,7 +99,7 @@ export async function buildExecutorServer(deps: ExecutorServerDeps): Promise run(token, parsed.data), (cause) => internalFailure(request.log, cause)); }); }; const Nothing = z.strictObject({}); diff --git a/src/http/keep-alive.ts b/src/http/keep-alive.ts new file mode 100644 index 0000000..3f11417 --- /dev/null +++ b/src/http/keep-alive.ts @@ -0,0 +1,33 @@ +import type { FastifyReply } from 'fastify'; + +// Node's fetch drops a response that sends no headers, or no new body bytes, for 5 minutes. Some answers here take longer +// (a model session, an apply that pulls a large image), and the caller then gives up on work that goes on and finishes. +// So the response starts at once (200, JSON) and a space goes out every `heartbeatMs`: JSON allows leading whitespace, so +// the caller's `response.json()` reads the answer as it always did. The status is 200 whatever happens, so a failure travels +// in the body: the caller decides what an answer means, as it already does. + +export const HEARTBEAT_MS = 30_000; + +/** + * Answers `reply` with what `work` returns, as JSON, kept alive meanwhile. `hangUp` fires when the caller closes the + * connection before the answer: `work` decides whether that stops it (a model run: yes; an apply: never, half done is worse). + * A `work` that throws is answered with `failure(cause)`. + */ +export async function answerWhenDone(reply: FastifyReply, options: { readonly heartbeatMs?: number }, work: (hangUp: AbortSignal) => Promise, failure: (cause: unknown) => unknown): Promise { + const { raw } = reply; + reply.hijack(); + raw.writeHead(200, { 'content-type': 'application/json; charset=utf-8' }); + raw.flushHeaders(); + const hangUp = new AbortController(); + raw.once('close', () => { + if (!raw.writableFinished) hangUp.abort(new Error('the caller closed the connection')); + }); + const heartbeat = setInterval(() => raw.write(' '), options.heartbeatMs ?? HEARTBEAT_MS); + try { + raw.end(JSON.stringify(await work(hangUp.signal))); + } catch (cause) { + raw.end(JSON.stringify(failure(cause))); + } finally { + clearInterval(heartbeat); + } +} diff --git a/src/http/model-server.ts b/src/http/model-server.ts index 87013bf..656b7d3 100644 --- a/src/http/model-server.ts +++ b/src/http/model-server.ts @@ -2,6 +2,7 @@ import Fastify from 'fastify'; import type { FastifyInstance } from 'fastify'; import type { ClaudeRunRequest, ClaudeRunResult } from '../app/model/claude.ts'; import { ModelRunRequest } from '../domain/model-api.ts'; +import { answerWhenDone } from './keep-alive.ts'; import type { ModelRunResponse } from '../domain/model-api.ts'; // The model sidecar's HTTP surface, for the driver on the internal network: run one prompt, one at a time. @@ -37,28 +38,19 @@ export async function buildModelServer(deps: ModelServerDeps): Promise `${issue.path.join('.')}: ${issue.message}`).join('; ') }); if (busy) return reply.send({ outcome: 'failed', kind: 'busy', detail: 'another model run is in progress' } satisfies ModelRunResponse); busy = true; - const { raw } = reply; - reply.hijack(); - raw.writeHead(200, { 'content-type': 'application/json; charset=utf-8' }); - raw.flushHeaders(); - const hangUp = new AbortController(); - raw.once('close', () => { - if (!raw.writableFinished) hangUp.abort(new Error('the driver closed the connection')); - }); - const heartbeat = setInterval(() => raw.write(' '), deps.heartbeatMs ?? 30_000); - try { - const { executor, profile, prompt, output_schema: outputSchema, session_token: sessionToken } = parsed.data; - const result = await deps.run({ model: executor, profile, prompt, outputSchema, timeoutMs: deps.timeoutMs, ...(sessionToken === undefined ? {} : { sessionToken }) }, hangUp.signal); - const answer: ModelRunResponse = result.ok ? { outcome: 'done', output: result.output, cost_usd: result.costUsd, turns: result.turns } : { outcome: 'failed', kind: result.kind, detail: result.detail, ...(result.costUsd === undefined ? {} : { cost_usd: result.costUsd }) }; - raw.end(JSON.stringify(answer)); - } catch (cause) { + await answerWhenDone( + reply, + { ...(deps.heartbeatMs === undefined ? {} : { heartbeatMs: deps.heartbeatMs }) }, + async (hangUp) => { + const { executor, profile, prompt, output_schema: outputSchema, session_token: sessionToken } = parsed.data; + const result = await deps.run({ model: executor, profile, prompt, outputSchema, timeoutMs: deps.timeoutMs, ...(sessionToken === undefined ? {} : { sessionToken }) }, hangUp); + return result.ok ? ({ outcome: 'done', output: result.output, cost_usd: result.costUsd, turns: result.turns } satisfies ModelRunResponse) : ({ outcome: 'failed', kind: result.kind, detail: result.detail, ...(result.costUsd === undefined ? {} : { cost_usd: result.costUsd }) } satisfies ModelRunResponse); + }, // The status line is already sent, so a failure travels in the body like every other one. - const detail = cause instanceof Error ? cause.message : String(cause); - raw.end(JSON.stringify({ outcome: 'failed', kind: 'unavailable', detail } satisfies ModelRunResponse)); - } finally { - clearInterval(heartbeat); + (cause) => ({ outcome: 'failed', kind: 'unavailable', detail: cause instanceof Error ? cause.message : String(cause) }) satisfies ModelRunResponse, + ).finally(() => { busy = false; - } + }); }); app.get('/health', () => ({ ok: true })); diff --git a/src/infra/http-executor-client.test.ts b/src/infra/http-executor-client.test.ts index cf7d81b..7ddf803 100644 --- a/src/infra/http-executor-client.test.ts +++ b/src/infra/http-executor-client.test.ts @@ -85,6 +85,11 @@ describe('HTTP executor client failures', () => { expect(await client.backup({ project: 'jackett' })).toEqual({ outcome: 'unavailable', detail: 'executor failed (500): unexpected executor error' }); }); + it('a failure that comes after the executor already answered 200 (it keeps the connection alive) is read from the body', async () => { + const client = createHttpExecutorClient({ baseUrl: 'http://executor:8081', fetch: answering({ error: 'internal', message: 'unexpected executor error' }, 200) }); + expect(await client.backup({ project: 'jackett' })).toEqual({ outcome: 'unavailable', detail: 'executor failed: unexpected executor error' }); + }); + it('an abort is not swallowed: the driver’s time limit stays in charge', async () => { const controller = new AbortController(); const client = createHttpExecutorClient({ diff --git a/src/infra/http-executor-client.ts b/src/infra/http-executor-client.ts index 6107e3e..c957ce2 100644 --- a/src/infra/http-executor-client.ts +++ b/src/infra/http-executor-client.ts @@ -53,6 +53,8 @@ export function createHttpExecutorClient(options: HttpExecutorClientOptions): Ex } if (response.status >= 400 && response.status < 500) return { outcome: 'unavailable', detail: `executor refused the request (${String(response.status)}): ${messageOf(payload)}` }; if (!response.ok) return { outcome: 'unavailable', detail: `executor failed (${String(response.status)}): ${messageOf(payload)}` }; + // The executor answers 200 at once and keeps the connection alive, so a failure that comes later is in the body. + if (typeof payload === 'object' && payload !== null && 'error' in payload) return { outcome: 'unavailable', detail: `executor failed: ${messageOf(payload)}` }; const parsed = schema.safeParse(payload); if (!parsed.success) return { outcome: 'unavailable', detail: `executor answered ${path} in an unexpected shape: ${parsed.error.issues[0]?.message ?? 'invalid'}` }; return parsed.data;