From 081bc3f956c6c4d0095afc1aa70c4058065073d8 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 3 Oct 2026 14:39:02 +0000 Subject: [PATCH] fix(api): let large uploads finish and abort stalled ones Node's default 5-minute requestTimeout cut off a 500 MiB video upload on any uplink slower than about 14 Mbit/s, so the video cap was unusable over ordinary connections. - The API server's requestTimeout is 30 minutes, enough for the largest upload at about 2.5 Mbit/s. headersTimeout keeps its 60-second default. - Each streamed upload aborts when the client sends no bytes for 60 seconds: the route answers 408, the partial file is deleted and no row is written, so a slow-drip client cannot hold a connection for the full request timeout. Time spent writing to disk does not count. - The store and the upload guard no longer wait for a body to confirm its cancellation, which a stalled or hostile client may never do. - A guard test checks that every file the executor attaches records the thread's group, which the render_plot chart fallback once dropped. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01MRtGG9ARfqqhma3t85Ro2u --- apps/api/src/artifact-http.test.ts | 123 +++++++++++++++++- apps/api/src/artifact-http.ts | 56 +++++++- apps/api/src/index.ts | 15 ++- packages/adapters/src/artifacts.ts | 5 +- .../adapters/src/thread-artifacts.test.ts | 15 +++ 5 files changed, 205 insertions(+), 9 deletions(-) diff --git a/apps/api/src/artifact-http.test.ts b/apps/api/src/artifact-http.test.ts index 0a2a6fc..d95711f 100644 --- a/apps/api/src/artifact-http.test.ts +++ b/apps/api/src/artifact-http.test.ts @@ -1,8 +1,20 @@ +import { mkdtemp, readdir, rm } from "node:fs/promises"; +import { createServer } from "node:http"; +import { tmpdir } from "node:os"; +import path from "node:path"; import type { ArtifactStore } from "@rakazo/adapter-kit"; +import { LocalArtifactStore } from "@rakazo/adapters"; import type { PrismaClient } from "@rakazo/db"; import { Hono } from "hono"; -import { describe, expect, it, vi } from "vitest"; -import { contentDisposition, mountArtifactHttpRoutes, parseByteRange } from "./artifact-http.js"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { + API_SERVER_OPTIONS, + contentDisposition, + mountArtifactHttpRoutes, + parseByteRange, + stallGuardedChunks, + UploadStalledError, +} from "./artifact-http.js"; describe("parseByteRange", () => { it("ignores an absent header or another unit", () => { @@ -154,3 +166,110 @@ describe("artifact HTTP routes", () => { expect(artifacts.putStream).not.toHaveBeenCalled(); }); }); + +describe("upload timeouts", () => { + const dirs: string[] = []; + afterEach(async () => { + await Promise.all(dirs.splice(0).map((dir) => rm(dir, { recursive: true, force: true }))); + }); + + it("gives the server time for large uploads but keeps the header timeout", () => { + const server = createServer(API_SERVER_OPTIONS); + expect(server.requestTimeout).toBe(30 * 60 * 1000); + expect(server.headersTimeout).toBe(createServer().headersTimeout); + }); + + /** A body that sends `chunks` at `gapMs` intervals, then stays open forever if `hang`. */ + function drip(chunks: string[], gapMs: number, hang: boolean) { + const cancel = vi.fn(() => new Promise(() => undefined)); + let sent = 0; + const body = new ReadableStream({ + async pull(controller) { + if (sent === chunks.length) { + if (!hang) controller.close(); + else await new Promise(() => undefined); + return; + } + if (sent > 0) await new Promise((resolve) => setTimeout(resolve, gapMs)); + controller.enqueue(new TextEncoder().encode(chunks[sent++])); + }, + cancel, + }); + return { body, cancel }; + } + + it("throws once the body stops sending and cancels it without waiting", async () => { + const { body, cancel } = drip(["ab", "cd"], 5, true); + const seen: string[] = []; + await expect( + (async () => { + for await (const chunk of stallGuardedChunks(body, 50)) { + seen.push(new TextDecoder().decode(chunk)); + } + })(), + ).rejects.toBeInstanceOf(UploadStalledError); + expect(seen).toEqual(["ab", "cd"]); + expect(cancel).toHaveBeenCalled(); + }); + + async function postDrip(stallMs: number, body: ReadableStream) { + const root = await mkdtemp(path.join(tmpdir(), "rakazo-artifact-upload-")); + dirs.push(root); + const create = vi.fn(async ({ data }: { data: Record }) => ({ + id: "art-1", + createdAt: new Date(), + ...data, + })); + const prisma = { + artifact: { create }, + spaceMember: { + findFirst: vi.fn(async () => ({ + userId: "user-1", + spaceId: "space-1", + member: { user: { email: "user@example.test" } }, + })), + }, + deploymentSettings: { findUnique: vi.fn(async () => null) }, + bot: { findFirst: vi.fn(async () => ({ id: "bot-1" })) }, + }; + const hono = new Hono(); + mountArtifactHttpRoutes(hono, { + prisma: prisma as unknown as PrismaClient, + artifacts: new LocalArtifactStore(root), + sessionUserId: async () => "user-1", + trustedOrigin: () => true, + uploadStallTimeoutMs: stallMs, + }); + const response = await hono.request( + new Request("http://localhost/api/artifacts?botId=bot-1&name=clip.mp4", { + method: "POST", + headers: { "content-type": "video/mp4" }, + body, + duplex: "half", + } as RequestInit & { duplex: "half" }), + ); + // The store creates the space directory before it reads the body, so it must exist here. + const stored = await readdir(path.join(root, "artifacts", "space-1")); + return { response, create, stored }; + } + + it("aborts a stalled upload with 408 and deletes the partial file", async () => { + const { body, cancel } = drip(["partial video bytes"], 0, true); + const { response, create, stored } = await postDrip(50, body); + expect(response.status).toBe(408); + expect(await response.json()).toEqual({ error: "Upload stalled." }); + expect(cancel).toHaveBeenCalled(); + expect(create).not.toHaveBeenCalled(); + expect(stored).toEqual([]); + }); + + it("keeps a slow but steady upload going past the stall timeout", async () => { + const { body } = drip(["a", "b", "c", "d", "e", "f"], 30, false); + const { response, create, stored } = await postDrip(100, body); + expect(response.status).toBe(201); + expect(create).toHaveBeenCalledWith({ + data: expect.objectContaining({ size: 6, mimeType: "video/mp4", botId: "bot-1" }), + }); + expect(stored).toHaveLength(1); + }); +}); diff --git a/apps/api/src/artifact-http.ts b/apps/api/src/artifact-http.ts index d8b9df5..f74a0ea 100644 --- a/apps/api/src/artifact-http.ts +++ b/apps/api/src/artifact-http.ts @@ -15,6 +15,52 @@ import { cancelBody } from "./http-body.js"; export const ARTIFACT_HTTP_PATH = "/api/artifacts"; +/** + * Node's default 5-minute requestTimeout cuts off a 500 MiB video upload on any uplink slower than + * about 14 Mbit/s; 30 minutes fits it at about 2.5 Mbit/s. headersTimeout keeps its default, and + * the per-upload stall timeout stops a slow-drip client from holding a connection that long. + */ +export const API_SERVER_OPTIONS = { requestTimeout: 30 * 60 * 1000 }; +/** An upload that sends no bytes for this long is aborted and its partial file deleted. */ +export const ARTIFACT_UPLOAD_STALL_TIMEOUT_MS = 60_000; + +export class UploadStalledError extends Error { + constructor() { + super("Upload stalled"); + this.name = "UploadStalledError"; + } +} + +/** + * Yields the body's chunks and throws UploadStalledError when the next chunk takes longer than + * stallMs. Only time spent waiting for the client counts, not time spent writing to disk. + */ +export async function* stallGuardedChunks( + body: ReadableStream, + stallMs: number, +): AsyncGenerator { + const reader = body.getReader(); + try { + for (;;) { + let timer: ReturnType | undefined; + const stalled = new Promise((_, reject) => { + timer = setTimeout(() => reject(new UploadStalledError()), stallMs); + }); + try { + const { done, value } = await Promise.race([reader.read(), stalled]); + if (done) return; + yield value; + } finally { + clearTimeout(timer); + } + } + } finally { + // Never wait on the client here: a stalled or hostile body may never confirm the cancel. + void reader.cancel().catch(() => undefined); + reader.releaseLock(); + } +} + export type ArtifactHttpDeps = { prisma: PrismaClient; artifacts: ArtifactStore; @@ -22,6 +68,8 @@ export type ArtifactHttpDeps = { sessionUserId: (request: Request) => Promise; /** Same rule as CORS. Raw uploads can be CORS "simple" requests, so the origin is checked. */ trustedOrigin: (origin: string) => boolean; + /** Overrides ARTIFACT_UPLOAD_STALL_TIMEOUT_MS (tests). */ + uploadStallTimeoutMs?: number; }; /** Generated ids only; anything else is not an artifact. */ @@ -210,7 +258,7 @@ export function mountArtifactHttpRoutes(app: Hono, deps: ArtifactHttpDeps) { app.post(ARTIFACT_HTTP_PATH, async (c) => { const request = c.req.raw; - const reject = (status: 400 | 401 | 403 | 404 | 413 | 415, error: string) => { + const reject = (status: 400 | 401 | 403 | 404 | 408 | 413 | 415, error: string) => { cancelBody(request.body); return c.json({ error }, status); }; @@ -266,12 +314,16 @@ export function mountArtifactHttpRoutes(app: Hono, deps: ArtifactHttpDeps) { runId: null, name, mimeType, - stream: request.body, + stream: stallGuardedChunks( + request.body, + deps.uploadStallTimeoutMs ?? ARTIFACT_UPLOAD_STALL_TIMEOUT_MS, + ), maxBytes, signal: request.signal, }); } catch (error) { if (error instanceof ArtifactTooLargeError) return reject(413, "File is too large."); + if (error instanceof UploadStalledError) return reject(408, "Upload stalled."); throw error; } if (row.size === 0) { diff --git a/apps/api/src/index.ts b/apps/api/src/index.ts index ba0a224..6933a28 100644 --- a/apps/api/src/index.ts +++ b/apps/api/src/index.ts @@ -7,6 +7,7 @@ import { serve } from "@hono/node-server"; import { SERVICE_NAMES } from "@rakazo/logging"; import { createRootLogger } from "@rakazo/logging/axiom"; import { createApp } from "./app.js"; +import { API_SERVER_OPTIONS } from "./artifact-http.js"; import { loadEnv } from "./env.js"; const logger = createRootLogger(SERVICE_NAMES.api); @@ -14,9 +15,17 @@ const logger = createRootLogger(SERVICE_NAMES.api); try { const env = loadEnv(); const { app, stop } = await createApp({ ...env, logger }); - const server = serve({ fetch: app.fetch, port: env.port, hostname: env.apiHost }, () => { - logger.info("api listening", { "http.host": env.apiHost, "http.port": env.port }); - }); + const server = serve( + { + fetch: app.fetch, + port: env.port, + hostname: env.apiHost, + serverOptions: API_SERVER_OPTIONS, + }, + () => { + logger.info("api listening", { "http.host": env.apiHost, "http.port": env.port }); + }, + ); // Long-lived connections (threads.subscribe SSE streams) never end on their // own, so server.close() alone waits forever for them. Track sockets and diff --git a/packages/adapters/src/artifacts.ts b/packages/adapters/src/artifacts.ts index 4ba47ef..ddf2ae1 100644 --- a/packages/adapters/src/artifacts.ts +++ b/packages/adapters/src/artifacts.ts @@ -169,8 +169,9 @@ async function* byteChunks( yield value; } } finally { - // Stop the producer (e.g. an upload body) when the store gives up early. - await reader.cancel().catch(() => undefined); + // Stop the producer (e.g. an upload body) when the store gives up early, without waiting on + // a producer that may never confirm. + void reader.cancel().catch(() => undefined); reader.releaseLock(); } } diff --git a/packages/adapters/src/thread-artifacts.test.ts b/packages/adapters/src/thread-artifacts.test.ts index fb79536..1e0c90f 100644 --- a/packages/adapters/src/thread-artifacts.test.ts +++ b/packages/adapters/src/thread-artifacts.test.ts @@ -1,3 +1,4 @@ +import { readFileSync } from "node:fs"; import type { AdapterContext, ArtifactStore, @@ -229,3 +230,17 @@ describe("current-turn thread files", () => { expect(attachableFileMaxBytes(buffered, "video/mp4")).toBe(10 * 1024 * 1024); }); }); + +describe("executor attachments", () => { + it("records the thread's group on every file the executor attaches", () => { + // Driving render_plot's PNG fallback needs a full run; this guards the wiring that once + // dropped groupId, so the chart could not be opened in group conversations. + const source = readFileSync(new URL("./executor.ts", import.meta.url), "utf8"); + const calls = source.split("attachWorkspaceFileToThread(").slice(1); + expect(calls.length).toBeGreaterThanOrEqual(2); + for (const call of calls) { + const input = call.slice(0, call.indexOf("operationId:")); + expect(input).toContain("groupId: thread.groupId ?? undefined"); + } + }); +});