diff --git a/src/server.ts b/src/server.ts index 4fd5c7ac6..79d347c92 100644 --- a/src/server.ts +++ b/src/server.ts @@ -199,9 +199,14 @@ async function handleResponses( : `Provider unreachable: ${err instanceof Error ? err.message : String(err)}`; return formatErrorResponse(502, "upstream_error", msg); } - return new Response(relayWithAbort(upstreamResponse.body, upstream), { + const headers = sanitizePassthroughHeaders(upstreamResponse.headers); + const isEventStream = headers.get("content-type")?.toLowerCase().includes("text/event-stream") ?? false; + const body = isEventStream + ? relaySseWithHeartbeat(upstreamResponse.body, upstream) + : relayWithAbort(upstreamResponse.body, upstream); + return new Response(body, { status: upstreamResponse.status, - headers: sanitizePassthroughHeaders(upstreamResponse.headers), + headers, }); } @@ -343,6 +348,56 @@ export function relayWithAbort( }); } +export function relaySseWithHeartbeat( + body: ReadableStream | null, + upstream: AbortController, + heartbeatMs = 15_000, +): ReadableStream | null { + if (!body) return null; + const reader = body.getReader(); + const heartbeat = new TextEncoder().encode(": opencodex keepalive\n\n"); + let timer: ReturnType | undefined; + let closed = false; + + const cleanup = () => { + closed = true; + if (timer) clearInterval(timer); + timer = undefined; + }; + + return new ReadableStream({ + start(controller) { + timer = setInterval(() => { + if (closed) return; + try { + controller.enqueue(heartbeat); + } catch { + cleanup(); + } + }, heartbeatMs); + }, + async pull(controller) { + try { + const { done, value } = await reader.read(); + if (done) { + cleanup(); + controller.close(); + return; + } + controller.enqueue(value); + } catch (err) { + cleanup(); + try { controller.error(err); } catch { /* already torn down */ } + } + }, + cancel(reason) { + cleanup(); + upstream.abort(reason); + reader.cancel(reason).catch(() => {}); + }, + }); +} + /** * Bun's fetch auto-decompresses the response body but leaves the upstream `content-encoding` * (and a now-stale `content-length`) on `response.headers`. Relaying those with the already-decoded diff --git a/src/service.ts b/src/service.ts index 9a27be965..d4264e1f2 100644 --- a/src/service.ts +++ b/src/service.ts @@ -85,6 +85,12 @@ function systemdEnvironmentAssignment(name: string, value: string | undefined): return `Environment=${systemdQuote(`${name}=${value}`)}`; } +function systemdOutputTarget(value: string): string { + // StandardOutput/StandardError use output specifiers such as append:/path. + // Quoting the full specifier makes systemd reject it as an invalid output target. + return value.replace(/%/g, "%%").replace(/\n/g, "\\n"); +} + function sh(cmd: string): string { return execSync(cmd, { encoding: "utf8", stdio: ["pipe", "pipe", "pipe"] }).trim(); } @@ -183,8 +189,8 @@ ExecStart=${systemdQuote(bun)} ${systemdQuote(cli)} start Restart=on-failure RestartSec=5 ${envLines} -StandardOutput=${systemdQuote(`append:${log}`)} -StandardError=${systemdQuote(`append:${log}`)} +StandardOutput=${systemdOutputTarget(`append:${log}`)} +StandardError=${systemdOutputTarget(`append:${log}`)} [Install] WantedBy=default.target diff --git a/tests/passthrough-abort.test.ts b/tests/passthrough-abort.test.ts index b5d05366a..51ee86bb6 100644 --- a/tests/passthrough-abort.test.ts +++ b/tests/passthrough-abort.test.ts @@ -1,5 +1,5 @@ import { describe, expect, test } from "bun:test"; -import { linkAbortSignal, relayWithAbort } from "../src/server"; +import { linkAbortSignal, relaySseWithHeartbeat, relayWithAbort } from "../src/server"; function streamFromChunks(chunks: Uint8Array[]): ReadableStream { let i = 0; @@ -46,6 +46,24 @@ describe("passthrough relayWithAbort (RC2, passthrough path)", () => { expect(ac.signal.aborted).toBe(false); }); + test("SSE passthrough emits heartbeat comments while upstream is silent", async () => { + const ac = new AbortController(); + const body = new ReadableStream({ pull() { return new Promise(() => {}); } }); + const relayed = relaySseWithHeartbeat(body, ac, 5)!; + const reader = relayed.getReader(); + const first = await Promise.race([ + reader.read(), + new Promise((_, reject) => setTimeout(() => reject(new Error("heartbeat timeout")), 200)), + ]); + + expect(first.done).toBe(false); + expect(new TextDecoder().decode(first.value)).toBe(": opencodex keepalive\n\n"); + + await reader.cancel("client gone"); + expect(ac.signal.aborted).toBe(true); + expect(ac.signal.reason).toBe("client gone"); + }); + test("turn-level abort signal aborts the upstream fetch before headers arrive", () => { const upstream = new AbortController(); const turn = new AbortController(); diff --git a/tests/service.test.ts b/tests/service.test.ts new file mode 100644 index 000000000..062414159 --- /dev/null +++ b/tests/service.test.ts @@ -0,0 +1,13 @@ +import { describe, expect, test } from "bun:test"; +import { buildUnit } from "../src/service"; + +describe("systemd service unit", () => { + test("uses unquoted append targets for service logs", () => { + const unit = buildUnit(); + + expect(unit).toContain("StandardOutput=append:"); + expect(unit).toContain("StandardError=append:"); + expect(unit).not.toContain('StandardOutput="append:'); + expect(unit).not.toContain('StandardError="append:'); + }); +});