diff --git a/apps/web/__tests__/unit/developer-api-contract.test.ts b/apps/web/__tests__/unit/developer-api-contract.test.ts index 3a379e3e49..d6f3d43dbb 100644 --- a/apps/web/__tests__/unit/developer-api-contract.test.ts +++ b/apps/web/__tests__/unit/developer-api-contract.test.ts @@ -4,7 +4,13 @@ import { MySqlDialect } from "drizzle-orm/mysql-core"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; const mocks = vi.hoisted(() => ({ - keyRows: [] as { appId: string; environment?: string }[], + keyRows: [] as { + appId: string; + keyId?: string; + ownerId?: string; + environment?: string; + }[], + hashKey: vi.fn(), accounts: [] as { balanceMicroCredits: number }[], videoStats: [] as Record[], storedVideos: [] as Record[], @@ -63,9 +69,7 @@ vi.mock("@cap/env", () => ({ NEXTAUTH_SECRET: "unit-test-signing-secret", }), })); -vi.mock("@/lib/developer-key-hash", () => ({ - hashKey: async (key: string) => `hash:${key}`, -})); +vi.mock("@/lib/developer-key-hash", () => ({ hashKey: mocks.hashKey })); vi.mock("@/lib/developer-video-start", () => ({ redriveDeveloperVideoProcessing: vi.fn(), })); @@ -101,8 +105,10 @@ vi.mock("@/lib/infra-video-upload", () => ({ abortInfraUpload: vi.fn(), })); +import { Hono } from "hono"; import * as sdk from "@/app/api/developer/sdk/v1/[...route]/route"; import * as rest from "@/app/api/developer/v1/[...route]/route"; +import { withDeveloperSecretAuth } from "@/app/api/utils"; import { developerOpenApiSpec } from "@/lib/developer-openapi"; import { MICRO_CREDITS_PER_DOLLAR, @@ -112,6 +118,7 @@ import { } from "@/lib/developer-pricing"; import { DEVELOPER_KEY_REQUESTS_PER_MINUTE, + DEVELOPER_OWNER_REQUESTS_PER_MINUTE, DEVELOPER_UPLOAD_TOKEN_REQUESTS_PER_MINUTE, } from "@/lib/developer-rate-limits"; import { createDeveloperUploadToken } from "@/lib/developer-upload-token"; @@ -121,10 +128,15 @@ import { } from "@/lib/infra-pricing"; import { getAppStorage } from "@/lib/infra-storage"; +const KEY_LIMIT = DEVELOPER_KEY_REQUESTS_PER_MINUTE; +const OWNER_LIMIT = DEVELOPER_OWNER_REQUESTS_PER_MINUTE; +const keyLeft = (sent: number) => String(KEY_LIMIT - sent); + type Method = "GET" | "POST" | "PUT" | "PATCH" | "DELETE" | "OPTIONS"; const ORIGIN = "https://cap.so"; let keySerial = 0; +let ownerSerial = 0; const freshKey = (prefix: "csk" | "cpk" | "cut") => prefix === "cut" ? createDeveloperUploadToken({ @@ -170,7 +182,14 @@ const billingOf = async (billing: Record) => ({ beforeEach(() => { vi.clearAllMocks(); - mocks.keyRows = [{ appId: "app-1", environment: "development" }]; + mocks.hashKey.mockImplementation(async (key: string) => `hash:${key}`); + mocks.keyRows = [ + { + appId: "app-1", + ownerId: `owner-${++ownerSerial}`, + environment: "development", + }, + ]; mocks.accounts = [{ balanceMicroCredits: 4_532_000 }]; mocks.videoStats = [{ createdVideos: 5, uploadedVideos: 4 }]; mocks.storedVideos = [ @@ -220,22 +239,25 @@ describe("rate limit headers", () => { const first = await getUsage(key); expect(first.status).toBe(200); expect(rateLimit(first)).toEqual({ - limit: "60", - remaining: "59", + limit: String(KEY_LIMIT), + remaining: keyLeft(1), reset: "60", retryAfter: null, }); vi.advanceTimersByTime(20_500); const second = await getUsage(key); - expect(rateLimit(second)).toMatchObject({ remaining: "58", reset: "40" }); + expect(rateLimit(second)).toMatchObject({ + remaining: keyLeft(2), + reset: "40", + }); - for (let request = 3; request <= 60; request++) await getUsage(key); + for (let request = 3; request <= KEY_LIMIT; request++) await getUsage(key); vi.advanceTimersByTime(4_500); const limited = await getUsage(key); expect(limited.status).toBe(429); expect(rateLimit(limited)).toEqual({ - limit: "60", + limit: String(KEY_LIMIT), remaining: "0", reset: "35", retryAfter: "35", @@ -249,7 +271,7 @@ describe("rate limit headers", () => { const afterWait = await getUsage(key); expect(afterWait.status).toBe(200); expect(rateLimit(afterWait)).toMatchObject({ - remaining: "59", + remaining: keyLeft(1), reset: "60", }); }); @@ -258,7 +280,7 @@ describe("rate limit headers", () => { vi.useFakeTimers(); vi.setSystemTime(new Date("2026-11-12T10:00:00Z")); const key = freshKey("csk"); - for (let request = 1; request <= 60; request++) await getUsage(key); + for (let request = 1; request <= KEY_LIMIT; request++) await getUsage(key); for (const [elapsed, expected] of [ [0, "60"], @@ -286,17 +308,18 @@ describe("rate limit headers", () => { }); const other = await getUsage(freshKey("csk")); - expect(rateLimit(upper).remaining).toBe("59"); - expect(rateLimit(lower).remaining).toBe("58"); - expect(rateLimit(spaced).remaining).toBe("57"); - expect(rateLimit(other).remaining).toBe("59"); + expect(rateLimit(upper).remaining).toBe(keyLeft(1)); + expect(rateLimit(lower).remaining).toBe(keyLeft(2)); + expect(rateLimit(spaced).remaining).toBe(keyLeft(3)); + expect(rateLimit(other).remaining).toBe(keyLeft(1)); }); - it("carries the headers on errors too", async () => { + it("carries the headers on errors too, and spends none of a key's budget on a key that was not accepted", async () => { const key = freshKey("csk"); + const accepted = mocks.keyRows; mocks.keyRows = []; const unauthorized = await getUsage(key); - mocks.keyRows = [{ appId: "app-1" }]; + mocks.keyRows = accepted; const notFound = await call(rest, "GET", "/api/developer/v1/nope", { key }); const wrongMethod = await call(rest, "POST", "/api/developer/v1/usage", { key, @@ -311,10 +334,10 @@ describe("rate limit headers", () => { rateLimit(response).remaining, ]), ).toEqual([ - [401, "59"], - [404, "58"], - [405, "57"], - [500, "56"], + [401, null], + [404, keyLeft(1)], + [405, keyLeft(2)], + [500, keyLeft(3)], ]); }); @@ -378,7 +401,7 @@ describe("rate limit shared across servers", () => { expect(limited.status).toBe(429); expect(rateLimit(limited)).toEqual({ - limit: "60", + limit: String(KEY_LIMIT), remaining: "0", reset: "60", retryAfter: "60", @@ -386,7 +409,7 @@ describe("rate limit shared across servers", () => { expect(await limited.json()).toMatchObject({ error: "Rate limit exceeded", code: "rate_limited", - hint: "Wait for the Retry-After seconds, then retry. Each key allows 60 requests per minute.", + hint: `Wait for the Retry-After seconds, then retry. Each key allows ${KEY_LIMIT} requests per minute.`, }); expect(mocks.billing).not.toHaveBeenCalled(); }); @@ -418,13 +441,13 @@ describe("rate limit shared across servers", () => { it("lets the shared rule overrule a count this server kept alone", async () => { const key = freshKey("csk"); - for (let request = 1; request <= 60; request++) await getUsage(key); + for (let request = 1; request <= KEY_LIMIT; request++) await getUsage(key); const allowed = await getUsage(key); expect(allowed.status).toBe(200); expect(rateLimit(allowed)).toMatchObject({ - limit: "60", + limit: String(KEY_LIMIT), remaining: "0", retryAfter: null, }); @@ -439,7 +462,9 @@ describe("rate limit shared across servers", () => { await getUsage(freshKey("csk")); await initiate(token); - const [first, again, other, upload] = firewallCalls(); + const [first, again, other, upload] = firewallCalls().filter( + (asked) => asked.ruleId !== "rl_developer_api_owner", + ); expect([first, again, other, upload].map((asked) => asked?.ruleId)).toEqual( [ "rl_developer_api_key", @@ -475,12 +500,12 @@ describe("rate limit shared across servers", () => { const first = await getUsage(key); expect(rateLimit(first)).toEqual({ - limit: "60", - remaining: "59", + limit: String(KEY_LIMIT), + remaining: keyLeft(1), reset: "60", retryAfter: null, }); - for (let request = 2; request <= 60; request++) { + for (let request = 2; request <= KEY_LIMIT; request++) { expect((await getUsage(key)).status).toBe(200); } vi.advanceTimersByTime(25_000); @@ -488,7 +513,7 @@ describe("rate limit shared across servers", () => { expect(limited.status).toBe(429); expect(rateLimit(limited)).toEqual({ - limit: "60", + limit: String(KEY_LIMIT), remaining: "0", reset: "35", retryAfter: "35", @@ -496,6 +521,411 @@ describe("rate limit shared across servers", () => { }); }); +describe("one account's keys share a budget", () => { + const ownedBy = (ownerId: string) => { + mocks.keyRows = [{ appId: "app-1", ownerId }]; + }; + + it("allows the account budget across every key of an account, then answers 429", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-11-12T10:00:00Z")); + ownedBy("owner-shared"); + // Every key stays under its own limit, so only the account budget can + // refuse the request after them. + const perKey = Math.min(KEY_LIMIT - 1, OWNER_LIMIT); + const keys = Array.from({ length: Math.ceil(OWNER_LIMIT / perKey) }, () => + freshKey("csk"), + ); + expect(keys.length).toBeGreaterThan(1); + + for (const [index, key] of keys.entries()) { + const sends = Math.min(perKey, OWNER_LIMIT - index * perKey); + for (let request = 1; request <= sends; request++) { + expect((await getUsage(key)).status).toBe(200); + } + } + vi.advanceTimersByTime(20_000); + const limited = await getUsage(freshKey("csk")); + + expect(limited.status).toBe(429); + expect(rateLimit(limited)).toEqual({ + limit: String(KEY_LIMIT), + remaining: "0", + reset: "40", + retryAfter: "40", + }); + expect(await limited.json()).toMatchObject({ + code: "rate_limited", + hint: `Wait for the Retry-After seconds, then retry. Each account, across all of its keys, allows ${OWNER_LIMIT} requests per minute.`, + }); + expect(mocks.billing).toHaveBeenCalledTimes(OWNER_LIMIT); + + ownedBy("owner-other"); + expect((await getUsage(freshKey("csk"))).status).toBe(200); + + ownedBy("owner-shared"); + vi.advanceTimersByTime(40_001); + expect((await getUsage(keys[0] as string)).status).toBe(200); + }); + + it("takes the account from the key lookup, with no query of its own", async () => { + const response = await call(rest, "GET", "/api/developer/v1/nope", { + key: freshKey("csk"), + }); + + expect(response.status).toBe(404); + expect(mocks.selected).toHaveLength(1); + expect(Object.keys(mocks.selected[0] ?? {}).sort()).toEqual([ + "appId", + "keyId", + "ownerId", + ]); + expect(mocks.hashKey).toHaveBeenCalledTimes(1); + }); + + it("asks the shared rule about the account by its id", async () => { + vi.stubEnv("NODE_ENV", "production"); + ownedBy("owner-asked"); + + await getUsage(freshKey("csk")); + + const asked = mocks.firewall.mock.calls.map(([ruleId, options]) => [ + ruleId, + options.rateLimitKey, + ]); + expect(asked).toContainEqual(["rl_developer_api_owner", "owner-asked"]); + expect(asked).toHaveLength(2); + }); + + it("rejects when the shared rule for the account says so", async () => { + vi.stubEnv("NODE_ENV", "production"); + ownedBy("owner-limited"); + mocks.firewall.mockImplementation(async (ruleId: string) => ({ + rateLimited: ruleId === "rl_developer_api_owner", + })); + + const limited = await getUsage(freshKey("csk")); + + expect(limited.status).toBe(429); + expect(rateLimit(limited)).toMatchObject({ + limit: String(KEY_LIMIT), + remaining: "0", + retryAfter: "60", + }); + expect(mocks.billing).not.toHaveBeenCalled(); + }); +}); + +describe("secret key check behind other middleware", () => { + const guarded = (preset: Record) => + new Hono<{ Variables: Record }>() + .use(async (c, next) => { + for (const [name, value] of Object.entries(preset)) c.set(name, value); + await next(); + }) + .use(withDeveloperSecretAuth) + .get("/", (c) => + c.json({ + appId: c.get("developerAppId"), + keyId: c.get("developerKeyId") ?? null, + }), + ); + const send = ( + preset: Record, + headers: Record = {}, + ) => guarded(preset).request("https://cap.so/", { headers }); + + it("still asks for a secret key when an app was set by something other than a secret key", async () => { + const response = await send({ developerAppId: "app-from-token" }); + + expect(response.status).toBe(401); + expect(await response.json()).toMatchObject({ + code: "invalid_secret_key", + }); + expect(mocks.hashKey).not.toHaveBeenCalled(); + }); + + it("takes the app and key from the secret key, not from what was set before", async () => { + mocks.keyRows = [{ appId: "app-1", keyId: "key-1", ownerId: "owner-1" }]; + + const response = await send( + { developerAppId: "app-from-token" }, + { Authorization: `Bearer ${freshKey("csk")}` }, + ); + + expect(response.status).toBe(200); + expect(await response.json()).toEqual({ appId: "app-1", keyId: "key-1" }); + expect(mocks.hashKey).toHaveBeenCalledTimes(1); + }); + + it("does not look the key up again once the rate limiter accepted it", async () => { + const response = await send({ + developerAppId: "app-1", + developerKeyId: "key-1", + developerKeyType: "secret", + }); + + expect(response.status).toBe(200); + expect(await response.json()).toEqual({ appId: "app-1", keyId: "key-1" }); + expect(mocks.hashKey).not.toHaveBeenCalled(); + }); +}); + +describe("credentials that are not accepted", () => { + const from = (address: string) => ({ "x-forwarded-for": address }); + + it.each([ + ["no credential", undefined, 401, "invalid_secret_key"], + [ + "a credential with no prefix", + "Bearer not-a-key", + 401, + "invalid_secret_key", + ], + ["a public key", "Bearer cpk_public_1", 403, "public_key_not_allowed"], + [ + "an oversized key", + `Bearer csk_${"a".repeat(200)}`, + 401, + "invalid_secret_key", + ], + [ + "a key with characters no key has", + "Bearer csk_abc!def", + 401, + "invalid_secret_key", + ], + ] as const)( + "refuses %s without hashing, looking up or counting anything", + async (_label, authorization, status, code) => { + vi.stubEnv("NODE_ENV", "production"); + + const response = await call(rest, "GET", "/api/developer/v1/usage", { + headers: authorization ? { Authorization: authorization } : {}, + }); + + expect(response.status).toBe(status); + expect(await response.json()).toMatchObject({ code }); + expect(rateLimit(response)).toEqual({ + limit: null, + remaining: null, + reset: null, + retryAfter: null, + }); + expect(mocks.hashKey).not.toHaveBeenCalled(); + expect(mocks.selected).toHaveLength(0); + expect(mocks.firewall).not.toHaveBeenCalled(); + }, + ); + + it("answers an unknown key 401 on any path, and stops looking keys up for an address after 30 of them in a minute", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-11-12T10:00:00Z")); + mocks.keyRows = []; + const address = from("203.0.113.9"); + + for (let attempt = 1; attempt <= 30; attempt++) { + const refused = await call(rest, "GET", "/api/developer/v1/nope", { + key: freshKey("csk"), + headers: address, + }); + expect(refused.status).toBe(401); + } + expect(mocks.hashKey).toHaveBeenCalledTimes(30); + + const limited = await getUsage(freshKey("csk"), address); + expect(limited.status).toBe(429); + expect(limited.headers.get("Retry-After")).toBe("60"); + expect(await limited.json()).toMatchObject({ + code: "rate_limited", + error: "Too many requests with a key or token that was not accepted", + }); + expect(mocks.hashKey).toHaveBeenCalledTimes(31); + + mocks.hashKey.mockClear(); + mocks.selected = []; + for (let attempt = 1; attempt <= 50; attempt++) { + expect((await getUsage(freshKey("csk"), address)).status).toBe(429); + } + expect(mocks.hashKey).not.toHaveBeenCalled(); + expect(mocks.selected).toHaveLength(0); + + const elsewhere = await getUsage(freshKey("csk"), from("198.51.100.4")); + expect(elsewhere.status).toBe(401); + + vi.advanceTimersByTime(60_001); + expect((await getUsage(freshKey("csk"), address)).status).toBe(401); + }); + + it("keeps a key it accepted in the last minute working from an address that is being refused", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-11-12T11:00:00Z")); + const address = from("203.0.113.77"); + const accepted = mocks.keyRows; + const customer = freshKey("csk"); + expect((await getUsage(customer, address)).status).toBe(200); + + mocks.keyRows = []; + for (let attempt = 1; attempt <= 31; attempt++) { + await getUsage(freshKey("csk"), address); + } + expect((await getUsage(freshKey("csk"), address)).status).toBe(429); + + mocks.keyRows = accepted; + expect((await getUsage(customer, address)).status).toBe(200); + }); + + it("counts that minute from the key's last accepted request, not from when its count began", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-11-12T11:30:00Z")); + const address = from("203.0.113.78"); + const accepted = mocks.keyRows; + const customer = freshKey("csk"); + expect((await getUsage(customer, address)).status).toBe(200); + vi.advanceTimersByTime(59_000); + expect((await getUsage(customer, address)).status).toBe(200); + + mocks.keyRows = []; + for (let attempt = 1; attempt <= 31; attempt++) { + await getUsage(freshKey("csk"), address); + } + expect((await getUsage(freshKey("csk"), address)).status).toBe(429); + + mocks.keyRows = accepted; + vi.advanceTimersByTime(2_000); + expect((await getUsage(customer, address)).status).toBe(200); + + vi.advanceTimersByTime(61_000); + mocks.keyRows = []; + for (let attempt = 1; attempt <= 31; attempt++) { + await getUsage(freshKey("csk"), address); + } + mocks.keyRows = accepted; + expect((await getUsage(customer, address)).status).toBe(429); + }); + + it("counts by the address the platform reports, not one the caller writes", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-11-12T12:00:00Z")); + vi.stubEnv("VERCEL", "1"); + mocks.keyRows = []; + const statuses: number[] = []; + + for (let attempt = 1; attempt <= 32; attempt++) { + const response = await getUsage(freshKey("csk"), { + "x-real-ip": "203.0.113.200", + "x-forwarded-for": `10.0.0.${attempt}, 203.0.113.200`, + "x-client-ip": `10.9.9.${attempt}`, + }); + statuses.push(response.status); + } + + expect(statuses.slice(0, 30)).toEqual(Array(30).fill(401)); + expect(statuses.slice(30)).toEqual([429, 429]); + }); + + it("asks the shared rule about the address, sending it no credential", async () => { + vi.stubEnv("NODE_ENV", "production"); + mocks.keyRows = []; + const key = freshKey("csk"); + + const refused = await getUsage(key, { "x-real-ip": "203.0.113.50" }); + + expect(refused.status).toBe(401); + expect(mocks.firewall).toHaveBeenCalledTimes(1); + const [ruleId, options] = mocks.firewall.mock.calls[0] ?? []; + expect(ruleId).toBe("rl_developer_auth_failure"); + expect(options.rateLimitKey).toBeUndefined(); + expect(options.request.headers.get("x-real-ip")).toBe("203.0.113.50"); + expect(JSON.stringify([...options.request.headers])).not.toContain(key); + }); + + it("answers 429 as soon as the shared rule says the address has sent too many", async () => { + vi.stubEnv("NODE_ENV", "production"); + mocks.keyRows = []; + mocks.firewall.mockResolvedValue({ rateLimited: true }); + const address = { "x-real-ip": "203.0.113.60" }; + + expect((await getUsage(freshKey("csk"), address)).status).toBe(429); + mocks.hashKey.mockClear(); + mocks.firewall.mockClear(); + expect((await getUsage(freshKey("csk"), address)).status).toBe(429); + expect(mocks.hashKey).not.toHaveBeenCalled(); + expect(mocks.firewall).not.toHaveBeenCalled(); + }); + + it("does not count an upload token that is malformed, expired or simply absent", async () => { + vi.stubEnv("NODE_ENV", "production"); + const send = (authorization?: string) => + call(sdk, "POST", "/api/developer/sdk/v1/upload/multipart/initiate", { + headers: { + "Content-Type": "application/json", + ...(authorization ? { Authorization: authorization } : {}), + }, + body: "{}", + }); + const expired = createDeveloperUploadToken({ + appId: "app-1", + videoId: "video-expired", + expiresAt: Date.now() - 1, + corsOrigin: null, + }); + + const answers = [ + await send(), + await send("Bearer csk_secret_on_the_wrong_route"), + await send("Bearer cpk_public_1"), + await send(`Bearer ${expired}`), + ]; + + expect(answers.map((response) => response.status)).toEqual([ + 401, 401, 403, 401, + ]); + expect(mocks.firewall).not.toHaveBeenCalled(); + for (const response of answers) { + expect(rateLimit(response).limit).toBeNull(); + } + }); + + it("counts a forged upload token against the address and still words the refusal as before", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-11-12T13:00:00Z")); + const forged = () => + call(sdk, "POST", "/api/developer/sdk/v1/upload/multipart/initiate", { + key: `${freshKey("cut")}x`, + headers: { + "Content-Type": "application/json", + "x-forwarded-for": "203.0.113.120", + }, + body: "{}", + }); + + const first = await forged(); + expect(first.status).toBe(401); + expect(await first.json()).toMatchObject({ code: "invalid_upload_token" }); + for (let attempt = 2; attempt <= 30; attempt++) await forged(); + + const limited = await forged(); + expect(limited.status).toBe(429); + expect(limited.headers.get("Access-Control-Allow-Origin")).toBe("*"); + + const genuine = await call( + sdk, + "POST", + "/api/developer/sdk/v1/upload/multipart/initiate", + { + key: freshKey("cut"), + headers: { + "Content-Type": "application/json", + "x-forwarded-for": "203.0.113.120", + }, + body: "{}", + }, + ); + expect(genuine.status).toBe(200); + }); +}); + describe("upload tokens have their own budget", () => { const initiate = (key: string) => call(sdk, "POST", "/api/developer/sdk/v1/upload/multipart/initiate", { @@ -531,8 +961,8 @@ describe("upload tokens have their own budget", () => { expect(rateLimit(await initiate(freshKey("cut"))).remaining).toBe("1199"); }); - it("keeps the secret key at 60 a minute", async () => { - expect(rateLimit(await getUsage(freshKey("csk"))).limit).toBe("60"); + it("keeps the secret key at 300 a minute", async () => { + expect(rateLimit(await getUsage(freshKey("csk"))).limit).toBe("300"); }); it("documents the limits the limiter applies", async () => { @@ -556,6 +986,12 @@ describe("upload tokens have their own budget", () => { expect(spec.info.description).toContain( `each upload token to ${token} requests per minute`, ); + expect(docs).toContain( + `All the keys of one account together allow ${DEVELOPER_OWNER_REQUESTS_PER_MINUTE.toLocaleString("en-US")} requests per minute`, + ); + expect(spec.info.description).toContain( + `all the keys of one account together to ${DEVELOPER_OWNER_REQUESTS_PER_MINUTE}`, + ); }); }); @@ -588,7 +1024,7 @@ describe("browser access", () => { await from("POST", "/api/developer/v1/usage"), ]; const sentWithKey = answers.length - 1; - for (let request = sentWithKey; request < 60; request++) { + for (let request = sentWithKey; request < KEY_LIMIT; request++) { await from("GET", "/api/developer/v1/usage"); } answers.push(await from("GET", "/api/developer/v1/usage")); diff --git a/apps/web/__tests__/unit/developer-sdk-upload-billing.test.ts b/apps/web/__tests__/unit/developer-sdk-upload-billing.test.ts index 9243e84410..9fe21d0738 100644 --- a/apps/web/__tests__/unit/developer-sdk-upload-billing.test.ts +++ b/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[], videoUpdates: [] as Record[], 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); diff --git a/apps/web/__tests__/unit/developer-transactions-api.test.ts b/apps/web/__tests__/unit/developer-transactions-api.test.ts index aa8e2591f7..95ef9c4d0e 100644 --- a/apps/web/__tests__/unit/developer-transactions-api.test.ts +++ b/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 () => { diff --git a/apps/web/__tests__/unit/developer-upload-harness.ts b/apps/web/__tests__/unit/developer-upload-harness.ts index 0c20200196..ad5f4e3f76 100644 --- a/apps/web/__tests__/unit/developer-upload-harness.ts +++ b/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; diff --git a/apps/web/__tests__/unit/developer-upload-token-flow.test.ts b/apps/web/__tests__/unit/developer-upload-token-flow.test.ts index f725addd9a..e0a0094e8c 100644 --- a/apps/web/__tests__/unit/developer-upload-token-flow.test.ts +++ b/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); }); }); diff --git a/apps/web/__tests__/unit/developer-video-processing.test.ts b/apps/web/__tests__/unit/developer-video-processing.test.ts index 9a0dad60c8..bc5ab0b334 100644 --- a/apps/web/__tests__/unit/developer-video-processing.test.ts +++ b/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" }); diff --git a/apps/web/__tests__/unit/developer-video-state.test.ts b/apps/web/__tests__/unit/developer-video-state.test.ts index 31c48ab860..4a4341b99f 100644 --- a/apps/web/__tests__/unit/developer-video-state.test.ts +++ b/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({ diff --git a/apps/web/__tests__/unit/infra-agent-prompt.test.ts b/apps/web/__tests__/unit/infra-agent-prompt.test.ts index 350d4c5618..bf20bd1c6c 100644 --- a/apps/web/__tests__/unit/infra-agent-prompt.test.ts +++ b/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", () => { diff --git a/apps/web/__tests__/unit/process-developer-video-workflow.test.ts b/apps/web/__tests__/unit/process-developer-video-workflow.test.ts index 9d684cc34e..b7e5638693 100644 --- a/apps/web/__tests__/unit/process-developer-video-workflow.test.ts +++ b/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.", + ], + ]); + }); }); diff --git a/apps/web/app/api/utils.ts b/apps/web/app/api/utils.ts index 401853552e..df7b4d3088 100644 --- a/apps/web/app/api/utils.ts +++ b/apps/web/app/api/utils.ts @@ -11,7 +11,7 @@ import { } from "@cap/database/schema"; import { buildEnv } from "@cap/env"; import { and, eq, gt, isNull } from "drizzle-orm"; -import type { Context, Next } from "hono"; +import type { Context } from "hono"; import { cors } from "hono/cors"; import { createMiddleware } from "hono/factory"; import { HTTPException } from "hono/http-exception"; @@ -24,6 +24,7 @@ import { import { hashKey } from "@/lib/developer-key-hash"; import { DEVELOPER_KEY_REQUESTS_PER_MINUTE, + DEVELOPER_OWNER_REQUESTS_PER_MINUTE, DEVELOPER_RATE_LIMIT_WINDOW_SECONDS, DEVELOPER_UPLOAD_TOKEN_REQUESTS_PER_MINUTE, } from "@/lib/developer-rate-limits"; @@ -37,12 +38,23 @@ import { uploadTokenRefusal, verifyDeveloperUploadToken, } from "@/lib/developer-upload-token"; -import { RATE_LIMIT_IDS, sharedRateLimitDecision } from "@/lib/rate-limit"; +import { + isRateLimited, + RATE_LIMIT_IDS, + rateLimitClientAddress, + sharedRateLimitDecision, +} from "@/lib/rate-limit"; const RATE_LIMIT_WINDOW_MS = DEVELOPER_RATE_LIMIT_WINDOW_SECONDS * 1000; const RATE_LIMIT_MAX_ENTRIES = 10_000; type DeveloperRateLimit = { ruleId: string; limit: number; subject: string }; +type DeveloperWindow = { + limited: boolean; + shared: boolean; + count: number; + resetAt: number; +}; const requestCounts = new Map< string, @@ -58,46 +70,44 @@ async function credentialDigest(credential: string) { return Buffer.from(digest).toString("hex"); } -// The firewall rule counts a credential across every server instance and its +function sweepRequestCounts(now: number) { + rateLimitRequestCounter++; + if (rateLimitRequestCounter % 100 !== 0) return; + for (const [k, v] of requestCounts) { + if (now > v.resetAt) requestCounts.delete(k); + } + if (requestCounts.size <= RATE_LIMIT_MAX_ENTRIES) return; + for (const [k, v] of requestCounts) { + if (requestCounts.size <= RATE_LIMIT_MAX_ENTRIES) break; + if (v.count < v.limit) requestCounts.delete(k); + } + if (requestCounts.size <= RATE_LIMIT_MAX_ENTRIES) return; + const byExpiry = [...requestCounts.entries()].sort( + (a, b) => a[1].resetAt - b[1].resetAt, + ); + for (const [k] of byExpiry) { + if (requestCounts.size <= RATE_LIMIT_MAX_ENTRIES) break; + requestCounts.delete(k); + } +} + +// The firewall rule counts a bucket across every server instance and its // answer is final. The window kept here fills the headers, and decides only // when the firewall cannot answer. A 429 from the rule waits a whole window, -// because this instance cannot see when the shared one began. -async function limitDeveloperRequest( +// because this instance cannot see when the shared one began. Only a +// credential that has already been accepted is counted, so a caller without +// one cannot fill this map or spend a firewall call. +async function countDeveloperRequest( c: Context, - next: Next, - { ruleId, limit, subject }: DeveloperRateLimit, -) { - const credential = - c.req.header("authorization")?.trim().split(/\s+/)[1] ?? - c.req.header("x-forwarded-for") ?? - "unknown"; - const key = `${ruleId}|${credential}`; + { ruleId, limit }: DeveloperRateLimit, + bucket: string, + sharedKey: string, +): Promise { + const key = `${ruleId}|${bucket}`; const now = Date.now(); - - rateLimitRequestCounter++; - if (rateLimitRequestCounter % 100 === 0) { - for (const [k, v] of requestCounts) { - if (now > v.resetAt) requestCounts.delete(k); - } - if (requestCounts.size > RATE_LIMIT_MAX_ENTRIES) { - for (const [k, v] of requestCounts) { - if (requestCounts.size <= RATE_LIMIT_MAX_ENTRIES) break; - if (v.count < v.limit) requestCounts.delete(k); - } - if (requestCounts.size > RATE_LIMIT_MAX_ENTRIES) { - const byExpiry = [...requestCounts.entries()].sort( - (a, b) => a[1].resetAt - b[1].resetAt, - ); - for (const [k] of byExpiry) { - if (requestCounts.size <= RATE_LIMIT_MAX_ENTRIES) break; - requestCounts.delete(k); - } - } - } - } + sweepRequestCounts(now); let entry = requestCounts.get(key); - if (!entry || now > entry.resetAt) { entry = { count: 1, resetAt: now + RATE_LIMIT_WINDOW_MS, limit }; requestCounts.set(key, entry); @@ -107,51 +117,270 @@ async function limitDeveloperRequest( const { count, resetAt } = entry; const shared = await sharedRateLimitDecision(ruleId, { - key: await credentialDigest(credential), + key: sharedKey, headers: c.req.raw.headers, }); - const limited = shared ?? count > limit; + return { + limited: shared ?? count > limit, + shared: shared !== undefined, + count, + resetAt, + }; +} - const resetSeconds = String( - shared +function resetSeconds(window: DeveloperWindow): string { + return String( + window.shared && window.limited ? DEVELOPER_RATE_LIMIT_WINDOW_SECONDS - : Math.max(1, Math.ceil((resetAt - now) / 1000)), + : Math.max(1, Math.ceil((window.resetAt - Date.now()) / 1000)), ); +} + +function setRateLimitHeaders( + c: Context, + { limit }: DeveloperRateLimit, + window: DeveloperWindow, +) { c.header("X-RateLimit-Limit", String(limit)); c.header( "X-RateLimit-Remaining", - String(limited ? 0 : Math.max(0, limit - count)), + String(window.limited ? 0 : Math.max(0, limit - window.count)), ); - c.header("X-RateLimit-Reset", resetSeconds); - - if (limited) { - c.header("Retry-After", resetSeconds); - return c.json( - apiErrorBody( - "rate_limited", - "Rate limit exceeded", - `Wait for the Retry-After seconds, then retry. Each ${subject} allows ${limit} requests per minute.`, - ), - 429, - ); - } - - await next(); + c.header("X-RateLimit-Reset", resetSeconds(window)); } -const developerRateLimit = (rule: DeveloperRateLimit) => - createMiddleware((c, next) => limitDeveloperRequest(c, next, rule)); +function rateLimited( + c: Context, + { limit, subject }: DeveloperRateLimit, + window: DeveloperWindow, +) { + c.header("Retry-After", resetSeconds(window)); + return c.json( + apiErrorBody( + "rate_limited", + "Rate limit exceeded", + `Wait for the Retry-After seconds, then retry. Each ${subject} allows ${limit} requests per minute.`, + ), + 429, + ); +} -export const developerRateLimiter = developerRateLimit({ +const KEY_RATE_LIMIT: DeveloperRateLimit = { ruleId: RATE_LIMIT_IDS.DEVELOPER_API_KEY, limit: DEVELOPER_KEY_REQUESTS_PER_MINUTE, subject: "key", -}); - -export const developerUploadRateLimiter = developerRateLimit({ +}; +const OWNER_RATE_LIMIT: DeveloperRateLimit = { + ruleId: RATE_LIMIT_IDS.DEVELOPER_API_OWNER, + limit: DEVELOPER_OWNER_REQUESTS_PER_MINUTE, + subject: "account, across all of its keys,", +}; +const UPLOAD_TOKEN_RATE_LIMIT: DeveloperRateLimit = { ruleId: RATE_LIMIT_IDS.DEVELOPER_UPLOAD_TOKEN, limit: DEVELOPER_UPLOAD_TOKEN_REQUESTS_PER_MINUTE, subject: "upload token", +}; + +// Once an address has sent too many credentials that were not accepted, this +// instance stops looking up what it sends until the window is over. A key it +// accepted within the last minute is still let through, so a customer who +// shares an address with such a caller keeps working. +const blockedAddresses = new Map(); +const acceptedAt = new Map(); + +function recentlyAccepted(credential: string): boolean { + const at = acceptedAt.get(credential); + return at !== undefined && Date.now() - at < RATE_LIMIT_WINDOW_MS; +} + +function rememberAccepted(credential: string) { + const now = Date.now(); + if ( + !acceptedAt.has(credential) && + acceptedAt.size >= RATE_LIMIT_MAX_ENTRIES + ) { + for (const [key, at] of acceptedAt) { + if (now - at >= RATE_LIMIT_WINDOW_MS) acceptedAt.delete(key); + } + if (acceptedAt.size >= RATE_LIMIT_MAX_ENTRIES) acceptedAt.clear(); + } + acceptedAt.set(credential, now); +} + +function authFailuresBlocked(c: Context): boolean { + const address = rateLimitClientAddress(c.req.raw.headers); + if (!address) return false; + const until = blockedAddresses.get(address); + if (until === undefined) return false; + if (until > Date.now()) return true; + blockedAddresses.delete(address); + return false; +} + +async function tooManyAuthFailures(c: Context): Promise { + const limited = await isRateLimited(RATE_LIMIT_IDS.DEVELOPER_AUTH_FAILURE, { + headers: c.req.raw.headers, + }); + const address = limited + ? rateLimitClientAddress(c.req.raw.headers) + : undefined; + if (address) { + const now = Date.now(); + if (blockedAddresses.size >= RATE_LIMIT_MAX_ENTRIES) { + for (const [blocked, until] of blockedAddresses) { + if (until <= now) blockedAddresses.delete(blocked); + } + if (blockedAddresses.size >= RATE_LIMIT_MAX_ENTRIES) { + blockedAddresses.clear(); + } + } + blockedAddresses.set(address, now + RATE_LIMIT_WINDOW_MS); + } + return limited; +} + +function authFailuresLimited(c: Context) { + c.header("Retry-After", String(DEVELOPER_RATE_LIMIT_WINDOW_SECONDS)); + return c.json( + apiErrorBody( + "rate_limited", + "Too many requests with a key or token that was not accepted", + "Wait for the Retry-After seconds, then retry with a valid key or upload token.", + ), + 429, + ); +} + +const bearerCredential = (c: Context) => + c.req.header("authorization")?.trim().split(/\s+/)[1]; + +// Every key issued is csk_ and 30 characters from a 32 letter alphabet. Any +// credential outside this shape is refused without being hashed or looked up. +const SECRET_KEY_SHAPE = /^csk_[A-Za-z0-9_-]{1,124}$/; + +type DeveloperSecretKey = { appId: string; keyId: string; ownerId: string }; + +async function findDeveloperSecretKey( + credential: string, +): Promise { + const keyHash = await hashKey(credential); + const [row] = await db() + .select({ + appId: developerApps.id, + keyId: developerApiKeys.id, + ownerId: developerApps.ownerId, + }) + .from(developerApiKeys) + .innerJoin( + developerApps, + and( + eq(developerApiKeys.appId, developerApps.id), + isNull(developerApps.deletedAt), + ), + ) + .where( + and( + eq(developerApiKeys.keyHash, keyHash), + eq(developerApiKeys.keyType, "secret"), + isNull(developerApiKeys.revokedAt), + ), + ) + .limit(1); + if (!row) return null; + debouncedLastUsedUpdate(keyHash); + return row; +} + +const unknownSecretKey = () => + apiErrorBody( + "invalid_secret_key", + "Invalid or revoked secret key", + "Copy the current secret key for your app from https://cap.so/dashboard/infra. Regenerating keys revokes the previous pair.", + ); + +type DeveloperSecretVariables = { + developerAppId: string; + developerKeyId: string; + developerOwnerId: string; + developerKeyType: "secret"; +}; + +// Every path that accepts a secret key sets the same variables, so a route +// gets the key's id whichever middleware looked it up. +function setDeveloperSecretKey( + c: Context<{ Variables: DeveloperSecretVariables }>, + key: DeveloperSecretKey, +) { + c.set("developerAppId", key.appId); + c.set("developerKeyId", key.keyId); + c.set("developerOwnerId", key.ownerId); + c.set("developerKeyType", "secret"); +} + +// A request without a well formed key costs nothing here and is answered by +// the route. A well formed key is looked up once: an unknown one counts +// against the address that sent it, a known one against its own budget and +// its owner's. +export const developerRateLimiter = createMiddleware<{ + Variables: DeveloperSecretVariables; +}>(async (c, next) => { + const credential = bearerCredential(c); + if (!credential || !SECRET_KEY_SHAPE.test(credential)) return next(); + if (authFailuresBlocked(c) && !recentlyAccepted(credential)) { + return authFailuresLimited(c); + } + + const key = await findDeveloperSecretKey(credential); + if (!key) { + if (await tooManyAuthFailures(c)) return authFailuresLimited(c); + return c.json(unknownSecretKey(), 401); + } + setDeveloperSecretKey(c, key); + rememberAccepted(credential); + + const [byKey, byOwner] = await Promise.all([ + credentialDigest(credential).then((digest) => + countDeveloperRequest(c, KEY_RATE_LIMIT, credential, digest), + ), + countDeveloperRequest(c, OWNER_RATE_LIMIT, key.ownerId, key.ownerId), + ]); + setRateLimitHeaders(c, KEY_RATE_LIMIT, byKey); + if (byKey.limited) return rateLimited(c, KEY_RATE_LIMIT, byKey); + if (byOwner.limited) { + c.header("X-RateLimit-Remaining", "0"); + c.header("X-RateLimit-Reset", resetSeconds(byOwner)); + return rateLimited(c, OWNER_RATE_LIMIT, byOwner); + } + await next(); +}); + +// A token is checked by its signature alone, so this costs no lookup. One +// that was never issued counts against the address that sent it, and the +// route still gives the refusal its usual wording. +export const developerUploadRateLimiter = createMiddleware(async (c, next) => { + const credential = bearerCredential(c); + if (!credential?.startsWith(UPLOAD_TOKEN_PREFIX)) return next(); + + const check = verifyDeveloperUploadToken(credential); + if (!check.ok) { + if ( + check.reason === "invalid" && + (authFailuresBlocked(c) || (await tooManyAuthFailures(c))) + ) { + return authFailuresLimited(c); + } + return next(); + } + + const window = await countDeveloperRequest( + c, + UPLOAD_TOKEN_RATE_LIMIT, + credential, + await credentialDigest(credential), + ); + setRateLimitHeaders(c, UPLOAD_TOKEN_RATE_LIMIT, window); + if (window.limited) return rateLimited(c, UPLOAD_TOKEN_RATE_LIMIT, window); + await next(); }); const LAST_USED_DEBOUNCE_MS = 5 * 60 * 1000; @@ -253,9 +482,6 @@ export const developerSdkCors = cors({ ], }); -const bearerCredential = (c: Context) => - c.req.header("authorization")?.trim().split(/\s+/)[1]; - export const refuseDeveloperPublicKey = createMiddleware(async (c, next) => { if (isDeveloperPublicKey(bearerCredential(c))) { return c.json(publicKeyRefusal(), 403); @@ -360,15 +586,15 @@ export const withDeveloperUploadToken = createMiddleware<{ }); export const withDeveloperSecretAuth = createMiddleware<{ - Variables: { - developerAppId: string; - developerKeyId: string; - developerKeyType: "secret"; - }; + Variables: DeveloperSecretVariables; }>(async (c, next) => { - const authorization = c.req.header("authorization"); - const authHeader = authorization?.split(" ")[1]; - if (!authorization?.trim()) { + // Only a secret key the rate limiter already accepted skips the lookup. An + // app id set by any other middleware, such as an upload token's, does not. + if (c.get("developerAppId") && c.get("developerKeyType") === "secret") { + return next(); + } + + if (!c.req.header("authorization")?.trim()) { return c.json( apiErrorBody( "invalid_secret_key", @@ -378,7 +604,8 @@ export const withDeveloperSecretAuth = createMiddleware<{ 401, ); } - if (isDeveloperPublicKey(bearerCredential(c))) { + const credential = bearerCredential(c); + if (isDeveloperPublicKey(credential)) { return c.json( apiErrorBody( "public_key_not_allowed", @@ -388,7 +615,7 @@ export const withDeveloperSecretAuth = createMiddleware<{ 403, ); } - if (!authHeader?.startsWith("csk_")) { + if (!credential || !SECRET_KEY_SHAPE.test(credential)) { return c.json( apiErrorBody( "invalid_secret_key", @@ -399,42 +626,10 @@ export const withDeveloperSecretAuth = createMiddleware<{ ); } - const keyHash = await hashKey(authHeader); - const [row] = await db() - .select({ appId: developerApps.id, keyId: developerApiKeys.id }) - .from(developerApiKeys) - .innerJoin( - developerApps, - and( - eq(developerApiKeys.appId, developerApps.id), - isNull(developerApps.deletedAt), - ), - ) - .where( - and( - eq(developerApiKeys.keyHash, keyHash), - eq(developerApiKeys.keyType, "secret"), - isNull(developerApiKeys.revokedAt), - ), - ) - .limit(1); - - if (!row) { - return c.json( - apiErrorBody( - "invalid_secret_key", - "Invalid or revoked secret key", - "Copy the current secret key for your app from https://cap.so/dashboard/infra. Regenerating keys revokes the previous pair.", - ), - 401, - ); - } - - debouncedLastUsedUpdate(keyHash); + const key = await findDeveloperSecretKey(credential); + if (!key) return c.json(unknownSecretKey(), 401); - c.set("developerAppId", row.appId); - c.set("developerKeyId", row.keyId); - c.set("developerKeyType", "secret" as const); + setDeveloperSecretKey(c, key); await next(); }); diff --git a/apps/web/content/docs/api/get-started.mdx b/apps/web/content/docs/api/get-started.mdx index 1c3e5fcfc2..bbefad8fa1 100644 --- a/apps/web/content/docs/api/get-started.mdx +++ b/apps/web/content/docs/api/get-started.mdx @@ -351,7 +351,7 @@ Add this to `index.html`, below the upload script: - **`ready`** means the video can be played. Processing has made at least one of `hlsUrl` and `mp4Url`, and the poster. - **`hlsUrl` can still be `null` for a moment after `ready`.** The MP4 is often finished first and the adaptive HLS renditions follow a few seconds later. The loop above keeps checking a few more times for `hlsUrl`, then plays the MP4. - **`errored`** is final. `errorReason` says why, for example that the file could not be read as a video. It is written for you and can mention your credit balance, so log it and show your users your own message. A failed video is not billed and cannot be processed again: create a new upload. -- **Pace the checks.** A secret key allows 60 requests per minute, and every status check your server makes for any user counts toward it. Leave at least 3 seconds between checks and slow down from there. On a `429`, wait for the `Retry-After` seconds and ask again. See [Rate Limits](/docs/api/rest-api#rate-limits). +- **Pace the checks.** A secret key allows 300 requests per minute, and every status check your server makes for any user counts toward it. Leave at least 3 seconds between checks and slow down from there. On a `429`, wait for the `Retry-After` seconds and ask again. See [Rate Limits](/docs/api/rest-api#rate-limits). ## 5. Play it diff --git a/apps/web/content/docs/api/rest-api.mdx b/apps/web/content/docs/api/rest-api.mdx index c2e81b3699..845b97bc59 100644 --- a/apps/web/content/docs/api/rest-api.mdx +++ b/apps/web/content/docs/api/rest-api.mdx @@ -130,9 +130,9 @@ A `4xx` response will not succeed if the same request is sent again, with two ex ## Rate Limits -A secret key allows 60 requests per minute, counted across every REST endpoint. An upload token allows 1,200 requests per minute, counted across the SDK endpoints. Playback URLs need no key and are not counted. +A secret key allows 300 requests per minute, counted across every REST endpoint. All the keys of one account together allow 1,200 requests per minute, so an account with several apps shares that total between them. An upload token allows 1,200 requests per minute, counted across the SDK endpoints. Playback URLs need no key and are not counted. -Every response carries three headers: +Every response to an accepted key or upload token carries three headers: | Header | Meaning | |--------|---------| @@ -144,6 +144,8 @@ Past the limit the API answers `429` with the code `rate_limited` and a `Retry-A `X-RateLimit-Remaining` and `X-RateLimit-Reset` are a guide. Requests can be answered by more than one server and each reports what it has counted itself, so `X-RateLimit-Remaining` can read higher than what is left and can differ from one response to the next. Use it to slow down early, not to decide that the next request will succeed. +A request whose key or upload token is not accepted is answered `401` and spends none of a key's budget. Too many of those from one address in a minute are answered `429` until the minute is over. + Each upload token has its own count, separate from your secret key's and from every other upload. Each uploaded part spends one presign request from it, so one upload can sign twenty parts a second. ## Pagination @@ -564,7 +566,7 @@ The `status` field moves from `uploading` to `processing` when the upload comple A failed video is not billed. It stays in the list with its `errorReason`, and its uploaded file is removed 24 hours after processing started, after which `storedBytes` is `0`. It cannot be processed again, so create a new video and upload the file to that one. -Poll this endpoint every few seconds after completing an upload, and stop when `status` is `ready` or `errored`. Each call counts toward the key's 60 requests per minute. +Poll this endpoint every few seconds after completing an upload, and stop when `status` is `ready` or `errored`. Each call counts toward the key's 300 requests per minute. A video is `ready` as soon as its first output can be played, so playback can start while the rest is still encoding. The playback URLs are `null` until then. `hlsUrl` is an adaptive HLS master playlist, `mp4Url` is a progressive MP4 and `posterUrl` is a poster image. They need no API key and can be loaded from any origin. [Play a Video](/docs/api/rest-api#play-a-video) shows how to use them, and [Delete Video](/docs/api/rest-api#delete-video) says when they stop working. @@ -622,7 +624,7 @@ curl -X DELETE https://cap.so/api/developer/v1/videos/abc123def456 \ Send an audio or video file and get text back. Transcription costs $0.10 per audio hour, billed by the second from the app's credit balance. -A transcript moves through these statuses: `uploading` (waiting for its file), `queued`, `processing`, then `complete` or `error`. Poll [Get Transcript](/docs/api/rest-api#get-transcript) every few seconds until the status is `complete` or `error`. Each key allows 60 requests per minute. +A transcript moves through these statuses: `uploading` (waiting for its file), `queued`, `processing`, then `complete` or `error`. Poll [Get Transcript](/docs/api/rest-api#get-transcript) every few seconds until the status is `complete` or `error`. Each key allows 300 requests per minute. An app can have 20 transcripts `queued` or `processing` at once. Starting another answers `429` with the code `too_many_transcripts` and a `Retry-After` header of 15 seconds. Wait for one to finish, then send the request again. Transcripts in `uploading` do not count until they are started. @@ -1550,7 +1552,7 @@ The full video object returned by list and get endpoints: - **Use secret keys server-side only.** The `csk_` key should never appear in client-side code. Create each upload on your server and send only its `cut_` upload token to the browser. - **Poll video status after upload.** After completing a multipart upload, poll `GET /videos/:id/status` every few seconds until `status` is `ready` or `errored`, then play `hlsUrl` when it is present and `mp4Url` otherwise. Keep polling until `hlsUrl` is set if the player should use HLS. See [Play a Video](/docs/api/rest-api#play-a-video). - **Switch on `code`.** Every error carries a stable `code`. The `error` text can change. -- **Watch the rate limit headers.** A secret key allows 60 requests per minute and an upload token 1,200. Slow down as `X-RateLimit-Remaining` nears zero, and wait for `Retry-After` seconds after a `429`. +- **Watch the rate limit headers.** A secret key allows 300 requests per minute and an upload token 1,200. Slow down as `X-RateLimit-Remaining` nears zero, and wait for `Retry-After` seconds after a `429`. - **Store your keys securely.** Use environment variables or a secrets manager. - **Check who is uploading before you create an upload.** Your server decides who gets a token, so authenticate the user first and set `userId` to tie the video to them. - **Scope every read to the signed-in user.** Pass your session's user as `userId` when you list, get, check or delete a video on a user's behalf. Refuse the request on your side when there is no signed-in user, rather than sending an empty `userId`. diff --git a/apps/web/data/infra-agent-prompt.ts b/apps/web/data/infra-agent-prompt.ts index 4a23a7f1f2..e048b24a62 100644 --- a/apps/web/data/infra-agent-prompt.ts +++ b/apps/web/data/infra-agent-prompt.ts @@ -1,3 +1,4 @@ +import { DEVELOPER_KEY_REQUESTS_PER_MINUTE } from "@/lib/developer-rate-limits"; import { INFRA_MCP_PATH } from "@/lib/infra-mcp-path"; export const CAP_INFRA_POINTER_PROMPT = @@ -34,7 +35,7 @@ Cap Infra is Cap's video and speech API. All you need is the app's secret key an 5. Wait until it plays. From the server, poll ${calls.videoStatus} no faster than once every 2 seconds. Stop when status is ready or errored, and show errorReason if it failed. 6. Play it. Show the video with its embedUrl, or build a player from hlsUrl and mp4Url in ${calls.video}. Use hlsUrl when it is present and mp4Url otherwise. shareUrl is the link to send to someone. 7. Transcribe when I ask. Transcripts are billed per audio hour, so confirm with me first. For a public URL call ${calls.transcribeUrl}. For a file call ${calls.transcribeFile}, PUT the file to the signed URL it returns, then call ${calls.startTranscript}. Poll ${calls.transcript} at the same pace until status is complete or error. -8. Stay inside the limit. A secret key allows 60 requests a minute across every endpoint. After a 429, wait for the number of seconds in the Retry-After header before you send the request again. +8. Stay inside the limit. A secret key allows ${DEVELOPER_KEY_REQUESTS_PER_MINUTE} requests a minute across every endpoint. After a 429, wait for the number of seconds in the Retry-After header before you send the request again. 9. Watch the credits. ${calls.usage} shows the balance and whether uploads are allowed. A 402 means the app needs credit, so tell me and I will add it in the console. 10. Use the MCP server if you want tools. Cap Infra has a hosted MCP server at ${CAP_INFRA_MCP_URL} that takes the same secret key as Authorization: Bearer. Its requests count toward the same limit. diff --git a/apps/web/lib/developer-openapi.ts b/apps/web/lib/developer-openapi.ts index d72554f54b..4eac42d79c 100644 --- a/apps/web/lib/developer-openapi.ts +++ b/apps/web/lib/developer-openapi.ts @@ -6,6 +6,7 @@ import { } from "@/lib/developer-pricing"; import { DEVELOPER_KEY_REQUESTS_PER_MINUTE, + DEVELOPER_OWNER_REQUESTS_PER_MINUTE, DEVELOPER_RATE_LIMIT_WINDOW_SECONDS, DEVELOPER_UPLOAD_TOKEN_REQUESTS_PER_MINUTE, } from "@/lib/developer-rate-limits"; @@ -547,7 +548,7 @@ export const developerOpenApiSpec = (baseUrl: string) => ({ "", "Every error is a JSON object with an `error` string and a stable `code`, and usually a `hint` that says what to change. Validation errors (code `invalid_request`) add an `issues` array naming each failing field. A request body must be JSON sent with `Content-Type: application/json`: malformed JSON returns 400 `invalid_json` and any other content type returns 415 `unsupported_media_type`. A known path called with a method it does not support returns 405 `method_not_allowed` with an Allow header, and an unknown path returns 404 `not_found`. A 4xx response will not succeed if the same request is sent again, with two exceptions that say to wait and repeat it: 429, and 409 with code `completion_in_progress`. Retry 5xx with backoff.", "", - `A secret key is limited to ${DEVELOPER_KEY_REQUESTS_PER_MINUTE} requests per minute across the REST operations, and each upload token to ${DEVELOPER_UPLOAD_TOKEN_REQUESTS_PER_MINUTE} requests per minute across the SDK operations. Responses carry X-RateLimit-Limit, X-RateLimit-Remaining and X-RateLimit-Reset, and a 429 also carries Retry-After. Remaining and Reset are what the server that answered has counted, so they are a guide and Remaining can read higher than what is left. A 429 is authoritative: wait for its Retry-After seconds, at most ${DEVELOPER_RATE_LIMIT_WINDOW_SECONDS}, then retry.`, + `A secret key is limited to ${DEVELOPER_KEY_REQUESTS_PER_MINUTE} requests per minute across the REST operations, all the keys of one account together to ${DEVELOPER_OWNER_REQUESTS_PER_MINUTE}, and each upload token to ${DEVELOPER_UPLOAD_TOKEN_REQUESTS_PER_MINUTE} requests per minute across the SDK operations. Responses to an accepted key or token carry X-RateLimit-Limit, X-RateLimit-Remaining and X-RateLimit-Reset, and a 429 also carries Retry-After. Remaining and Reset are what the server that answered has counted, so they are a guide and Remaining can read higher than what is left. A 429 is authoritative: wait for its Retry-After seconds, at most ${DEVELOPER_RATE_LIMIT_WINDOW_SECONDS}, then retry.`, "", "The REST operations take a secret key and send no CORS headers, so a page cannot call them. Call them from your server. The SDK operations and the playback URLs can be called from any origin.", "", @@ -2212,7 +2213,7 @@ export const developerOpenApiSpec = (baseUrl: string) => ({ RateLimited: { ...json( ref("Error"), - `More requests in a minute than the credential allows: ${DEVELOPER_KEY_REQUESTS_PER_MINUTE} for a secret key, ${DEVELOPER_UPLOAD_TOKEN_REQUESTS_PER_MINUTE} for an upload token (code rate_limited).`, + `More requests in a minute than the credential allows: ${DEVELOPER_KEY_REQUESTS_PER_MINUTE} for a secret key, ${DEVELOPER_OWNER_REQUESTS_PER_MINUTE} for all the keys of one account, ${DEVELOPER_UPLOAD_TOKEN_REQUESTS_PER_MINUTE} for an upload token (code rate_limited).`, ), headers: { ...rateLimitHeaders, diff --git a/apps/web/lib/developer-rate-limits.ts b/apps/web/lib/developer-rate-limits.ts index 8579ba0100..5da9b83a79 100644 --- a/apps/web/lib/developer-rate-limits.ts +++ b/apps/web/lib/developer-rate-limits.ts @@ -1,6 +1,12 @@ export const DEVELOPER_RATE_LIMIT_WINDOW_SECONDS = 60; -export const DEVELOPER_KEY_REQUESTS_PER_MINUTE = 60; +export const DEVELOPER_KEY_REQUESTS_PER_MINUTE = 300; + +// One account can hold many apps, each with its own key. Their requests also +// count together, so more apps do not mean an ever larger total. +export const DEVELOPER_OWNER_REQUESTS_PER_MINUTE = 1200; + +export const DEVELOPER_AUTH_FAILURES_PER_MINUTE = 30; // An upload signs one part per request and the smallest part is 5 MB, so this // is the ceiling on upload speed: 1,200 parts a minute is 6 GB a minute. diff --git a/apps/web/lib/developer-video-processing.ts b/apps/web/lib/developer-video-processing.ts index 72dfe7a3ab..60c01bf603 100644 --- a/apps/web/lib/developer-video-processing.ts +++ b/apps/web/lib/developer-video-processing.ts @@ -732,7 +732,9 @@ export async function dispatchDeveloperVideoPart( if (started.status === "busy" || started.status === "unreachable") { await mutateEncode(videoId, encodeId, (processing) => { const released = releasePart(processing, part, token); - return released ? { processing: released } : null; + return released + ? { processing: { ...released, busyAt: Date.now() } } + : null; }); return "busy"; } diff --git a/apps/web/lib/developer-video-state.ts b/apps/web/lib/developer-video-state.ts index bb41e0e736..9c348e0364 100644 --- a/apps/web/lib/developer-video-state.ts +++ b/apps/web/lib/developer-video-state.ts @@ -361,8 +361,13 @@ export function developerVideoRedrive( : null; } if (video.status === "processing") { + // A video that waited for a free worker is timed from its last busy + // answer, so one that starts encoding late is not started over while it + // encodes. One whose workflow stopped stops being answered busy. return !video.processing || - now - video.processing.startedAt >= STALE_PROCESSING_MS + now - + Math.max(video.processing.startedAt, video.processing.busyAt ?? 0) >= + STALE_PROCESSING_MS ? "start" : null; } diff --git a/apps/web/lib/infra-video-upload.ts b/apps/web/lib/infra-video-upload.ts index 9a6b5d199d..2dd6966ab8 100644 --- a/apps/web/lib/infra-video-upload.ts +++ b/apps/web/lib/infra-video-upload.ts @@ -888,15 +888,53 @@ const UNRECORDED_FILE_MAX_AGE_MS = 24 * 60 * 60 * 1000; const UNRECORDED_PAGE = 200; const UNRECORDED_ROWS_PER_RUN = 1_000; +type StoredUpload = "stale" | "fresh" | "uploading" | "missing" | "unknown"; + +function storedUpload(s3Key: string, now: Date): Promise { + return Effect.gen(function* () { + const [bucket] = yield* S3Buckets.getBucketAccess(Option.none()); + const head = yield* Effect.either(bucket.headObject(s3Key)); + if (head._tag === "Right") { + const modified = head.right.LastModified?.getTime() ?? 0; + return modified <= now.getTime() - UNRECORDED_FILE_MAX_AGE_MS + ? ("stale" as const) + : ("fresh" as const); + } + if (!storedObjectIsMissing(head.left)) return "unknown" as const; + const open = yield* bucket.multipart.listUploads({ prefix: s3Key }); + if ( + open.IsTruncated || + (open.Uploads ?? []).some((upload) => upload.Key === s3Key) + ) { + return "uploading" as const; + } + // An upload assembled between the two reads above shows in neither, so + // the file is looked for once more now that no upload is open. + const again = yield* Effect.either(bucket.headObject(s3Key)); + if (again._tag === "Right") return "fresh" as const; + return storedObjectIsMissing(again.left) + ? ("missing" as const) + : ("unknown" as const); + }) + .pipe(provideOptionalAuth, runPromise) + .catch(() => "unknown" as const); +} + +// A video with no file is retired only after storage is asked again under its +// row lock, and never while a completion's claim is recent. Initiate opens an +// upload under the same lock, and completion can assemble a file only from an +// upload that is still open. A claim left by a completion that died before +// assembling keeps the video only until that claim would be released anyway. async function retireUnrecordedVideo( appId: string, videoId: string, now: Date, + missingKey?: string, ) { return db().transaction(async (tx) => { await lockInfraAccount(tx, appId); - const [current] = await tx + const read = tx .select({ duration: developerVideos.duration, deletedAt: developerVideos.deletedAt, @@ -904,8 +942,28 @@ async function retireUnrecordedVideo( .from(developerVideos) .where(eq(developerVideos.id, videoId)) .limit(1); + const [current] = await (missingKey === undefined + ? read + : read.for("update")); if (!current || current.duration !== null) return false; if (current.deletedAt) return true; + if (missingKey !== undefined) { + const [claim] = await tx + .select({ + status: developerVideos.status, + updatedAt: developerVideos.updatedAt, + }) + .from(developerVideos) + .where(eq(developerVideos.id, videoId)) + .limit(1); + if ( + claim?.status === "processing" && + now.getTime() - claim.updatedAt.getTime() < STALE_UPLOAD_CLAIM_MS + ) { + return false; + } + if ((await storedUpload(missingKey, now)) !== "missing") return false; + } await tx .update(developerVideos) @@ -946,27 +1004,27 @@ async function unrecordedVideosPage( .limit(UNRECORDED_PAGE); } +// A video that never received a file is retired too, or abandoned uploads +// would stay listed for good. It is left alone while an upload for it is +// still open, and whenever storage cannot say for certain what is there. async function removeUnrecordedFile( row: { id: string; appId: string; s3Key: string | null }, now: Date, ) { const s3Key = row.s3Key; if (!s3Key) return false; - const stale = await Effect.gen(function* () { - const [bucket] = yield* S3Buckets.getBucketAccess(Option.none()); - const object = yield* bucket.headObject(s3Key); - const modified = object.LastModified?.getTime() ?? 0; - return modified <= now.getTime() - UNRECORDED_FILE_MAX_AGE_MS; - }) - .pipe(provideOptionalAuth, runPromise) - .catch(() => false); - if (!stale) return false; - const retired = await retireUnrecordedVideo(row.appId, row.id, now).catch( - (error) => { - console.error("Error retiring an unrecorded upload:", row.id, error); - return false; - }, - ); + const file = await storedUpload(s3Key, now); + if (file !== "stale" && file !== "missing") return false; + const retired = await retireUnrecordedVideo( + row.appId, + row.id, + now, + file === "missing" ? s3Key : undefined, + ).catch((error) => { + console.error("Error retiring an unrecorded upload:", row.id, error); + return false; + }); + if (file === "missing") return retired; return retired && (await removeStoredFile(s3Key)); } diff --git a/apps/web/lib/rate-limit.ts b/apps/web/lib/rate-limit.ts index de19a60980..47738a6681 100644 --- a/apps/web/lib/rate-limit.ts +++ b/apps/web/lib/rate-limit.ts @@ -1,5 +1,6 @@ import { checkRateLimit } from "@vercel/firewall"; import { headers as nextHeaders } from "next/headers"; +import { DEVELOPER_AUTH_FAILURES_PER_MINUTE } from "@/lib/developer-rate-limits"; const FIREWALL_FORWARDED_HEADER_NAMES = [ "host", @@ -223,10 +224,14 @@ export const RATE_LIMIT_IDS = { DEVELOPER_APP_CREATE: "rl_developer_app_create", /** Developer key rotations from the console and the agent API. Suggested: 10 / 1h per key (user). */ DEVELOPER_KEY_ROTATE: "rl_developer_key_rotate", - /** Developer REST API requests per secret key. 60 / 1m per key (hash of the credential). */ + /** Developer REST API requests per secret key. 300 / 1m per key (hash of the credential). */ DEVELOPER_API_KEY: "rl_developer_api_key", /** Developer SDK upload requests per upload token. 1200 / 1m per key (hash of the credential). */ DEVELOPER_UPLOAD_TOKEN: "rl_developer_upload_token", + /** Developer REST API requests per account, across all of its keys. 1200 / 1m per key (owner id). */ + DEVELOPER_API_OWNER: "rl_developer_api_owner", + /** Developer API requests whose key or upload token was not accepted. 30 / 1m per IP. */ + DEVELOPER_AUTH_FAILURE: "rl_developer_auth_failure", } as const; type FallbackRateLimit = { limit: number; windowMs: number }; @@ -290,6 +295,10 @@ const FALLBACK_RATE_LIMITS: Record = { limit: 10, windowMs: 60 * MINUTE_MS, }, + [RATE_LIMIT_IDS.DEVELOPER_AUTH_FAILURE]: { + limit: DEVELOPER_AUTH_FAILURES_PER_MINUTE, + windowMs: MINUTE_MS, + }, }; const DEFAULT_FALLBACK_RATE_LIMIT: FallbackRateLimit = { diff --git a/apps/web/workflows/process-developer-video.ts b/apps/web/workflows/process-developer-video.ts index 2e16734644..b92c775934 100644 --- a/apps/web/workflows/process-developer-video.ts +++ b/apps/web/workflows/process-developer-video.ts @@ -56,7 +56,7 @@ export async function processDeveloperVideoWorkflow( const { videoId, encodeId } = payload; try { - const deadline = Date.now() + PROCESSING_DEADLINE_MS; + let deadline = Date.now() + PROCESSING_DEADLINE_MS; for (let unclear = 0; ; ) { const prepared = await prepareVideo(videoId, encodeId); @@ -103,6 +103,17 @@ export async function processDeveloperVideoWorkflow( Promise.all(silent.map((part) => checkPart(videoId, encodeId, part))), ]); const busy = dispatched.includes("busy"); + // Every worker answering busy while nothing of this video is encoding + // is a wait for room, not processing time. The deadline starts over, so + // a long line of other encodes does not time this video out. + if ( + dispatched.length > 0 && + dispatched.every((answer) => answer === "busy") && + silent.length === 0 && + !waiting + ) { + deadline = Date.now() + PROCESSING_DEADLINE_MS; + } await sleep( `${busy ? RETRY_SECONDS : Math.min(MAX_POLL_SECONDS, 2 + poll * 2)}s`, ); diff --git a/packages/database/types/developer-video.ts b/packages/database/types/developer-video.ts index 646a741045..353ee492b3 100644 --- a/packages/database/types/developer-video.ts +++ b/packages/database/types/developer-video.ts @@ -57,6 +57,8 @@ export interface DeveloperVideoProcessingState { mp4: DeveloperVideoPartState; attempts: Record; lostJobs?: Partial>; + // When every worker last answered busy to a start of this encode. + busyAt?: number; unavailableChecks: number; poster: { output: string } | null; bytes: {