diff --git a/apps/cloud/src/engine/execution-rate-limit.node.test.ts b/apps/cloud/src/engine/execution-rate-limit.node.test.ts index 22d570bce..678697edc 100644 --- a/apps/cloud/src/engine/execution-rate-limit.node.test.ts +++ b/apps/cloud/src/engine/execution-rate-limit.node.test.ts @@ -1,9 +1,16 @@ -import { describe, expect, it } from "@effect/vitest"; +import { afterEach, describe, expect, it } from "@effect/vitest"; +import { env } from "cloudflare:workers"; import { Data, Effect } from "effect"; +import type * as Tracer from "effect/Tracer"; import type { ExecutionEngine } from "@executor-js/execution"; -import { makeExecutionRateLimiter } from "./execution-rate-limit"; +import { + ExecutionRateLimiterDO, + makeCloudExecutionRateLimiter, + makeExecutionRateLimiter, + RateLimitCounterError, +} from "./execution-rate-limit"; import { RATE_LIMIT_BLOCKED_MESSAGE } from "./execution-limit-messages"; const ORG = "org_test"; @@ -150,7 +157,15 @@ describe("execution rate limiter — paid exemption", () => { it("fails open when the counter itself is unreachable", async () => { const limiter = makeExecutionRateLimiter( - () => Effect.fail(new UpstreamDownError({ which: "counter DO" })), + (organizationId) => + Effect.fail( + new RateLimitCounterError({ + organizationId, + code: "unknown", + reason: "counter DO unreachable", + cause: null, + }), + ), { limit: 10, isExempt: () => Effect.succeed(false), @@ -169,3 +184,250 @@ describe("execution rate limiter — paid exemption", () => { }); }); }); + +// --------------------------------------------------------------------------- +// Counter observability — the production DO wiring. +// +// The counter increment used to be a bare one-argument `Effect.tryPromise` +// with no span, so a Durable Object fault reached error reporting as +// `UnknownError: An error occurred in Effect.tryPromise` with no application +// frames, and the 2s check budget was invisible in traces. These tests pin +// the named spans, the typed/classified failure, and the timeout override — +// all read through the REAL `makeCloudExecutionRateLimiter` wiring against a +// fake `EXECUTION_RATE_LIMITER` binding. +// --------------------------------------------------------------------------- + +type RecordedSpan = { + readonly name: string; + readonly attributes: Map; +}; + +/** A tracer that keeps every span it is asked to open, with its attributes. */ +const recordingTracer = (recorded: Array): Tracer.Tracer => { + let nextId = 1; + return { + span: (options) => { + let status: Tracer.SpanStatus = { _tag: "Started", startTime: options.startTime }; + const attributes = new Map(); + recorded.push({ name: options.name, attributes }); + const id = String(nextId++).padStart(16, "0"); + return { + _tag: "Span", + name: options.name, + spanId: id, + traceId: "00000000000000000000000000000001", + parent: options.parent, + annotations: options.annotations, + get status() { + return status; + }, + attributes, + links: options.links, + sampled: options.sampled, + kind: options.kind, + end: (endTime, exit) => { + status = { _tag: "Ended", startTime: options.startTime, endTime, exit }; + }, + attribute: (key, value) => { + attributes.set(key, value); + }, + event: () => undefined, + addLinks: () => undefined, + }; + }, + }; +}; + +const spanNamed = (recorded: ReadonlyArray, name: string): RecordedSpan => { + const span = recorded.find((candidate) => candidate.name === name); + expect( + span, + `a span named ${name} is recorded (got: ${recorded.map((s) => s.name).join(", ") || "none"})`, + ).toBeDefined(); + return span ?? { name, attributes: new Map() }; +}; + +/** A counter-DO namespace whose single RPC method behaves as the test says. */ +const namespaceReturning = (increment: () => Promise) => ({ + idFromName: (name: string) => ({ toString: () => name }), + get: () => ({ increment }), +}); + +/** The three worker vars the production limiter reads at construction. */ +type CounterEnv = { + EXECUTION_RATE_LIMITER?: unknown; + EXECUTION_RATE_LIMIT_PER_HOUR?: string; + EXECUTION_RATE_LIMIT_CHECK_TIMEOUT_MS?: string; +}; + +const cloudEnv: CounterEnv = env; +const savedEnv: CounterEnv = { ...cloudEnv }; + +// Restore rather than leak: the limiter is built from the worker env, and a +// stale binding or budget would silently retune a later test. +afterEach(() => { + delete cloudEnv.EXECUTION_RATE_LIMITER; + delete cloudEnv.EXECUTION_RATE_LIMIT_PER_HOUR; + delete cloudEnv.EXECUTION_RATE_LIMIT_CHECK_TIMEOUT_MS; + Object.assign(cloudEnv, savedEnv); +}); + +// The exact platform fault behind the production issue: Cloudflare resets the +// object and the RPC rejects with a plain Error. The reference id is synthetic. +const DO_STORAGE_RESET = + "Internal error in Durable Object storage caused object to be reset; reference = 0000000000000000"; + +// oxlint-disable-next-line executor/no-promise-reject, executor/no-error-constructor -- boundary: the Durable Object RPC is a promise that rejects with a platform Error; the test double has to fail the same way for the classification to mean anything +const rejectStorageReset = (): Promise => Promise.reject(new Error(DO_STORAGE_RESET)); + +/** Build the production limiter against a fake binding and worker env. */ +const cloudLimiter = (options: { + readonly namespace: ReturnType; + readonly limit: string; + readonly timeoutMs?: string; +}): ReturnType => { + cloudEnv.EXECUTION_RATE_LIMITER = options.namespace; + cloudEnv.EXECUTION_RATE_LIMIT_PER_HOUR = options.limit; + if (options.timeoutMs !== undefined) + cloudEnv.EXECUTION_RATE_LIMIT_CHECK_TIMEOUT_MS = options.timeoutMs; + return makeCloudExecutionRateLimiter(() => Effect.succeed(false)); +}; + +const runExecuteTraced = ( + limiter: ReturnType, + recorded: Array, +) => + Effect.runPromise( + limiter + .decorate(ORG, engineStub) + .execute("code", { onElicitation: () => Effect.die("elicitation is not exercised here") }) + .pipe(Effect.withTracer(recordingTracer(recorded))), + ); + +describe("execution rate limiter — counter observability", () => { + it("reports a counter-DO fault as a typed, classified error on a named span", async () => { + const recorded: Array = []; + const limiter = cloudLimiter({ + namespace: namespaceReturning(rejectStorageReset), + limit: "10", + }); + const result = await runExecuteTraced(limiter, recorded); + + // Fail-open semantics are unchanged: a broken counter never blocks. + expect(result).toMatchObject({ result: "ran" }); + + const increment = spanNamed(recorded, "rate_limit.increment"); + expect(increment.attributes.get("rate_limit.counter.error_tag")).toBe("RateLimitCounterError"); + expect(increment.attributes.get("rate_limit.counter.error_code")).toBe("storage_reset"); + + const check = spanNamed(recorded, "rate_limit.check"); + expect(check.attributes.get("rate_limit.check.failed_open")).toBe(true); + expect(check.attributes.get("rate_limit.check.timed_out")).toBe(false); + expect(check.attributes.get("rate_limit.check.error_tag")).toBe("RateLimitCounterError"); + }); + + it("honours the EXECUTION_RATE_LIMIT_CHECK_TIMEOUT_MS override", async () => { + const recorded: Array = []; + const limiter = cloudLimiter({ + // Answers well past the tiny budget, with a count far over the cap. If + // the override were ignored the default 2s budget would let that count + // through and BLOCK the execution; honouring it times the check out and + // fails open instead, so the two outcomes are distinguishable without + // measuring wall time. + namespace: namespaceReturning( + () => new Promise((resolve) => setTimeout(() => resolve(999), 200)), + ), + limit: "10", + timeoutMs: "5", + }); + const result = await runExecuteTraced(limiter, recorded); + + expect(result).toMatchObject({ result: "ran" }); + + const check = spanNamed(recorded, "rate_limit.check"); + expect(check.attributes.get("rate_limit.check.timed_out")).toBe(true); + expect(check.attributes.get("rate_limit.check.error_tag")).toBe("RateLimitCheckTimeoutError"); + }); + + // The healthy check's span (count / limit / blocked / failed_open) is NOT + // asserted here: the cloud e2e scenario "the rate-limit counter check is + // visible in the exported spans" pins it on the real workerd + Durable + // Object topology, against the spans the worker actually exports. The two + // cases above stay because neither is reachable from the e2e harness — it + // has no fault seam for a Durable Object RPC, and the check budget is a + // process-wide worker var that the shared cloud boot cannot vary per + // scenario without disabling the backstop for the whole run. +}); + +// --------------------------------------------------------------------------- +// Counter Durable Object +// --------------------------------------------------------------------------- + +/** Minimal DO storage that counts the calls the increment path makes. */ +const fakeStorage = () => { + const values = new Map(); + const calls = { put: 0, setAlarm: 0, deleteAll: 0 }; + return { + calls, + storage: { + get: (key: string) => Promise.resolve(values.get(key)), + put: (key: string, value: unknown) => { + calls.put += 1; + values.set(key, value); + return Promise.resolve(); + }, + setAlarm: () => { + calls.setAlarm += 1; + return Promise.resolve(); + }, + deleteAll: () => { + calls.deleteAll += 1; + values.clear(); + return Promise.resolve(); + }, + }, + }; +}; + +const makeCounter = (storage: ReturnType["storage"]) => + // oxlint-disable-next-line executor/no-double-cast -- test double: only the four storage methods the counter uses are implemented + new ExecutionRateLimiterDO({ storage } as unknown as DurableObjectState, {} as Env); + +describe("execution rate-limit counter DO", () => { + it("writes the purge alarm once per window instead of on every increment", async () => { + // The alarm only has to outlive the window; rewriting it on every call put + // a second durable write on the hot path with the input gate closed. + const fake = fakeStorage(); + const counter = makeCounter(fake.storage); + + expect(await counter.increment(7)).toBe(1); + expect(await counter.increment(7)).toBe(2); + expect(await counter.increment(7)).toBe(3); + + expect(fake.calls.put, "every increment still persists the count").toBe(3); + expect(fake.calls.setAlarm, "the purge alarm is written once for the window").toBe(1); + }); + + it("moves the purge alarm when the window rolls", async () => { + const fake = fakeStorage(); + const counter = makeCounter(fake.storage); + + await counter.increment(7); + await counter.increment(7); + expect(await counter.increment(8), "a new window restarts the count").toBe(1); + + expect(fake.calls.setAlarm, "one alarm write per window, not per increment").toBe(2); + }); + + it("re-arms the purge alarm after it has fired", async () => { + const fake = fakeStorage(); + const counter = makeCounter(fake.storage); + + await counter.increment(7); + await counter.alarm(); + expect(fake.calls.deleteAll, "the alarm purges the counter's storage").toBe(1); + + expect(await counter.increment(7), "the purge reset the window's count").toBe(1); + expect(fake.calls.setAlarm, "a purged counter schedules a fresh purge").toBe(2); + }); +}); diff --git a/apps/cloud/src/engine/execution-rate-limit.ts b/apps/cloud/src/engine/execution-rate-limit.ts index 3b5ea8194..906bee252 100644 --- a/apps/cloud/src/engine/execution-rate-limit.ts +++ b/apps/cloud/src/engine/execution-rate-limit.ts @@ -27,7 +27,7 @@ // --------------------------------------------------------------------------- import { DurableObject, env } from "cloudflare:workers"; -import { Data, Effect } from "effect"; +import { Data, Effect, Predicate } from "effect"; import type * as Cause from "effect/Cause"; import type { ExecutionEngine } from "@executor-js/execution"; @@ -76,6 +76,66 @@ class RateLimitCheckTimeoutError extends Data.TaggedError("RateLimitCheckTimeout readonly timeoutMs: number; }> {} +/** + * Why a counter DO call failed, as a small closed vocabulary. + * + * The counter's failures are overwhelmingly transient Cloudflare platform + * faults, and they used to arrive at error reporting as one untyped + * `UnknownError: An error occurred in Effect.tryPromise` with no application + * frames — a group that says nothing and would eventually swallow a real + * misconfiguration too. The code is what makes a storage reset (retryable, + * expected) distinguishable from an overload or an outright unknown fault. + */ +export type RateLimitCounterErrorCode = + | "storage_reset" + | "overloaded" + | "exceeded_memory" + | "network" + | "unknown"; + +/** A counter DO call that failed, carrying the classification and the org. */ +export class RateLimitCounterError extends Data.TaggedError("RateLimitCounterError")<{ + readonly organizationId: string; + readonly code: RateLimitCounterErrorCode; + readonly reason: string; + readonly cause: unknown; +}> {} + +// Cloudflare surfaces these as plain `Error`s with documented message text; +// there is no structured code to read, so the message is the only signal. +// (A shared `classifyDurableObjectError` would be the right home for this once +// one exists.) +const counterErrorCode = (reason: string): RateLimitCounterErrorCode => { + if (/caused object to be reset/i.test(reason)) return "storage_reset"; + if (/overloaded/i.test(reason)) return "overloaded"; + if (/exceeded (its )?memory|out of memory/i.test(reason)) return "exceeded_memory"; + if (/network connection lost|connection.*(lost|reset)/i.test(reason)) return "network"; + return "unknown"; +}; + +/** + * The fail-open landing: record the outcome on the check span, warn, and allow + * the execution. Only failures that are NOT deliberate degradation reach the + * error reporter — see the call sites in `decide`. + */ +const failOpen = ( + error: unknown, + outcome: { readonly errorTag: string; readonly timedOut: boolean }, +): Effect.Effect => + Effect.gen(function* () { + yield* Effect.annotateCurrentSpan({ + "rate_limit.blocked": false, + "rate_limit.check.failed_open": true, + "rate_limit.check.timed_out": outcome.timedOut, + "rate_limit.check.error_tag": outcome.errorTag, + }); + yield* Effect.sync(() => { + console.warn("[rate-limit] execution rate limit check failed open:", error); + }); + if (!outcome.timedOut) yield* captureCauseEffect(error); + return { blocked: false } as const satisfies GateDecision; + }); + /** Internal sentinel for an exemption lookup that exceeded its time budget. */ class ExemptionCheckTimeoutError extends Data.TaggedError("ExemptionCheckTimeoutError")<{ readonly timeoutMs: number; @@ -106,6 +166,12 @@ type WindowRecord = { export class ExecutionRateLimiterDO extends DurableObject { private readonly counterStorage: DurableObjectState["storage"]; + /** + * The window this instance has already armed the purge alarm for. In-memory + * on purpose: it costs no storage read, and a fresh instance (eviction, cold + * start) simply re-arms on its first increment. + */ + private purgeArmedForWindow: number | null = null; constructor(ctx: DurableObjectState, doEnv: Env) { super(ctx, doEnv); @@ -119,11 +185,21 @@ export class ExecutionRateLimiterDO extends DurableObject { const stored = await this.counterStorage.get(WINDOW_RECORD_KEY); const count = stored && stored.windowId === windowId ? stored.count + 1 : 1; await this.counterStorage.put(WINDOW_RECORD_KEY, { windowId, count }); - await this.counterStorage.setAlarm(Date.now() + COUNTER_PURGE_AFTER_MS); + // The alarm only has to outlive the window, and it is set two windows out, + // so once per window is enough — rewriting it on every increment put a + // second durable write and an alarm-manager update on the hot path of + // every execution, with the input gate closed across all three. `count` + // back at 1 means the window rolled (or a purge already ran), so the + // deadline moves with it. + if (count === 1 || this.purgeArmedForWindow !== windowId) { + await this.counterStorage.setAlarm(Date.now() + COUNTER_PURGE_AFTER_MS); + this.purgeArmedForWindow = windowId; + } return count; } async alarm(): Promise { + this.purgeArmedForWindow = null; await this.counterStorage.deleteAll(); } } @@ -132,11 +208,17 @@ export class ExecutionRateLimiterDO extends DurableObject { // Client // --------------------------------------------------------------------------- -/** Count one execution for (organizationId, windowId); returns the new count. */ +/** + * Count one execution for (organizationId, windowId); returns the new count. + * + * The failure channel is typed rather than `unknown` so the fail-open path can + * tell a counter fault from a blown budget by tag, and so error reporting + * groups by cause instead of by one opaque `UnknownError`. + */ export type RateLimitIncrement = ( organizationId: string, windowId: number, -) => Effect.Effect; +) => Effect.Effect; export type ExecutionRateLimiter = { readonly decorate: ( @@ -239,9 +321,25 @@ export const makeExecutionRateLimiter = ( }), Effect.flatMap((count): Effect.Effect => { // Under the cap: no exemption lookup, no extra I/O. - if (count <= limit) return Effect.succeed({ blocked: false }); + if (count <= limit) + return Effect.as( + Effect.annotateCurrentSpan({ + "rate_limit.count": count, + "rate_limit.blocked": false, + "rate_limit.check.failed_open": false, + }), + { blocked: false }, + ); return Effect.gen(function* () { + yield* Effect.annotateCurrentSpan({ + "rate_limit.count": count, + "rate_limit.check.failed_open": false, + }); if (yield* resolveExemption(organizationId)) { + yield* Effect.annotateCurrentSpan({ + "rate_limit.blocked": false, + "rate_limit.exempt": true, + }); return { blocked: false } as const satisfies GateDecision; } // The only record that the backstop fired. A blocked execution is @@ -256,6 +354,10 @@ export const makeExecutionRateLimiter = ( `[rate-limit] blocked execution for ${organizationId}: ${count} > ${limit} in window ${windowId}`, ); }); + yield* Effect.annotateCurrentSpan({ + "rate_limit.blocked": true, + "rate_limit.exempt": false, + }); return { blocked: true, error: new ExecutionRateLimitExceededError({ @@ -267,15 +369,36 @@ export const makeExecutionRateLimiter = ( }), // FAIL OPEN: the backstop must never block executions because its // counter is unreachable or slow. + // + // A check that blew its own budget is DELIBERATE degradation, not an + // exception — the timeout exists precisely so a slow counter can't + // stall a user-facing execution. It is measured on the span + // (`rate_limit.check.timed_out`), where a step change in the rate is + // alertable, rather than paged per occurrence, which is what buried + // real counter failures under one opaque group. Everything else (RPC + // faults, a missing binding) still reports. + Effect.catchTag("RateLimitCheckTimeoutError", (error) => + failOpen(error, { errorTag: "RateLimitCheckTimeoutError", timedOut: true }), + ), + // A catch-all rather than a second `catchTag`: fail-open is a hard + // requirement and must not depend on the failure being one the types + // predicted. Effect.catch((error: unknown) => - Effect.gen(function* () { - yield* Effect.sync(() => { - console.warn("[rate-limit] execution rate limit check failed open:", error); - }); - yield* captureCauseEffect(error); - return { blocked: false } as const satisfies GateDecision; + failOpen(error, { + errorTag: Predicate.isTagged(error, "RateLimitCounterError") + ? "RateLimitCounterError" + : "unknown", + timedOut: false, }), ), + Effect.withSpan("rate_limit.check", { + attributes: { + "rate_limit.organization_id": organizationId, + "rate_limit.window_id": windowId, + "rate_limit.limit": limit, + "rate_limit.check.timeout_ms": timeoutMs, + }, + }), ); }); @@ -320,17 +443,53 @@ export const makeCloudExecutionRateLimiter = ( ); return makeExecutionRateLimiter(() => Effect.succeed(0)); } - return makeExecutionRateLimiter( - (organizationId, windowId) => - Effect.tryPromise(() => { + return makeExecutionRateLimiter(counterIncrement(namespace), { + limit, + timeoutMs: resolveCheckTimeoutMs(), + isExempt, + }); +}; + +/** + * The counter DO RPC, as a traced and typed increment. + * + * The span is the whole point: this call is a blocking, cold-startable hop on + * the execute hot path, and until it had one nothing about its duration was + * measurable — the only evidence it was slow was the fail-open warning 2s + * later. The typed error replaces Effect's generic `UnknownError`, which + * reported no org, no window, and no hint that a Durable Object was involved. + */ +const counterIncrement = + (namespace: RateLimiterNamespace): RateLimitIncrement => + (organizationId, windowId) => + Effect.tryPromise({ + try: () => { const stub = namespace.get( namespace.idFromName(organizationId), ) as ExecutionRateLimiterStub; return stub.increment(windowId); + }, + catch: (cause) => { + // oxlint-disable-next-line executor/no-instanceof-error, executor/no-unknown-error-message -- boundary: the Durable Object RPC rejects with a plain platform Error whose message text is the only classification signal Cloudflare gives + const reason = cause instanceof Error ? cause.message : String(cause); + return new RateLimitCounterError({ + organizationId, + code: counterErrorCode(reason), + reason, + cause, + }); + }, + }).pipe( + Effect.tapError((error) => + Effect.annotateCurrentSpan({ + "rate_limit.counter.error_tag": "RateLimitCounterError", + "rate_limit.counter.error_code": error.code, + }), + ), + Effect.withSpan("rate_limit.increment", { + attributes: { "rate_limit.window_id": windowId }, }), - { limit, isExempt }, - ); -}; + ); /** * The per-org hourly cap: the `EXECUTION_RATE_LIMIT_PER_HOUR` env override @@ -344,3 +503,18 @@ const resolveRateLimit = (): number => { const parsed = Number.parseInt(raw, 10); return Number.isInteger(parsed) && parsed > 0 ? parsed : EXECUTIONS_PER_ORG_PER_HOUR; }; + +/** + * The counter's time budget: the `EXECUTION_RATE_LIMIT_CHECK_TIMEOUT_MS` env + * override or `RATE_LIMIT_CHECK_TIMEOUT_MS` when it's unset or unparseable. + * Same precedent and same purpose as `EXECUTION_RATE_LIMIT_PER_HOUR`: the + * production 2s budget can't be blown on demand, so tests set a tiny one to + * drive the fail-open path deterministically. Production leaves it unset. + */ +const resolveCheckTimeoutMs = (): number => { + const raw = (env as { EXECUTION_RATE_LIMIT_CHECK_TIMEOUT_MS?: string }) + .EXECUTION_RATE_LIMIT_CHECK_TIMEOUT_MS; + if (raw === undefined) return RATE_LIMIT_CHECK_TIMEOUT_MS; + const parsed = Number.parseInt(raw, 10); + return Number.isInteger(parsed) && parsed > 0 ? parsed : RATE_LIMIT_CHECK_TIMEOUT_MS; +}; diff --git a/apps/cloud/src/env-augment.d.ts b/apps/cloud/src/env-augment.d.ts index ef0b54da4..bb31aa00a 100644 --- a/apps/cloud/src/env-augment.d.ts +++ b/apps/cloud/src/env-augment.d.ts @@ -60,6 +60,13 @@ declare global { // number to drive the backstop. Production leaves it unset. EXECUTION_RATE_LIMIT_PER_HOUR?: string; + // Optional override for the counter DO's check budget in milliseconds + // (defaults to RATE_LIMIT_CHECK_TIMEOUT_MS = 2000 when unset or + // unparseable). Same purpose as the cap override: the production budget + // can't be blown on demand, so tests set a tiny one to exercise the + // fail-open path. Production leaves it unset. + EXECUTION_RATE_LIMIT_CHECK_TIMEOUT_MS?: string; + // First-party OAuth apps (executor-owned provider registrations). Each // pair enables one-click connect through `first-party:`; an // unset pair simply ships no first-party app for that provider. The diff --git a/e2e/cloud/mcp-execution-limits.test.ts b/e2e/cloud/mcp-execution-limits.test.ts index 4c3272267..40bbb290d 100644 --- a/e2e/cloud/mcp-execution-limits.test.ts +++ b/e2e/cloud/mcp-execution-limits.test.ts @@ -21,8 +21,11 @@ import { RATE_LIMIT_BLOCKED_MESSAGE, } from "../../apps/cloud/src/engine/execution-limit-messages"; import { scenario } from "../src/scenario"; -import { Autumn, Billing, Mcp, Target } from "../src/services"; -import { E2E_EXECUTION_RATE_LIMIT } from "../setup/execution-limits"; +import { Autumn, Billing, Mcp, Target, Telemetry } from "../src/services"; +import { + E2E_EXECUTION_RATE_LIMIT, + E2E_EXECUTION_RATE_LIMIT_CHECK_TIMEOUT_MS, +} from "../setup/execution-limits"; import type { Identity } from "../src/target"; const emailOf = (identity: Identity): string => identity.credentials?.email ?? identity.label; @@ -162,6 +165,7 @@ scenario( Effect.gen(function* () { yield* Billing; const autumn = yield* Autumn; + const telemetry = yield* Telemetry; const target = yield* Target; const mcp = yield* Mcp; @@ -196,6 +200,109 @@ scenario( expect(metered.length, "only the allowed executions are metered — the blocked one is not").toBe( RATE_LIMIT, ); + + // The block is also legible in telemetry, not just to the client. This is + // the OTHER branch of the check — the allowed path is pinned by the span + // scenario below — and it is the one an operator reaches for when a + // customer reports being cut off mid-workload. + const blockedCheck = yield* telemetry.expectSpan({ + operation: "rate_limit.check", + attributes: { "rate_limit.organization_id": customerId, "rate_limit.blocked": "true" }, + }); + expect( + blockedCheck.span.tags["rate_limit.count"], + "the span carries the count that crossed the cap", + ).toBe(String(RATE_LIMIT + 1)); + expect( + blockedCheck.span.tags["rate_limit.exempt"], + "the org was not exempt — that is why it was blocked", + ).toBe("false"); + expect( + blockedCheck.span.tags["rate_limit.check.failed_open"], + "a real block is not a degraded check wearing a block's clothes", + ).toBe("false"); + }), +); + +scenario( + "Billing · the rate-limit counter check is visible in the exported spans", + { timeout: 180_000 }, + Effect.gen(function* () { + // The backstop's counter is a blocking Durable Object hop on the execute + // hot path. It used to run untraced, so the only production evidence it + // was slow was a fail-open warning arriving 2s later — and a counter fault + // reached the error reporter as an untyped wrapper with no application + // frames. This pins the measurement seam where it is actually read: the + // spans the worker EXPORTS, over the real workerd + Durable Object + // topology, not a span object built in a unit test. + yield* Billing; + const telemetry = yield* Telemetry; + const target = yield* Target; + const mcp = yield* Mcp; + + const identity = yield* target.newIdentity(); + const bearer = yield* mcp.mintBearer(emailOf(identity)); + const customerId = orgIdOf(bearer); + + const session = mcp.session(identity); + const ok = yield* session.call("execute", { code: "return 4 + 5;" }); + expect(ok.ok, "the execution runs").toBe(true); + expect(ok.text, "it returns its value").toContain("9"); + + // A second RPC on the same session, deliberately NOT an execute (it must + // not move the counter this scenario asserts on). The session DO ships its + // spans on a bounded, best-effort `waitUntil` flush after each RPC — a + // session that makes exactly one call in its whole life is asserting on + // that one flush winning a race, which it loses often enough on a loaded + // machine to be useless. A real client keeps talking to its session; so + // does this one. + yield* session.listTools(); + + const check = yield* telemetry.expectSpan({ + operation: "rate_limit.check", + attributes: { "rate_limit.organization_id": customerId }, + }); + expect( + check.span.tags["rate_limit.count"], + "the org's count for the window is on the span", + ).toBe("1"); + expect(check.span.tags["rate_limit.limit"], "so is the cap it was measured against").toBe( + String(RATE_LIMIT), + ); + expect(check.span.tags["rate_limit.blocked"], "an allowed execution is recorded as such").toBe( + "false", + ); + // The two attributes production alerting reads. Both false on a healthy + // check; a step change in either is the signal that used to arrive only as + // a stream of untyped error reports. + expect(check.span.tags["rate_limit.check.failed_open"], "the check did not fail open").toBe( + "false", + ); + // The budget on the span is the boot's OVERRIDE, not the compiled-in 2000ms + // default — the only end-to-end proof that + // EXECUTION_RATE_LIMIT_CHECK_TIMEOUT_MS actually reaches the worker and + // retunes the check. A knob that silently falls back to the constant would + // read as healthy here and as a working knob everywhere else. + expect( + check.span.tags["rate_limit.check.timeout_ms"], + "the budget on the span is the one the worker was booted with", + ).toBe(String(E2E_EXECUTION_RATE_LIMIT_CHECK_TIMEOUT_MS)); + + // The DO round trip itself is timed, inside the same trace as the check. + // Polled, not read once: the parent and the child ride different export + // batches, so a one-shot read here races the exporter and fails ~1 run in 5. + const increment = yield* telemetry.expectSpan({ + operation: "rate_limit.increment", + traceId: check.traceId, + }); + expect( + increment.span.tags["rate_limit.window_id"], + "the counter span is for the window the check decided against", + ).toBe(check.span.tags["rate_limit.window_id"]); + expect( + increment.span.tags["rate_limit.counter.error_code"], + "a healthy counter call carries no fault classification", + ).toBeUndefined(); }), ); @@ -209,6 +316,7 @@ scenario( // blocks a free org (the scenario above) must sail past the cap here. yield* Billing; const autumn = yield* Autumn; + const telemetry = yield* Telemetry; const target = yield* Target; const mcp = yield* Mcp; @@ -245,5 +353,25 @@ scenario( count: overCap, }); expect(metered.length, "every execution past the cap is still metered").toBe(overCap); + + // Third branch of the check: over the cap AND allowed. Without the reason + // on the span, this is indistinguishable in production from a check that + // never ran — which is exactly the ambiguity that made the 2026-08-18 + // block hard to explain. + const exemptCheck = yield* telemetry.expectSpan({ + operation: "rate_limit.check", + attributes: { "rate_limit.organization_id": customerId, "rate_limit.exempt": "true" }, + }); + expect( + Number(exemptCheck.span.tags["rate_limit.count"]), + "the exemption was recorded on a check that was genuinely over the cap", + ).toBeGreaterThan(RATE_LIMIT); + expect(exemptCheck.span.tags["rate_limit.blocked"], "and it allowed the execution").toBe( + "false", + ); + expect( + exemptCheck.span.tags["rate_limit.check.failed_open"], + "the org ran because it is exempt, not because the counter fell over", + ).toBe("false"); }), ); diff --git a/e2e/setup/cloud.boot.ts b/e2e/setup/cloud.boot.ts index 851ab85cc..4c24cd3ed 100644 --- a/e2e/setup/cloud.boot.ts +++ b/e2e/setup/cloud.boot.ts @@ -12,7 +12,10 @@ import { createEmulator } from "@executor-js/emulate"; import { bootProcesses, waitForBoot, waitForHttp } from "./boot"; import { AUTUMN_PLAN_SEED } from "./autumn-plans"; -import { E2E_EXECUTION_RATE_LIMIT } from "./execution-limits"; +import { + E2E_EXECUTION_RATE_LIMIT, + E2E_EXECUTION_RATE_LIMIT_CHECK_TIMEOUT_MS, +} from "./execution-limits"; export const cloudDir = fileURLToPath(new URL("../../apps/cloud/", import.meta.url)); @@ -126,6 +129,11 @@ export const bootCloud = async (options: CloudBootOptions): Promise // scenario's per-org execute count. Reaches the worker via // CLOUDFLARE_INCLUDE_PROCESS_ENV, same as ALLOW_LOCAL_NETWORK. EXECUTION_RATE_LIMIT_PER_HOUR: String(E2E_EXECUTION_RATE_LIMIT), + // Set to a value that is NOT the compiled-in default, so the span the + // worker exports proves this override was actually read rather than + // silently ignored (execution-limits.ts explains why it is longer, not + // shorter, than the default). + EXECUTION_RATE_LIMIT_CHECK_TIMEOUT_MS: String(E2E_EXECUTION_RATE_LIMIT_CHECK_TIMEOUT_MS), // Throwaway PGlite on its own port + dir so it never fights `bun dev`. DEV_DB_PORT: String(options.dbPort), DEV_DB_PATH: dbPath, diff --git a/e2e/setup/execution-limits.ts b/e2e/setup/execution-limits.ts index 0294644d6..16395523d 100644 --- a/e2e/setup/execution-limits.ts +++ b/e2e/setup/execution-limits.ts @@ -12,3 +12,18 @@ // comfortable headroom. If a scenario ever fails with the rate-limit backstop // message, it outgrew this cap: raise it here, never in the boot env alone. export const E2E_EXECUTION_RATE_LIMIT = 20; + +// The e2e worker's counter-check budget (EXECUTION_RATE_LIMIT_CHECK_TIMEOUT_MS), +// same two consumers as the cap above. +// +// This exists to make the OVERRIDE ITSELF observable end to end. The budget is +// stamped on the `rate_limit.check` span, so a value that differs from the +// compiled-in default (RATE_LIMIT_CHECK_TIMEOUT_MS = 2000) is the difference +// between "the worker read the env var" and "the worker fell back to the +// constant and nobody noticed". A knob nothing reads is worse than no knob. +// +// It is deliberately LONGER than the default, not shorter: a shorter budget +// would time the check out and fail open on every execution, which would take +// the backstop scenario down with it. Longer only means a genuinely wedged +// counter stalls an execution 3s instead of 2s, which no scenario depends on. +export const E2E_EXECUTION_RATE_LIMIT_CHECK_TIMEOUT_MS = 3000;