From bc9a7543df8408aee51842a03a623df75e3b5b56 Mon Sep 17 00:00:00 2001 From: Owen Greenhalgh Date: Wed, 10 Jun 2026 18:02:34 +0100 Subject: [PATCH] per-org aggregate caps with failopen on check/track --- .../honoMiddlewares/rateLimitMiddleware.ts | 28 ++++++++- server/src/honoUtils/HonoEnv.ts | 4 ++ .../balances/check/runCheckWithRollout.ts | 12 ++++ .../balances/track/runTrackWithRollout.ts | 5 ++ .../misc/rateLimiter/rateLimitConfigs.ts | 59 +++++++++++++++++++ .../misc/rateLimiter/rateLimitFactory.ts | 32 +++++++++- .../rate-limits/org-aggregate-config.test.ts | 54 +++++++++++++++++ 7 files changed, 190 insertions(+), 4 deletions(-) create mode 100644 server/tests/unit/rate-limits/org-aggregate-config.test.ts diff --git a/server/src/honoMiddlewares/rateLimitMiddleware.ts b/server/src/honoMiddlewares/rateLimitMiddleware.ts index b3e1ff4cf..6e8c85cb0 100644 --- a/server/src/honoMiddlewares/rateLimitMiddleware.ts +++ b/server/src/honoMiddlewares/rateLimitMiddleware.ts @@ -6,6 +6,7 @@ import { setRateLimitKeyInContext, } from "@/internal/misc/rateLimiter/rateLimitFactory"; import { + getOrgAggregateType, getRateLimitType, RateLimitType, } from "../internal/misc/rateLimiter/rateLimitConfigs"; @@ -39,8 +40,31 @@ export const rateLimitMiddleware = async (c: Context, next: Next) => { // 4. Get the appropriate limiter for this type const limiter = getLimiterForType(rateLimitType); - // 5. Apply rate limiting - return await limiter(c as Context, next); + const aggregateType = getOrgAggregateType(rateLimitType); + if (!aggregateType) { + // 5. Apply rate limiting + return await limiter(c as Context, next); + } + + // 5. Org-aggregate limiter wraps the per-customer one; the key slot is + // swapped between them since keyGenerator reads it at execution time. + setRateLimitKeyInContext( + c as Context, + getRateLimitKey({ c, rateLimitType: aggregateType }), + ); + const aggregateLimiter = getLimiterForType(aggregateType); + + let innerResponse: Response | undefined; + const aggregateResponse = await aggregateLimiter( + c as Context, + async () => { + setRateLimitKeyInContext(c as Context, rateLimitKey); + innerResponse = (await limiter(c as Context, next)) ?? undefined; + }, + ); + + // hono-rate-limiter discards next()'s return, so re-surface an inner 429. + return aggregateResponse ?? innerResponse; } catch (error) { ctx.logger.error( `Error checking rate limit, error: ${error}. Bypassing for now`, diff --git a/server/src/honoUtils/HonoEnv.ts b/server/src/honoUtils/HonoEnv.ts index c7f928183..3fbefb7b5 100644 --- a/server/src/honoUtils/HonoEnv.ts +++ b/server/src/honoUtils/HonoEnv.ts @@ -74,6 +74,10 @@ export type RequestContext = { fullCustomer?: FullCustomer; rolloutSnapshot?: RolloutSnapshot; + /** Org is over its aggregate rate cap — check/track flows skip the DB and + * serve their fail-open responses (allow / SQS queue) instead. */ + orgRateLimitDegraded?: boolean; + testOptions?: { skipCacheDeletion?: boolean; skipWebhooks?: boolean; diff --git a/server/src/internal/balances/check/runCheckWithRollout.ts b/server/src/internal/balances/check/runCheckWithRollout.ts index eda61ab2f..17cd689b0 100644 --- a/server/src/internal/balances/check/runCheckWithRollout.ts +++ b/server/src/internal/balances/check/runCheckWithRollout.ts @@ -18,6 +18,18 @@ export const runCheckWithRollout = async ({ body: ParsedCheckParams; requiredBalance: number; }): Promise> => { + if (ctx.orgRateLimitDegraded) { + return { + checkData: null, + response: getCheckFailOpenFallback({ + ctx, + body, + requiredBalance, + error: new Error("org aggregate rate cap exceeded"), + }) as Record, + }; + } + if (!isFullSubjectRolloutEnabled({ ctx })) { return runCheckLegacyFlow({ ctx, body, requiredBalance }); } diff --git a/server/src/internal/balances/track/runTrackWithRollout.ts b/server/src/internal/balances/track/runTrackWithRollout.ts index 23e88a116..622590075 100644 --- a/server/src/internal/balances/track/runTrackWithRollout.ts +++ b/server/src/internal/balances/track/runTrackWithRollout.ts @@ -24,6 +24,11 @@ export const runTrackWithRollout = async ({ apiVersion?: ApiVersion; }): Promise => { if (shouldUseTrackV3({ ctx })) { + if (ctx.orgRateLimitDegraded) { + const queuedResponse = await queueTrack({ ctx, body }); + if (queuedResponse) return queuedResponse; + } + return withRedisFailOpen({ source: "runTrackWithRollout", run: () => diff --git a/server/src/internal/misc/rateLimiter/rateLimitConfigs.ts b/server/src/internal/misc/rateLimiter/rateLimitConfigs.ts index 52a5f9cf1..821e6f2dd 100644 --- a/server/src/internal/misc/rateLimiter/rateLimitConfigs.ts +++ b/server/src/internal/misc/rateLimiter/rateLimitConfigs.ts @@ -13,8 +13,23 @@ export enum RateLimitType { ListCustomers = "list_customers", CustomerEntitiesGet = "customer_entities_get", Logs = "logs", + TrackOrg = "track_org", + CheckOrg = "check_org", + EntitiesGetOrg = "entities_get_org", } +// Org-wide aggregate caps summed across all of an org's customers — the +// per-customer limits never bind for many-customer storms (2026-06-08 incident). +const ORG_AGGREGATE_TYPES: Partial> = { + [RateLimitType.Track]: RateLimitType.TrackOrg, + [RateLimitType.Check]: RateLimitType.CheckOrg, + [RateLimitType.CustomerEntitiesGet]: RateLimitType.EntitiesGetOrg, +}; + +export const getOrgAggregateType = ( + type: RateLimitType, +): RateLimitType | undefined => ORG_AGGREGATE_TYPES[type]; + type RoutePattern = { method: string; url: string; @@ -120,6 +135,22 @@ const RATE_LIMIT_ROUTE_GROUPS: RateLimitRouteGroup[] = [ }, ]; +// Check-group routes that can fail open (allowed: true) when an org is over +// its aggregate cap; the establish routes in the group shed a 503 instead. +const CHECK_FAIL_OPEN_PATTERNS: RoutePattern[] = [ + route({ method: "POST", url: "/v1/check" }), + route({ method: "POST", url: "/v1/entitled" }), + route({ method: "POST", url: "/v1/balances.check" }), +]; + +export const isCheckFailOpenRoute = (c: Context): boolean => { + const method = c.req.method; + const path = c.req.path; + return CHECK_FAIL_OPEN_PATTERNS.some((pattern) => + matchRoute({ url: path, method, pattern }), + ); +}; + export const getRateLimitType = (c: Context) => { const method = c.req.method; const path = c.req.path; @@ -154,6 +185,9 @@ export type RateLimitConfig = { windowMs: number; notInRedis: boolean; scope: RateLimitScope; + // "degrade" = over-limit requests fail open (check -> allow, track -> SQS + // queue) instead of 429, so the cap sheds DB load without losing events. + overLimit?: "reject" | "degrade"; }; export const resolveRateLimit = ({ @@ -255,4 +289,29 @@ export const RATE_LIMIT_CONFIGS: Record = { notInRedis: false, scope: RateLimitScope.Org, }, + // 60s windows sized ~1.5-2x the highest legit per-org peak observed over 7d + // of prod traffic (check 157k/min, track 60k/min, entities.get 53k/min). + [RateLimitType.TrackOrg]: { + name: "track_org", + limit: 120_000, + windowMs: 60_000, + notInRedis: false, + scope: RateLimitScope.Org, + overLimit: "degrade", + }, + [RateLimitType.CheckOrg]: { + name: "check_org", + limit: 240_000, + windowMs: 60_000, + notInRedis: false, + scope: RateLimitScope.Org, + overLimit: "degrade", + }, + [RateLimitType.EntitiesGetOrg]: { + name: "entities_get_org", + limit: 90_000, + windowMs: 60_000, + notInRedis: false, + scope: RateLimitScope.Org, + }, }; diff --git a/server/src/internal/misc/rateLimiter/rateLimitFactory.ts b/server/src/internal/misc/rateLimiter/rateLimitFactory.ts index 7ae7fe61a..5371fa09f 100644 --- a/server/src/internal/misc/rateLimiter/rateLimitFactory.ts +++ b/server/src/internal/misc/rateLimiter/rateLimitFactory.ts @@ -1,14 +1,15 @@ import type { ApiVersion } from "@autumn/shared"; -import type { Context } from "hono"; +import type { Context, Next } from "hono"; import { rateLimiter } from "hono-rate-limiter"; import { logger } from "@/external/logtail/logtailUtils.js"; import { shouldUseRedis } from "@/external/redis/initRedis"; import type { HonoEnv } from "@/honoUtils/HonoEnv"; import { + isCheckFailOpenRoute, RATE_LIMIT_CONFIGS, type RateLimitConfig, RateLimitScope, - type RateLimitType, + RateLimitType, resolveRateLimit, } from "./rateLimitConfigs"; import { getOrgRateLimitOverride } from "./rateLimitOverridesStore"; @@ -60,11 +61,38 @@ export const rateLimitFactory = ({ return resolveRateLimit({ config, apiVersion }).limit; }; + // Over-limit "degrade": fail open instead of 429 — check routes get the + // allow-fallback via the ctx flag; establish routes shed a retryable 503. + const degradeHandler = async ( + c: Context, + next: Next, + ): Promise => { + const honoContext = c as Context; + const ctx = honoContext.get("ctx"); + + if (type === RateLimitType.CheckOrg && !isCheckFailOpenRoute(honoContext)) { + return c.json( + { + message: "Service is temporarily unavailable, please retry shortly.", + code: "service_unavailable", + env: ctx?.env, + }, + 503, + ); + } + + if (ctx) ctx.orgRateLimitDegraded = true; + c.header("Retry-After", undefined); + await next(); + return; + }; + const options = { windowMs, limit: dynamicLimit, standardHeaders: "draft-6" as const, keyGenerator: getRateLimitKeyFromContext, + ...(config.overLimit === "degrade" && { handler: degradeHandler }), }; let inMemoryLimiter: ReturnType | null = null; diff --git a/server/tests/unit/rate-limits/org-aggregate-config.test.ts b/server/tests/unit/rate-limits/org-aggregate-config.test.ts new file mode 100644 index 000000000..922739dc9 --- /dev/null +++ b/server/tests/unit/rate-limits/org-aggregate-config.test.ts @@ -0,0 +1,54 @@ +import { describe, expect, test } from "bun:test"; +import { + getOrgAggregateType, + RATE_LIMIT_CONFIGS, + RateLimitScope, + RateLimitType, +} from "@/internal/misc/rateLimiter/rateLimitConfigs.js"; + +describe("org aggregate rate limits", () => { + test("high-volume per-customer types map to an org aggregate", () => { + expect(getOrgAggregateType(RateLimitType.Track)).toBe( + RateLimitType.TrackOrg, + ); + expect(getOrgAggregateType(RateLimitType.Check)).toBe( + RateLimitType.CheckOrg, + ); + expect(getOrgAggregateType(RateLimitType.CustomerEntitiesGet)).toBe( + RateLimitType.EntitiesGetOrg, + ); + }); + + test("types without an aggregate return undefined", () => { + expect(getOrgAggregateType(RateLimitType.General)).toBeUndefined(); + expect(getOrgAggregateType(RateLimitType.Attach)).toBeUndefined(); + expect(getOrgAggregateType(RateLimitType.TrackOrg)).toBeUndefined(); + }); + + test("aggregate configs are org-scoped, redis-backed, 60s windows", () => { + const aggregates = [ + RateLimitType.TrackOrg, + RateLimitType.CheckOrg, + RateLimitType.EntitiesGetOrg, + ]; + for (const type of aggregates) { + const config = RATE_LIMIT_CONFIGS[type]; + expect(config.scope).toBe(RateLimitScope.Org); + expect(config.notInRedis).toBe(false); + expect(config.windowMs).toBe(60_000); + expect(config.limit).toBeGreaterThan(0); + } + }); + + test("check/track aggregates degrade (fail open) instead of rejecting", () => { + expect(RATE_LIMIT_CONFIGS[RateLimitType.CheckOrg].overLimit).toBe( + "degrade", + ); + expect(RATE_LIMIT_CONFIGS[RateLimitType.TrackOrg].overLimit).toBe( + "degrade", + ); + expect( + RATE_LIMIT_CONFIGS[RateLimitType.EntitiesGetOrg].overLimit, + ).toBeUndefined(); + }); +});