Skip to content

Commit 7354d00

Browse files
rkm1claude
andcommitted
fix(db): bound the Lambda connection pool and add a DB circuit breaker
The Lambda DB path handed PrismaPg a connection string with ?connection_limit=1, which @prisma/adapter-pg ignores, so each warm execution environment's pool used the pg default max:10. Under a burst of signups (or a pre-token cache-miss storm) the Lambda fleet could exhaust RDS max_connections and make Postgres reject connections from every client. Fix: construct an explicit pg.Pool({max:1, connectionTimeoutMillis, idleTimeoutMillis}) and pass it to PrismaPg so per-Lambda connection count is deterministic (env knobs LAMBDA_DATABASE_POOL_MAX / LAMBDA_DATABASE_CONNECT_TIMEOUT_MS); add a LAMBDA_DATABASE_PROXY_HOST override to route through an RDS Proxy when provisioned; add a per-environment circuit breaker (withLambdaDbBreaker over DatabaseCircuitBreaker) wired into pre-token (loadFromRds) and post-confirmation (provisioning transaction) so a saturated DB fails fast instead of being retried; and emit a pre-token cache_miss event for a miss-rate metric. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
1 parent 99dafe1 commit 7354d00

4 files changed

Lines changed: 264 additions & 23 deletions

File tree

apps/api/src/lambda/post-confirmation.ts

Lines changed: 25 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,7 @@ import type {
3434
PostConfirmationTriggerHandler,
3535
} from "aws-lambda";
3636
import { Logger } from "@aws-lambda-powertools/logger";
37-
import { getLambdaPrisma as getPrisma } from "../lib/lambda-prisma.js";
37+
import { getLambdaPrisma as getPrisma, withLambdaDbBreaker } from "../lib/lambda-prisma.js";
3838
import {
3939
PrismaClient,
4040
Prisma,
@@ -182,22 +182,30 @@ export const handler: PostConfirmationTriggerHandler = async (event) => {
182182

183183
const db = await getPrisma();
184184

185-
const result = await withHandleConflictRetry(() =>
186-
db.$transaction(
187-
async (tx) => provisionUserAndTenancy(tx, {
188-
cognitoSub,
189-
email,
190-
emailVerified: attrs.email_verified,
191-
federated,
192-
idpGroups,
193-
dateOfBirth,
194-
ageTier,
195-
providedHandle: attrs["custom:handle"],
196-
invitationCode,
197-
requestedMethod,
198-
}),
199-
{ timeout: 8000 },
200-
),
185+
// The provisioning transaction is the longest-held connection in a signup
186+
// burst (multi-statement, up to the 8s timeout). Run it under the Lambda
187+
// circuit breaker so a saturated DB trips fast-fail instead of being retried
188+
// into an already-exhausted instance.
189+
const result = await withLambdaDbBreaker(
190+
() =>
191+
withHandleConflictRetry(() =>
192+
db.$transaction(
193+
async (tx) => provisionUserAndTenancy(tx, {
194+
cognitoSub,
195+
email,
196+
emailVerified: attrs.email_verified,
197+
federated,
198+
idpGroups,
199+
dateOfBirth,
200+
ageTier,
201+
providedHandle: attrs["custom:handle"],
202+
invitationCode,
203+
requestedMethod,
204+
}),
205+
{ timeout: 8000 },
206+
),
207+
),
208+
"post_confirmation.provision",
201209
);
202210

203211
if (ageTier === "CHILD") {

apps/api/src/lambda/pre-token-generation.ts

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@ import type {
2525
} from "aws-lambda";
2626
import { Logger } from "@aws-lambda-powertools/logger";
2727
import { PrismaClient, type TenantRole } from "@prisma/client";
28-
import { getLambdaPrisma as getPrisma } from "../lib/lambda-prisma.js";
28+
import { getLambdaPrisma as getPrisma, withLambdaDbBreaker } from "../lib/lambda-prisma.js";
2929
import {
3030
ClaimsCache,
3131
createClaimsCacheFromEnv,
@@ -194,6 +194,11 @@ export const handler: PreTokenGenerationV2TriggerHandler = async (event) => {
194194
let cacheHit = !!claims;
195195

196196
if (!claims) {
197+
// RDS is consulted only on a genuine cache miss. Emit a filterable event so
198+
// a miss-rate metric can be derived (a miss storm — post-deploy, correlated
199+
// TTL expiry, or the first-login wave after a signup burst — is the path
200+
// that can exhaust DB connections; the warm cache is the primary defence).
201+
logger.info("pretoken.cache_miss", { cognitoSub, federated });
197202
const db = await getPrisma();
198203
// Read the user's last explicit tenant preference, even from an expired
199204
// cache row, so an admin-side switch-tenant call survives cache TTL.
@@ -206,7 +211,10 @@ export const handler: PreTokenGenerationV2TriggerHandler = async (event) => {
206211
error: (err as { code?: string })?.code ?? "unknown",
207212
});
208213
}
209-
const loaded = await loadFromRds(db, cognitoSub, federated, preferredTenantId);
214+
const loaded = await withLambdaDbBreaker(
215+
() => loadFromRds(db, cognitoSub, federated, preferredTenantId),
216+
"pretoken.load_from_rds",
217+
);
210218

211219
if (!loaded.user) {
212220
logger.warn("pretoken.drift", { cognitoSub });

apps/api/src/lib/lambda-prisma.ts

Lines changed: 92 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,9 @@
1+
import { Logger } from "@aws-lambda-powertools/logger";
12
import { getSecret } from "@aws-lambda-powertools/parameters/secrets";
23
import { PrismaPg } from "@prisma/adapter-pg";
34
import { PrismaClient } from "@prisma/client";
5+
import { Pool } from "pg";
6+
import { DatabaseCircuitBreaker } from "./database-circuit-breaker.js";
47

58
interface DbSecret {
69
username: string;
@@ -10,7 +13,73 @@ interface DbSecret {
1013
dbname: string;
1114
}
1215

16+
const logger = new Logger({ serviceName: "lambda-prisma" });
17+
18+
/**
19+
* Operational knobs — env-configurable with safe defaults (threshold-secrecy
20+
* rule: never compile operational limits into the public tarball).
21+
*
22+
* Why the pool MUST be capped here: each warm Lambda execution environment holds
23+
* its OWN pg pool, and every concurrent environment shares the RDS
24+
* `max_connections` budget (~107 usable on a t4g.micro). Under a signup burst —
25+
* or a `pre-token` cache-miss storm — an unbounded pool lets a single function
26+
* exhaust the instance, and Postgres then rejects connections from EVERY client
27+
* (a global outage, not just failed signups). `?connection_limit=1` in the
28+
* connection URL is a **no-op** under `@prisma/adapter-pg` (it is a Prisma-engine
29+
* parameter the `pg` driver ignores), so the cap is set on the pool object — the
30+
* only place it takes effect. See
31+
* trellis-internal `analysis/db-connection-management/signup-burst-connection-exhaustion.md`.
32+
*/
33+
const DEFAULT_POOL_MAX = 1;
34+
const DEFAULT_CONNECT_TIMEOUT_MS = 2000;
35+
const DEFAULT_IDLE_TIMEOUT_MS = 10_000;
36+
const DEFAULT_BREAKER_THRESHOLD = 5;
37+
const DEFAULT_BREAKER_COOLDOWN_MS = 30_000;
38+
1339
let prisma: PrismaClient | null = null;
40+
let pool: Pool | null = null;
41+
42+
/**
43+
* Per-execution-environment circuit breaker for the Lambda DB path.
44+
*
45+
* The request path (`DatabaseConnectionManager`) has a breaker; Lambda handlers
46+
* did not — so under DB saturation they would keep retrying into a saturated
47+
* instance and amplify the incident. Opening the breaker makes a saturated
48+
* environment fail fast (no connect-timeout wait, no slot held) until a cooldown
49+
* probe succeeds. Module scope = one warm Lambda environment, the right blast
50+
* radius. Wrap every RDS access in a handler with `withLambdaDbBreaker`.
51+
*/
52+
export const lambdaDbBreaker = new DatabaseCircuitBreaker({
53+
failureThreshold: Number(
54+
process.env.LAMBDA_DATABASE_BREAKER_THRESHOLD ?? DEFAULT_BREAKER_THRESHOLD,
55+
),
56+
cooldownMs: Number(
57+
process.env.LAMBDA_DATABASE_BREAKER_COOLDOWN_MS ?? DEFAULT_BREAKER_COOLDOWN_MS,
58+
),
59+
});
60+
61+
/**
62+
* Run a unit of DB work under the Lambda circuit breaker. Use around every RDS
63+
* access in a Lambda handler so connection-exhaustion failures trip the breaker
64+
* instead of being retried into a saturated instance. When the breaker is OPEN
65+
* the call throws immediately (message begins "Circuit breaker is OPEN").
66+
*/
67+
export async function withLambdaDbBreaker<T>(
68+
fn: () => Promise<T>,
69+
operation?: string,
70+
): Promise<T> {
71+
try {
72+
return await lambdaDbBreaker.execute(fn, { operation });
73+
} catch (err) {
74+
if (
75+
err instanceof Error &&
76+
err.message.startsWith("Circuit breaker is OPEN")
77+
) {
78+
logger.warn("lambda_db.breaker_open", { operation });
79+
}
80+
throw err;
81+
}
82+
}
1483

1584
/**
1685
* Build (and cache) a PrismaClient for standalone Lambda handlers.
@@ -21,20 +90,39 @@ let prisma: PrismaClient | null = null;
2190
* `DatabaseConnectionManager` (`ssl: { rejectUnauthorized: false }`); Lambda
2291
* handlers must do the same. Prisma 7 supplies the connection through a pg
2392
* driver adapter (the old `datasources` constructor option is gone), so the
24-
* `ssl` option goes on the adapter's pool config.
93+
* `ssl` option and the size cap go on the pool config.
94+
*
95+
* The pool is passed to `PrismaPg` as an explicit `pg.Pool` (NOT a
96+
* connection-string config) because the pool is the only place `max` takes
97+
* effect — see the knobs comment above.
2598
*
26-
* Cached at module scope so warm invocations reuse the client.
99+
* Cached at module scope so warm invocations reuse the client (and its single
100+
* connection).
27101
*/
28102
export async function getLambdaPrisma(): Promise<PrismaClient> {
29103
if (prisma) return prisma;
30104
const { username, password, host, port, dbname } = (await getSecret(
31105
process.env.DB_SECRET_ARN!,
32106
{ transform: "json" },
33107
)) as unknown as DbSecret;
34-
const adapter = new PrismaPg({
35-
connectionString: `postgresql://${username}:${encodeURIComponent(password)}@${host}:${port}/${dbname}?connection_limit=1`,
108+
109+
// When an RDS Proxy is provisioned, the infra injects its endpoint here so the
110+
// Lambda connects through the proxy (which multiplexes and caps the
111+
// connections the DB ever sees); fall back to the direct instance endpoint
112+
// when unset, so the default deployment is unchanged.
113+
const dbHost = process.env.LAMBDA_DATABASE_PROXY_HOST || host;
114+
115+
pool = new Pool({
116+
connectionString: `postgresql://${username}:${encodeURIComponent(password)}@${dbHost}:${port}/${dbname}`,
36117
ssl: { rejectUnauthorized: false },
118+
max: Number(process.env.LAMBDA_DATABASE_POOL_MAX ?? DEFAULT_POOL_MAX),
119+
connectionTimeoutMillis: Number(
120+
process.env.LAMBDA_DATABASE_CONNECT_TIMEOUT_MS ?? DEFAULT_CONNECT_TIMEOUT_MS,
121+
),
122+
idleTimeoutMillis: DEFAULT_IDLE_TIMEOUT_MS,
123+
allowExitOnIdle: false,
37124
});
125+
const adapter = new PrismaPg(pool);
38126
prisma = new PrismaClient({ adapter });
39127
return prisma;
40128
}
Lines changed: 137 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,137 @@
1+
/**
2+
* Unit Tests: Lambda Prisma helper (connection-exhaustion hardening)
3+
*
4+
* The Lambda DB path must cap each warm execution environment to a single
5+
* connection (so burst demand = concurrency, not concurrency × pg-default-10),
6+
* fail fast on connect, optionally route through an RDS Proxy, and trip a
7+
* circuit breaker under sustained DB failure. See trellis-internal
8+
* analysis/db-connection-management/signup-burst-connection-exhaustion.md.
9+
*/
10+
11+
import { beforeEach, describe, expect, it, vi } from "vitest";
12+
13+
const { mockGetSecret, PoolMock, PrismaPgMock, PrismaClientMock } = vi.hoisted(
14+
() => ({
15+
mockGetSecret: vi.fn(),
16+
PoolMock: vi.fn(),
17+
PrismaPgMock: vi.fn(),
18+
PrismaClientMock: vi.fn(),
19+
}),
20+
);
21+
22+
vi.mock("pg", () => ({ Pool: PoolMock }));
23+
vi.mock("@prisma/adapter-pg", () => ({ PrismaPg: PrismaPgMock }));
24+
vi.mock("@prisma/client", () => ({ PrismaClient: PrismaClientMock }));
25+
vi.mock("@aws-lambda-powertools/parameters/secrets", () => ({
26+
getSecret: mockGetSecret,
27+
}));
28+
vi.mock("@aws-lambda-powertools/logger", () => ({
29+
Logger: class {
30+
info = vi.fn();
31+
warn = vi.fn();
32+
error = vi.fn();
33+
},
34+
}));
35+
36+
const SECRET = {
37+
username: "u",
38+
password: "p@ss/word",
39+
host: "db.internal",
40+
port: 5432,
41+
dbname: "app",
42+
};
43+
44+
const IMPORT = "../../../src/lib/lambda-prisma.js";
45+
46+
describe("lambda-prisma", () => {
47+
beforeEach(() => {
48+
vi.resetModules();
49+
vi.clearAllMocks();
50+
mockGetSecret.mockResolvedValue(SECRET);
51+
process.env.DB_SECRET_ARN = "arn:secret";
52+
delete process.env.LAMBDA_DATABASE_POOL_MAX;
53+
delete process.env.LAMBDA_DATABASE_CONNECT_TIMEOUT_MS;
54+
delete process.env.LAMBDA_DATABASE_PROXY_HOST;
55+
delete process.env.LAMBDA_DATABASE_BREAKER_THRESHOLD;
56+
delete process.env.LAMBDA_DATABASE_BREAKER_COOLDOWN_MS;
57+
});
58+
59+
it("caps the pool at max:1 by default with a fail-fast connect timeout", async () => {
60+
const { getLambdaPrisma } = await import(IMPORT);
61+
await getLambdaPrisma();
62+
63+
expect(PoolMock).toHaveBeenCalledTimes(1);
64+
const opts = PoolMock.mock.calls[0][0];
65+
expect(opts.max).toBe(1);
66+
expect(opts.connectionTimeoutMillis).toBe(2000);
67+
expect(opts.ssl).toEqual({ rejectUnauthorized: false });
68+
69+
// PrismaPg must receive the explicit Pool instance, NOT a connection-string
70+
// config (the only place `max` actually takes effect under @prisma/adapter-pg).
71+
expect(PrismaPgMock).toHaveBeenCalledTimes(1);
72+
expect(PrismaPgMock.mock.calls[0][0]).toBe(PoolMock.mock.instances[0]);
73+
});
74+
75+
it("honours pool-max and connect-timeout env overrides", async () => {
76+
process.env.LAMBDA_DATABASE_POOL_MAX = "3";
77+
process.env.LAMBDA_DATABASE_CONNECT_TIMEOUT_MS = "1500";
78+
const { getLambdaPrisma } = await import(IMPORT);
79+
await getLambdaPrisma();
80+
81+
const opts = PoolMock.mock.calls[0][0];
82+
expect(opts.max).toBe(3);
83+
expect(opts.connectionTimeoutMillis).toBe(1500);
84+
});
85+
86+
it("connects to the direct instance host when no proxy host is set", async () => {
87+
const { getLambdaPrisma } = await import(IMPORT);
88+
await getLambdaPrisma();
89+
expect(PoolMock.mock.calls[0][0].connectionString).toContain(
90+
"@db.internal:5432/app",
91+
);
92+
});
93+
94+
it("routes through LAMBDA_DATABASE_PROXY_HOST when set", async () => {
95+
process.env.LAMBDA_DATABASE_PROXY_HOST = "proxy.internal";
96+
const { getLambdaPrisma } = await import(IMPORT);
97+
await getLambdaPrisma();
98+
99+
const cs = PoolMock.mock.calls[0][0].connectionString;
100+
expect(cs).toContain("@proxy.internal:5432/app");
101+
expect(cs).not.toContain("@db.internal");
102+
});
103+
104+
it("caches the client across calls (one pool per warm environment)", async () => {
105+
const { getLambdaPrisma } = await import(IMPORT);
106+
await getLambdaPrisma();
107+
await getLambdaPrisma();
108+
expect(PoolMock).toHaveBeenCalledTimes(1);
109+
expect(mockGetSecret).toHaveBeenCalledTimes(1);
110+
});
111+
112+
describe("withLambdaDbBreaker", () => {
113+
it("passes results through on success", async () => {
114+
const { withLambdaDbBreaker } = await import(IMPORT);
115+
await expect(withLambdaDbBreaker(async () => "ok")).resolves.toBe("ok");
116+
});
117+
118+
it("opens after the threshold, then fails fast without invoking fn", async () => {
119+
process.env.LAMBDA_DATABASE_BREAKER_THRESHOLD = "3";
120+
const { withLambdaDbBreaker } = await import(IMPORT);
121+
122+
const boom = vi.fn(async () => {
123+
throw new Error("connect ETIMEDOUT");
124+
});
125+
for (let i = 0; i < 3; i++) {
126+
await expect(withLambdaDbBreaker(boom, "op")).rejects.toThrow();
127+
}
128+
expect(boom).toHaveBeenCalledTimes(3);
129+
130+
// Breaker is OPEN — the next call short-circuits (fn NOT invoked again).
131+
await expect(withLambdaDbBreaker(boom, "op")).rejects.toThrow(
132+
/Circuit breaker is OPEN/,
133+
);
134+
expect(boom).toHaveBeenCalledTimes(3);
135+
});
136+
});
137+
});

0 commit comments

Comments
 (0)