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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01MRtGG9ARfqqhma3t85Ro2u

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<void>(() => undefined));
+    let sent = 0;
+    const body = new ReadableStream<Uint8Array>({
+      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<Uint8Array>) {
+    const root = await mkdtemp(path.join(tmpdir(), "rakazo-artifact-upload-"));
+    dirs.push(root);
+    const create = vi.fn(async ({ data }: { data: Record<string, unknown> }) => ({
+      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);
+  });
+});

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<Uint8Array>,
+  stallMs: number,
+): AsyncGenerator<Uint8Array> {
+  const reader = body.getReader();
+  try {
+    for (;;) {
+      let timer: ReturnType<typeof setTimeout> | undefined;
+      const stalled = new Promise<never>((_, 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<string | null>;
   /** 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) {

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

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();
   }
 }

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");
+    }
+  });
+});