diff --git a/apps/cloud/src/engine/execution-rate-limit.node.test.ts b/apps/cloud/src/engine/execution-rate-limit.node.test.ts new file mode 100644 index 000000000..22d570bce --- /dev/null +++ b/apps/cloud/src/engine/execution-rate-limit.node.test.ts @@ -0,0 +1,171 @@ +import { describe, expect, it } from "@effect/vitest"; +import { Data, Effect } from "effect"; + +import type { ExecutionEngine } from "@executor-js/execution"; + +import { makeExecutionRateLimiter } from "./execution-rate-limit"; +import { RATE_LIMIT_BLOCKED_MESSAGE } from "./execution-limit-messages"; + +const ORG = "org_test"; + +/** Stands in for whatever the real lookups fail with (Autumn down, DO down). */ +class UpstreamDownError extends Data.TaggedError("UpstreamDownError")<{ + readonly which: string; +}> {} + +// A stand-in engine: `execute` resolves to a marker, so a test tells an allowed +// execution (marker) from a blocked one (the gate's error result) by which of +// the two came back. A blocked decision never reaches the engine at all. +const engineStub: ExecutionEngine = { + execute: () => Effect.succeed({ result: "ran" }), + executeWithPause: () => Effect.succeed({ status: "completed", result: { result: "ran" } }), + resume: () => Effect.succeed(null), + getPausedExecution: () => Effect.succeed(null), + pausedExecutionCount: () => Effect.succeed(0), + hasPausedExecutions: () => Effect.succeed(false), + getDescription: Effect.succeed("stub"), +}; + +/** Counter that hands out a caller-controlled sequence of counts. */ +const countingIncrement = (counts: ReadonlyArray) => { + let calls = 0; + return () => Effect.succeed(counts[Math.min(calls++, counts.length - 1)] ?? 0); +}; + +const runExecute = (limiter: ReturnType) => + Effect.runPromise( + limiter.decorate(ORG, engineStub).execute("code", { + // Never invoked: the stub ignores it, and a blocked execution never runs. + onElicitation: () => Effect.die("elicitation is not exercised here"), + }), + ); + +describe("execution rate limiter — paid exemption", () => { + it("allows executions under the cap without consulting the exemption", async () => { + let exemptionCalls = 0; + const limiter = makeExecutionRateLimiter(countingIncrement([1]), { + limit: 10, + isExempt: () => { + exemptionCalls += 1; + return Effect.succeed(false); + }, + }); + + expect(await runExecute(limiter)).toMatchObject({ result: "ran" }); + // The whole point of resolving lazily: the common path costs no lookup. + expect(exemptionCalls).toBe(0); + }); + + it("blocks a non-exempt org over the cap", async () => { + const limiter = makeExecutionRateLimiter(countingIncrement([11]), { + limit: 10, + isExempt: () => Effect.succeed(false), + }); + + expect(await runExecute(limiter)).toMatchObject({ + result: null, + error: RATE_LIMIT_BLOCKED_MESSAGE, + }); + }); + + it("allows an exempt org over the cap", async () => { + const limiter = makeExecutionRateLimiter(countingIncrement([11]), { + limit: 10, + isExempt: () => Effect.succeed(true), + }); + + expect(await runExecute(limiter)).toMatchObject({ result: "ran" }); + }); + + it("caches the exemption so a paid org past the cap looks it up once", async () => { + let exemptionCalls = 0; + const limiter = makeExecutionRateLimiter(countingIncrement([11, 12, 13]), { + limit: 10, + exemptionTtlMs: 60_000, + now: () => 1_000, + isExempt: () => { + exemptionCalls += 1; + return Effect.succeed(true); + }, + }); + + for (let i = 0; i < 3; i += 1) + expect(await runExecute(limiter)).toMatchObject({ result: "ran" }); + expect(exemptionCalls).toBe(1); + }); + + it("re-resolves once the cached exemption expires", async () => { + let exemptionCalls = 0; + let clock = 1_000; + const limiter = makeExecutionRateLimiter(countingIncrement([11, 12]), { + limit: 10, + exemptionTtlMs: 1_000, + now: () => clock, + isExempt: () => { + exemptionCalls += 1; + return Effect.succeed(true); + }, + }); + + await runExecute(limiter); + clock += 5_000; + await runExecute(limiter); + expect(exemptionCalls).toBe(2); + }); + + it("blocks when the exemption cannot be resolved and nothing is cached", async () => { + // Deliberately NOT fail-open: an unresolvable exemption during an Autumn + // outage must not switch the backstop off, which is the one scenario it + // exists to cover. + const limiter = makeExecutionRateLimiter(countingIncrement([11]), { + limit: 10, + isExempt: () => Effect.fail(new UpstreamDownError({ which: "autumn" })), + }); + + expect(await runExecute(limiter)).toMatchObject({ + result: null, + error: RATE_LIMIT_BLOCKED_MESSAGE, + }); + }); + + it("honours a stale exemption when a later lookup fails", async () => { + let clock = 1_000; + let shouldFail = false; + const limiter = makeExecutionRateLimiter(countingIncrement([11, 12]), { + limit: 10, + exemptionTtlMs: 1_000, + now: () => clock, + isExempt: () => + shouldFail ? Effect.fail(new UpstreamDownError({ which: "autumn" })) : Effect.succeed(true), + }); + + expect(await runExecute(limiter)).toMatchObject({ result: "ran" }); + + // Cache expires and Autumn is now unreachable: the known-paid org keeps + // running rather than getting blocked mid-workload by a blip. + clock += 5_000; + shouldFail = true; + expect(await runExecute(limiter)).toMatchObject({ result: "ran" }); + }); + + it("fails open when the counter itself is unreachable", async () => { + const limiter = makeExecutionRateLimiter( + () => Effect.fail(new UpstreamDownError({ which: "counter DO" })), + { + limit: 10, + isExempt: () => Effect.succeed(false), + }, + ); + + expect(await runExecute(limiter)).toMatchObject({ result: "ran" }); + }); + + it("applies the cap when no exemption predicate is wired", async () => { + const limiter = makeExecutionRateLimiter(countingIncrement([11]), { limit: 10 }); + + expect(await runExecute(limiter)).toMatchObject({ + result: null, + error: RATE_LIMIT_BLOCKED_MESSAGE, + }); + }); +}); diff --git a/apps/cloud/src/engine/execution-rate-limit.ts b/apps/cloud/src/engine/execution-rate-limit.ts index b68d4c044..3b5ea8194 100644 --- a/apps/cloud/src/engine/execution-rate-limit.ts +++ b/apps/cloud/src/engine/execution-rate-limit.ts @@ -1,5 +1,5 @@ // --------------------------------------------------------------------------- -// Per-org execution rate limit — an abuse backstop independent of billing. +// Per-org execution rate limit — a free-tier abuse backstop. // // The balance gate (execution-gate.ts) depends on Autumn and fails open, so a // billing outage plus runaway automation could still run unbounded executions. @@ -8,9 +8,22 @@ // each MCP session lives in its own DO instance, so an in-memory counter // would be per-session and trivially bypassed by opening more sessions). // -// Like the balance gate it FAILS OPEN: an unreachable counter DO, a missing +// Paid organizations are EXEMPT. The cap was sized for free-tier abuse but +// applied to everyone, and on 2026-08-18 it blocked a paying customer +// mid-workload — their agent gave up on Executor and routed around it. Paid +// usage is what the balance gate and metered overage are for; this backstop +// has no business capping it. +// +// The exemption is resolved ONLY once the counter reports an org over the cap, +// so the common path (under the cap) costs the counter increment and nothing +// else. `isExempt` is an opaque predicate: this module still names no billing +// concept, and the Autumn coupling lives in `execution-stack-metered.ts`, +// which already owns that dependency. +// +// FAIL OPEN applies to the COUNTER: an unreachable counter DO, a missing // binding, or a slow call allows the execution (warn + Sentry). The backstop -// must never take executions down with it. +// must never take executions down with it. An unresolved EXEMPTION is the one +// thing that does not fail open — see `resolveExemption`. // --------------------------------------------------------------------------- import { DurableObject, env } from "cloudflare:workers"; @@ -25,14 +38,26 @@ import { RATE_LIMIT_BLOCKED_MESSAGE } from "./execution-limit-messages"; // Fixed window: all executions in the same clock hour share one counter. export const RATE_LIMIT_WINDOW_MS = 3_600_000; -// Calibration: the heaviest legitimate org runs ~1.1k executions per MONTH, -// so 1000 per HOUR is far above any human-driven usage and only trips on -// runaway automation (the incident this backstops: ~18k in 30 days would -// still pass, which is fine — that class of overrun is the balance gate's -// job; this catches tight loops). +// The cap for organizations WITHOUT a paid subscription. +// +// The original calibration ("the heaviest legitimate org runs ~1.1k executions +// per MONTH, so 1000 per HOUR is far above any human-driven usage") went stale +// inside six weeks: by 2026-08-18 a paying org was sustaining ~5.3k executions +// per DAY and crossed this cap in a single hour. Sizing a shared number +// against the largest customer is a losing game, so the number no longer tries +// to describe them — paid orgs are exempt below, and this now has only +// free-tier abuse to cover, which is what it was picked for. export const EXECUTIONS_PER_ORG_PER_HOUR = 1000; // Counter DO slower than this => fail open rather than stall executions. const RATE_LIMIT_CHECK_TIMEOUT_MS = 2_000; +// Exemption lookup slower than this => treat as unresolved. +const EXEMPTION_CHECK_TIMEOUT_MS = 2_000; +// An org over the cap is checked at most once per TTL rather than once per +// execution, so a paid org running far past it doesn't hammer the lookup. +const EXEMPTION_CACHE_TTL_MS = 60_000; +// Sweep guard, mirroring the balance gate's cache: one long-lived isolate can +// serve many orgs. +const EXEMPTION_CACHE_MAX_ENTRIES = 10_000; // The DO purges its storage this long after the last increment, so idle orgs // cost nothing. Two windows: long enough that an active window never purges. const COUNTER_PURGE_AFTER_MS = 2 * RATE_LIMIT_WINDOW_MS; @@ -51,6 +76,20 @@ class RateLimitCheckTimeoutError extends Data.TaggedError("RateLimitCheckTimeout readonly timeoutMs: number; }> {} +/** Internal sentinel for an exemption lookup that exceeded its time budget. */ +class ExemptionCheckTimeoutError extends Data.TaggedError("ExemptionCheckTimeoutError")<{ + readonly timeoutMs: number; +}> {} + +/** + * Whether an organization is exempt from the cap. Production passes a paid- + * subscription check; keeping it an opaque predicate is what lets this module + * stay free of any billing import. + */ +export type ExecutionRateLimitExemption = ( + organizationId: string, +) => Effect.Effect; + // --------------------------------------------------------------------------- // Counter Durable Object — one instance per organization (idFromName(orgId)). // Stores a single { windowId, count } record: an increment in a new window @@ -119,12 +158,76 @@ export const makeExecutionRateLimiter = ( readonly windowMs?: number; readonly timeoutMs?: number; readonly now?: () => number; + readonly isExempt?: ExecutionRateLimitExemption; + readonly exemptionTtlMs?: number; }, ): ExecutionRateLimiter => { const limit = options?.limit ?? EXECUTIONS_PER_ORG_PER_HOUR; const windowMs = options?.windowMs ?? RATE_LIMIT_WINDOW_MS; const timeoutMs = options?.timeoutMs ?? RATE_LIMIT_CHECK_TIMEOUT_MS; const now = options?.now ?? Date.now; + const isExempt = options?.isExempt; + const exemptionTtlMs = options?.exemptionTtlMs ?? EXEMPTION_CACHE_TTL_MS; + + const exemptionCache = new Map< + string, + { readonly exempt: boolean; readonly expiresAtMs: number } + >([]); + + const writeExemptionCache = (organizationId: string, exempt: boolean, nowMs: number): void => { + if (exemptionCache.size >= EXEMPTION_CACHE_MAX_ENTRIES) { + for (const [key, entry] of exemptionCache) { + if (entry.expiresAtMs <= nowMs) exemptionCache.delete(key); + } + // Still saturated after dropping expired entries: reset rather than grow. + if (exemptionCache.size >= EXEMPTION_CACHE_MAX_ENTRIES) exemptionCache.clear(); + } + exemptionCache.set(organizationId, { exempt, expiresAtMs: nowMs + exemptionTtlMs }); + }; + + /** + * Resolved only for orgs already over the cap, so the lookup never touches + * the common path. + * + * This is the one place that does NOT fail open. The balance gate already + * allows executions when Autumn is unreachable; if the exemption did too, + * an Autumn outage would switch this backstop off entirely — precisely the + * "billing outage plus runaway automation" case it exists to cover. A stale + * positive is honoured ahead of that fallback, so a blip can't flip a + * known-paid org into a block mid-workload. + */ + const resolveExemption = (organizationId: string): Effect.Effect => + Effect.suspend(() => { + if (!isExempt) return Effect.succeed(false); + const nowMs = now(); + const cached = exemptionCache.get(organizationId); + if (cached && cached.expiresAtMs > nowMs) return Effect.succeed(cached.exempt); + return isExempt(organizationId).pipe( + Effect.timeoutOrElse({ + duration: `${EXEMPTION_CHECK_TIMEOUT_MS} millis`, + orElse: () => + Effect.fail(new ExemptionCheckTimeoutError({ timeoutMs: EXEMPTION_CHECK_TIMEOUT_MS })), + }), + Effect.map((exempt) => { + writeExemptionCache(organizationId, exempt, nowMs); + return exempt; + }), + Effect.catch((error: unknown) => + Effect.gen(function* () { + yield* Effect.sync(() => { + console.warn( + `[rate-limit] exemption lookup failed for ${organizationId}; treating as ${ + cached ? "last known" : "not exempt" + }:`, + error, + ); + }); + yield* captureCauseEffect(error); + return cached?.exempt ?? false; + }), + ), + ); + }); const decide = (organizationId: string): Effect.Effect => Effect.suspend(() => { @@ -134,18 +237,34 @@ export const makeExecutionRateLimiter = ( duration: `${timeoutMs} millis`, orElse: () => Effect.fail(new RateLimitCheckTimeoutError({ timeoutMs })), }), - Effect.map( - (count): GateDecision => - count > limit - ? { - blocked: true, - error: new ExecutionRateLimitExceededError({ - organizationId, - message: RATE_LIMIT_BLOCKED_MESSAGE, - }), - } - : { blocked: false }, - ), + Effect.flatMap((count): Effect.Effect => { + // Under the cap: no exemption lookup, no extra I/O. + if (count <= limit) return Effect.succeed({ blocked: false }); + return Effect.gen(function* () { + if (yield* resolveExemption(organizationId)) { + return { blocked: false } as const satisfies GateDecision; + } + // The only record that the backstop fired. A blocked execution is + // never usage-tracked (the gate short-circuits before the tracker) + // and deliberately not sent to Sentry — a backstop stopping + // runaway automation is expected, not exceptional — so without + // this line a blocked org is invisible outside a bug report, which + // is how the 2026-08-18 block went unnoticed until a customer + // sent a screenshot. + yield* Effect.sync(() => { + console.warn( + `[rate-limit] blocked execution for ${organizationId}: ${count} > ${limit} in window ${windowId}`, + ); + }); + return { + blocked: true, + error: new ExecutionRateLimitExceededError({ + organizationId, + message: RATE_LIMIT_BLOCKED_MESSAGE, + }), + } as const satisfies GateDecision; + }); + }), // FAIL OPEN: the backstop must never block executions because its // counter is unreachable or slow. Effect.catch((error: unknown) => @@ -185,8 +304,13 @@ type RateLimiterNamespace = { * Production rate limiter backed by the `EXECUTION_RATE_LIMITER` counter DO. * When the binding is absent (unit-test workers, older local setups) the * limiter is disabled: every check passes, logged once at construction. + * + * `isExempt` decides which orgs the cap skips; production passes a paid- + * subscription check from `execution-stack-metered.ts`. */ -export const makeCloudExecutionRateLimiter = (): ExecutionRateLimiter => { +export const makeCloudExecutionRateLimiter = ( + isExempt: ExecutionRateLimitExemption, +): ExecutionRateLimiter => { const limit = resolveRateLimit(); const namespace = (env as { EXECUTION_RATE_LIMITER?: RateLimiterNamespace }) .EXECUTION_RATE_LIMITER; @@ -204,7 +328,7 @@ export const makeCloudExecutionRateLimiter = (): ExecutionRateLimiter => { ) as ExecutionRateLimiterStub; return stub.increment(windowId); }), - { limit }, + { limit, isExempt }, ); }; diff --git a/apps/cloud/src/engine/execution-stack-metered.ts b/apps/cloud/src/engine/execution-stack-metered.ts index 4ea70b6d4..6063e1456 100644 --- a/apps/cloud/src/engine/execution-stack-metered.ts +++ b/apps/cloud/src/engine/execution-stack-metered.ts @@ -26,6 +26,7 @@ import { } from "@executor-js/api/server"; import { AutumnService } from "../extensions/billing/service"; +import { hasPaidOrganizationSubscription } from "../extensions/billing/plans"; import type { DbService } from "../db/db"; import { CloudExecutionSeamsLayer } from "../engine/execution-stack"; import { makeExecutionLimitGate } from "./execution-gate"; @@ -34,7 +35,8 @@ import { withExecutionUsageTracking } from "./execution-usage"; // Usage-metering decorator bound to the billing service, plus the two // pre-execution guards this layer owns, ordered cheapest first: -// 1. rate-limit backstop (counter DO, independent of billing) +// 1. rate-limit backstop (counter DO; free-tier abuse only — paid orgs are +// exempt via the subscription lookup below) // 2. execution balance gate (Autumn check, cached 60s, fails open) // 3. usage tracking — fire-and-forget (`Effect.runFork`) so the billing // call can't stall a user-facing execution. @@ -48,7 +50,18 @@ export const CloudMeteringEngineDecorator: Layer.Layer autumn.checkExecutionBalance(organizationId), ); - const rateLimiter = makeCloudExecutionRateLimiter(); + // The limiter's paid-org exemption. This is the billing coupling the + // limiter module deliberately avoids owning, and it reads the same + // `PAID_AUTUMN_PLAN_IDS` config as the org-creation and seat gates so + // "paid" means one thing across the app. The limiter calls this only + // for orgs already over the cap and caches the answer, so the extra + // Autumn round trip stays off the hot path. + const rateLimiter = makeCloudExecutionRateLimiter((organizationId) => + Effect.map( + autumn.use((client) => client.customers.getOrCreate({ customerId: organizationId })), + (customer) => hasPaidOrganizationSubscription(customer.subscriptions), + ), + ); return { decorate: (engine, identity: EngineStackIdentity) => rateLimiter.decorate( diff --git a/e2e/cloud/mcp-execution-limits.test.ts b/e2e/cloud/mcp-execution-limits.test.ts index f80dcaac2..4c3272267 100644 --- a/e2e/cloud/mcp-execution-limits.test.ts +++ b/e2e/cloud/mcp-execution-limits.test.ts @@ -198,3 +198,52 @@ scenario( ); }), ); + +scenario( + "Billing · a paid org runs past the rate-limit backstop instead of being blocked", + { timeout: 180_000 }, + Effect.gen(function* () { + // The backstop is sized for free-tier abuse, and it used to apply to every + // org regardless of plan: on 2026-08-18 it blocked a paying customer + // mid-workload. This pins the exemption end to end — the same run that + // blocks a free org (the scenario above) must sail past the cap here. + yield* Billing; + const autumn = yield* Autumn; + const target = yield* Target; + const mcp = yield* Mcp; + + const identity = yield* target.newIdentity(); + const bearer = yield* mcp.mintBearer(emailOf(identity)); + const customerId = orgIdOf(bearer); + + // Enterprise, not Team: it carries no price and no card-required trial, so + // the emulator activates it inline rather than answering with a checkout + // URL. Its executions are unlimited, so the BALANCE gate can never be what + // allows or blocks here — the rate-limit backstop is isolated as the only + // guard under test. + yield* autumn.attachPlan(customerId, "enterprise"); + + const session = mcp.session(identity); + + // Deliberately past E2E_EXECUTION_RATE_LIMIT: the count that blocks a free + // org. Every one must run. + const overCap = RATE_LIMIT + 3; + for (let i = 1; i <= overCap; i++) { + const ok = yield* session.call("execute", { code: `return ${i};` }); + expect(ok.ok, `execution ${i} succeeds for a paid org (cap is ${RATE_LIMIT})`).toBe(true); + expect( + ok.text, + `execution ${i} returns its value rather than the backstop message`, + ).not.toContain(RATE_LIMIT_BLOCKED_MESSAGE); + } + + // Nothing was withheld from the meter either: a paid org past the cap is + // billed for every execution it ran. + const metered = yield* autumn.expectUsage({ + customerId, + featureId: "executions", + count: overCap, + }); + expect(metered.length, "every execution past the cap is still metered").toBe(overCap); + }), +); diff --git a/e2e/src/surfaces/autumn.ts b/e2e/src/surfaces/autumn.ts index a83149117..a401cf548 100644 --- a/e2e/src/surfaces/autumn.ts +++ b/e2e/src/surfaces/autumn.ts @@ -76,6 +76,14 @@ export interface AutumnSurface { * which drives `remaining` to zero. This is the setup a "blocked at cap" * scenario runs before opening the session it expects to be gated. */ readonly exhaustExecutions: (customerId: string) => Effect.Effect; + /** Put an org on a plan directly (`billing.attach`), skipping the hosted + * checkout. Only valid for a plan that needs no payment — Enterprise carries + * no price and no card-required trial, so the emulator activates its + * subscription inline. Team does have both, so it must still go through the + * browser checkout + `settleCheckout`. This is the setup a scenario runs when + * it needs a PAID org and the upgrade journey itself is not what's under + * test (e.g. the rate-limit backstop's paid exemption). */ + readonly attachPlan: (customerId: string, planId: string) => Effect.Effect; /** Arm a fault against a matching Autumn request (`POST /_emulate/faults`). */ readonly armFault: (input: FaultInput) => Effect.Effect; /** Clear every armed fault (`DELETE /_emulate/faults`) — the finalizer that @@ -155,6 +163,33 @@ export const makeAutumnSurface = (autumnUrl: string): AutumnSurface => { } }); + const attachPlan = (customerId: string, planId: string) => + Effect.gen(function* () { + const response = yield* Effect.promise(() => + fetch(`${autumnUrl}/v1/billing.attach`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ customer_id: customerId, plan_id: planId }), + }), + ); + if (!response.ok) { + return yield* Effect.fail( + `autumn billing.attach responded ${response.status}: ${yield* Effect.promise(() => response.text())}`, + ); + } + // A plan that needs payment answers with a checkout URL instead of + // activating; that silently leaves the org on Free, and a scenario that + // asked for a paid org would then assert against the wrong plan. + const body = (yield* Effect.promise(() => response.json())) as { + readonly payment_url?: string | null; + }; + if (body.payment_url) { + return yield* Effect.fail( + `autumn billing.attach returned a checkout URL for '${planId}': it requires payment, so it cannot be attached directly`, + ); + } + }); + const armFault = (input: FaultInput) => Effect.gen(function* () { const response = yield* Effect.promise(() => @@ -213,6 +248,7 @@ export const makeAutumnSurface = (autumnUrl: string): AutumnSurface => { usageEvents, settleCheckout, exhaustExecutions, + attachPlan, armFault, clearFaults, ledgerFor,