fix(web): Infra request limits per account and address, retire empty uploads, wait out a busy farm

apps/web/__tests__/unit/developer-sdk-upload-billing.test.ts

@@ -19,10 +19,12 @@ const mocks = vi.hoisted(() => ({
 	listParts: vi.fn(),
 	headObject: vi.fn(),
 	deleteObject: vi.fn(),
+	listUploads: vi.fn(),
 	unrecorded: [] as { id: string; appId: string; s3Key: string | null }[],
 	deletedAt: null as Date | null,
 	maxTier: "1080p" as string,
 	status: "uploading" as string,
+	claimedAt: new Date(0),
 	creditTransactions: [] as Record<string, unknown>[],
 	videoUpdates: [] as Record<string, unknown>[],
 	admission: vi.fn(),
@@ -89,6 +91,9 @@ vi.mock("@cap/database", async () => {
 				},
 			];
 		}
+		if (keys === "status,updatedAt") {
+			return [{ status: mocks.status, updatedAt: mocks.claimedAt }];
+		}
 		if (keys === "deletedAt,duration") {
 			return [
 				{
@@ -204,6 +209,7 @@ vi.mock("@cap/web-backend", async () => {
 						multipart: {
 							complete: mocks.complete,
 							listParts: mocks.listParts,
+							listUploads: mocks.listUploads,
 						},
 					},
 				]),
@@ -758,6 +764,8 @@ describe("removing files that were assembled but never recorded", () => {
 	const old = () => Effect.succeed({ LastModified: hoursAgo(25) });
 
 	beforeEach(() => {
+		mocks.status = "uploading";
+		mocks.claimedAt = hoursAgo(25);
 		mocks.storedDuration = null;
 		mocks.deletedAt = null;
 		mocks.events = [];
@@ -768,6 +776,8 @@ describe("removing files that were assembled but never recorded", () => {
 			return Effect.succeed({});
 		});
 		mocks.headObject.mockReset();
+		mocks.listUploads.mockReset();
+		mocks.listUploads.mockImplementation(() => Effect.succeed({ Uploads: [] }));
 		mocks.admission.mockReset();
 		mocks.admission.mockResolvedValue(admitted);
 		mocks.claim.mockReset();
@@ -902,6 +912,159 @@ describe("removing files that were assembled but never recorded", () => {
 		expect(mocks.videoUpdates).toEqual([]);
 	});
 
+	const missing = () =>
+		Effect.fail({ _tag: "S3Error", cause: { name: "NotFound" } });
+
+	it("retires a day-old video that never received a file", async () => {
+		mocks.unrecorded = [row("empty")];
+		mocks.headObject.mockImplementation(missing);
+
+		expect(await removeUnrecordedInfraFiles(now)).toBe(1);
+
+		expect(mocks.events).toEqual(["lock", "lock", "retire"]);
+		expect(mocks.videoUpdates).toEqual([{ deletedAt: now }]);
+		expect(mocks.deleteObject).not.toHaveBeenCalled();
+		expect(mocks.listUploads).toHaveBeenCalledTimes(2);
+		expect(mocks.listUploads).toHaveBeenCalledWith({
+			prefix: "developer/app-1/empty/video",
+		});
+	});
+
+	it("does not retire a video whose completion assembled the file before the sweep reached the lock", async () => {
+		mocks.unrecorded = [row("finishing")];
+		mocks.headObject
+			.mockImplementationOnce(missing)
+			.mockImplementation(() => Effect.succeed({ LastModified: now }));
+
+		expect(await removeUnrecordedInfraFiles(now)).toBe(0);
+		expect(mocks.videoUpdates).toEqual([]);
+		expect(mocks.deletedAt).toBe(null);
+		expect(mocks.deleteObject).not.toHaveBeenCalled();
+	});
+
+	it("retires a video with no file whose completion died holding its claim", async () => {
+		mocks.unrecorded = [row("abandoned")];
+		mocks.status = "processing";
+		mocks.claimedAt = hoursAgo(1);
+		mocks.headObject.mockImplementation(missing);
+
+		expect(await removeUnrecordedInfraFiles(now)).toBe(1);
+		expect(mocks.videoUpdates).toEqual([{ deletedAt: now }]);
+	});
+
+	it("leaves a video with no file alone while a completion's claim is recent", async () => {
+		mocks.unrecorded = [row("claimed")];
+		mocks.status = "processing";
+		mocks.claimedAt = new Date(now.getTime() - 60_000);
+		mocks.headObject.mockImplementation(missing);
+
+		expect(await removeUnrecordedInfraFiles(now)).toBe(0);
+		expect(mocks.videoUpdates).toEqual([]);
+		expect(mocks.deletedAt).toBe(null);
+	});
+
+	it("does not retire a video whose upload is assembled between storage's file read and its upload listing", async () => {
+		mocks.unrecorded = [row("assembling")];
+		let assembled = false;
+		let listings = 0;
+		mocks.headObject.mockImplementation(() =>
+			assembled ? Effect.succeed({ LastModified: now }) : missing(),
+		);
+		mocks.listUploads.mockImplementation(() => {
+			listings++;
+			if (listings === 2) assembled = true;
+			return Effect.succeed({ Uploads: [] });
+		});
+
+		expect(await removeUnrecordedInfraFiles(now)).toBe(0);
+		expect(mocks.videoUpdates).toEqual([]);
+		expect(mocks.deletedAt).toBe(null);
+		expect(mocks.deleteObject).not.toHaveBeenCalled();
+	});
+
+	it("does not retire a video with no file when storage shows a file or an open upload once the row is locked", async () => {
+		mocks.unrecorded = [row("late")];
+		mocks.headObject
+			.mockImplementationOnce(missing)
+			.mockImplementation(() => Effect.succeed({ LastModified: hoursAgo(0) }));
+
+		expect(await removeUnrecordedInfraFiles(now)).toBe(0);
+
+		mocks.headObject.mockReset();
+		mocks.headObject.mockImplementation(missing);
+		mocks.listUploads
+			.mockImplementationOnce(() => Effect.succeed({ Uploads: [] }))
+			.mockImplementation(() =>
+				Effect.succeed({
+					Uploads: [
+						{
+							Key: "developer/app-1/late/video",
+							UploadId: "upload-2",
+							Initiated: now,
+						},
+					],
+				}),
+			);
+
+		expect(await removeUnrecordedInfraFiles(now)).toBe(0);
+		expect(mocks.videoUpdates).toEqual([]);
+		expect(mocks.deletedAt).toBe(null);
+	});
+
+	it("leaves a video with no file alone while an upload for it is still open", async () => {
+		mocks.unrecorded = [row("sending")];
+		mocks.headObject.mockImplementation(missing);
+		mocks.listUploads.mockImplementation(() =>
+			Effect.succeed({
+				Uploads: [
+					{
+						Key: "developer/app-1/sending/video",
+						UploadId: "upload-1",
+						Initiated: hoursAgo(1),
+					},
+				],
+			}),
+		);
+
+		expect(await removeUnrecordedInfraFiles(now)).toBe(0);
+		expect(mocks.videoUpdates).toEqual([]);
+	});
+
+	it("leaves a video with no file alone when storage cannot say what is there", async () => {
+		mocks.unrecorded = [row("unsure")];
+		mocks.headObject.mockImplementation(() =>
+			Effect.fail({ _tag: "S3Error", cause: { name: "SlowDown" } }),
+		);
+
+		expect(await removeUnrecordedInfraFiles(now)).toBe(0);
+
+		mocks.headObject.mockImplementation(missing);
+		mocks.listUploads.mockImplementation(() =>
+			Effect.fail({ _tag: "S3Error", cause: { name: "InternalError" } }),
+		);
+
+		expect(await removeUnrecordedInfraFiles(now)).toBe(0);
+
+		mocks.listUploads.mockImplementation(() =>
+			Effect.succeed({ Uploads: [], IsTruncated: true }),
+		);
+
+		expect(await removeUnrecordedInfraFiles(now)).toBe(0);
+		expect(mocks.videoUpdates).toEqual([]);
+	});
+
+	it("does not retire a video with no file when a completion recorded it first", async () => {
+		mocks.unrecorded = [row("raced")];
+		mocks.headObject.mockImplementation(() => {
+			mocks.storedDuration = 60;
+			return missing();
+		});
+
+		expect(await removeUnrecordedInfraFiles(now)).toBe(0);
+		expect(mocks.videoUpdates).toEqual([]);
+		expect(mocks.deletedAt).toBe(null);
+	});
+
 	it("counts only the files it removed", async () => {
 		mocks.unrecorded = [row("a"), row("b")];
 		mocks.headObject.mockImplementation(old);

apps/web/__tests__/unit/developer-transactions-api.test.ts

@@ -137,6 +137,7 @@ vi.mock("@/lib/infra-video-upload", () => ({
 
 import * as rest from "@/app/api/developer/v1/[...route]/route";
 import { developerOpenApiSpec } from "@/lib/developer-openapi";
+import { DEVELOPER_KEY_REQUESTS_PER_MINUTE } from "@/lib/developer-rate-limits";
 import { readAppLedger } from "@/lib/infra-ledger-store";
 
 let serial = 0;
@@ -363,9 +364,15 @@ describe("GET /transactions", () => {
 		const first = await get("/transactions", key);
 		const usageAfter = await get("/usage/nope", key);
 
-		expect(first.headers.get("X-RateLimit-Limit")).toBe("60");
-		expect(first.headers.get("X-RateLimit-Remaining")).toBe("59");
-		expect(usageAfter.headers.get("X-RateLimit-Remaining")).toBe("58");
+		expect(first.headers.get("X-RateLimit-Limit")).toBe(
+			String(DEVELOPER_KEY_REQUESTS_PER_MINUTE),
+		);
+		expect(first.headers.get("X-RateLimit-Remaining")).toBe(
+			String(DEVELOPER_KEY_REQUESTS_PER_MINUTE - 1),
+		);
+		expect(usageAfter.headers.get("X-RateLimit-Remaining")).toBe(
+			String(DEVELOPER_KEY_REQUESTS_PER_MINUTE - 2),
+		);
 	});
 
 	it("answers 405 for a method the path does not take", async () => {

apps/web/__tests__/unit/developer-upload-harness.ts

@@ -81,7 +81,7 @@ async function answer(query: string, params: unknown[]) {
 		}
 		const stored = storedKey(key, appId);
 		if (!stored || stored.revokedAt) return { rows: [] };
-		return { rows: [[appId, keyIdOf(key)]] };
+		return { rows: [[appId, keyIdOf(key), `owner-${appId}`]] };
 	}
 	if (query.includes("from `developer_videos` inner join `developer_apps`")) {
 		const [keyId, , videoId, appId] = params;

apps/web/__tests__/unit/developer-upload-token-flow.test.ts

@@ -42,6 +42,9 @@ vi.mock("@/lib/developer-transcript-store", () => ({
 
 const sdk = await import("@/app/api/developer/sdk/v1/[...route]/route");
 const rest = await import("@/app/api/developer/v1/[...route]/route");
+const { DEVELOPER_KEY_REQUESTS_PER_MINUTE: KEY_LIMIT } = await import(
+	"@/lib/developer-rate-limits"
+);
 const { createDeveloperUploadToken } = await import(
 	"@/lib/developer-upload-token"
 );
@@ -354,14 +357,16 @@ describe("creating an upload with the secret key", () => {
 		const send = () => post(rest, "/api/developer/v1/uploads", key, {});
 
 		const first = await send();
-		expect(first.headers.get("X-RateLimit-Limit")).toBe("60");
-		expect(first.headers.get("X-RateLimit-Remaining")).toBe("59");
-		for (let request = 2; request <= 60; request++) await send();
+		expect(first.headers.get("X-RateLimit-Limit")).toBe(String(KEY_LIMIT));
+		expect(first.headers.get("X-RateLimit-Remaining")).toBe(
+			String(KEY_LIMIT - 1),
+		);
+		for (let request = 2; request <= KEY_LIMIT; request++) await send();
 		const limited = await send();
 
 		expect(limited.status).toBe(429);
 		expect(await limited.json()).toMatchObject({ code: "rate_limited" });
-		expect(inserts()).toHaveLength(60);
+		expect(inserts()).toHaveLength(KEY_LIMIT);
 	});
 });
 

apps/web/__tests__/unit/developer-video-processing.test.ts

@@ -1245,6 +1245,16 @@ describe("developer video dispatch", () => {
 		expect(video().processing?.attempts.mp4).toBe(0);
 	});
 
+	it("records when the media server last answered busy", async () => {
+		mocks.startJob.mockResolvedValue({ status: "busy" });
+		const before = Date.now();
+
+		await dispatchDeveloperVideoPart("video-1", "encode-1", "mp4");
+
+		expect(video().processing?.busyAt).toBeGreaterThanOrEqual(before);
+		expect(video().processing?.busyAt).toBeLessThanOrEqual(Date.now());
+	});
+
 	it("keeps the part for a later try when the media server cannot be reached", async () => {
 		mocks.startJob.mockResolvedValue({ status: "unreachable" });
 

apps/web/__tests__/unit/developer-video-state.test.ts

@@ -532,6 +532,20 @@ describe("developer video redrive", () => {
 		expect(developerVideoRedrive(startedAt(7), now)).toBe("start");
 	});
 
+	it("times a video that waited for a free worker from its last busy answer", () => {
+		const waited = (busyHoursAgo: number) =>
+			developerVideo({
+				processing: processingState({
+					startedAt: now - 7 * 3_600_000,
+					busyAt: now - busyHoursAgo * 3_600_000,
+				}),
+			});
+
+		expect(developerVideoRedrive(waited(1), now)).toBeNull();
+		expect(developerVideoRedrive(waited(5.9), now)).toBeNull();
+		expect(developerVideoRedrive(waited(6), now)).toBe("start");
+	});
+
 	it("retries adaptive streaming only once its retry time has passed", () => {
 		const parked = (retryAt: number) =>
 			developerVideo({

apps/web/__tests__/unit/infra-agent-prompt.test.ts

@@ -6,6 +6,7 @@ import {
 	CAP_INFRA_MCP_URL,
 } from "@/data/infra-agent-prompt";
 import { developerOpenApiSpec } from "@/lib/developer-openapi";
+import { DEVELOPER_KEY_REQUESTS_PER_MINUTE } from "@/lib/developer-rate-limits";
 
 const METHODS = ["get", "post", "put", "patch", "delete"] as const;
 
@@ -62,7 +63,9 @@ describe("Cap Infra agent prompt", () => {
 		}
 		expect(CAP_INFRA_AGENT_PROMPT).toContain("every 2 seconds");
 		expect(CAP_INFRA_AGENT_PROMPT).toContain("Retry-After");
-		expect(CAP_INFRA_AGENT_PROMPT).toContain("60 requests a minute");
+		expect(CAP_INFRA_AGENT_PROMPT).toContain(
+			`${DEVELOPER_KEY_REQUESTS_PER_MINUTE} requests a minute`,
+		);
 	});
 
 	it("links only to the API reference and the hosted MCP server", () => {

apps/web/__tests__/unit/process-developer-video-workflow.test.ts

@@ -222,4 +222,60 @@ describe("process developer video workflow", () => {
 			],
 		]);
 	});
+
+	it("keeps a video waiting past three hours while every worker answers busy, then encodes it", async () => {
+		mocks.prepare.mockResolvedValue("ready");
+		mocks.dispatch.mockResolvedValue("busy");
+		const started = Date.now();
+		const busyFor = 5 * 60 * 60;
+		let waited = 0;
+		const now = vi
+			.spyOn(Date, "now")
+			.mockImplementation(() => started + waited * 1000);
+		mocks.sleep.mockImplementation(async (wait: string) => {
+			waited += Number.parseInt(wait, 10);
+		});
+		mocks.readParts.mockImplementation(async () =>
+			waited < busyFor
+				? { mp4: "dispatch", hls: "dispatch" }
+				: { mp4: "done", hls: "done" },
+		);
+
+		const result = await run();
+		now.mockRestore();
+
+		expect(result).toEqual({ success: true });
+		expect(waited).toBeGreaterThanOrEqual(busyFor);
+		expect(mocks.fail).not.toHaveBeenCalled();
+		expect(mocks.sweep).toHaveBeenCalledWith("video-1", "encode-1");
+	});
+
+	it("still stops at the three hour deadline when one output waits for room while the other has gone quiet", async () => {
+		mocks.prepare.mockResolvedValue("ready");
+		mocks.dispatch.mockResolvedValue("busy");
+		mocks.check.mockResolvedValue("unreachable");
+		mocks.readParts.mockResolvedValue({ mp4: "dispatch", hls: "check" });
+		const started = Date.now();
+		let waited = 0;
+		const now = vi
+			.spyOn(Date, "now")
+			.mockImplementation(() => started + waited * 1000);
+		mocks.sleep.mockImplementation(async (wait: string) => {
+			waited += Number.parseInt(wait, 10);
+		});
+
+		const result = await run();
+		now.mockRestore();
+
+		expect(result).toEqual({ success: true });
+		expect(waited).toBeGreaterThanOrEqual(3 * 60 * 60);
+		expect(waited).toBeLessThan(3 * 60 * 60 + 60);
+		expect(mocks.fail.mock.calls).toEqual([
+			[
+				"video-1",
+				"encode-1",
+				"Processing took too long and was stopped. Upload the video again.",
+			],
+		]);
+	});
 });