Skip to content

Commit 342c2fc

Browse files
ericallamTrigger.dev RepoOps
authored andcommitted
perf(webhook-engine,webapp): drop the endpoint cache and read ingress endpoints from replicas
Mono-RevId: bc687ee0208c28621389640049735e5c1ab07a98
1 parent 08d1ddb commit 342c2fc

8 files changed

Lines changed: 78 additions & 117 deletions

File tree

‎apps/webapp/app/env.server.ts‎

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -2128,10 +2128,6 @@ const EnvironmentSchema = z
21282128
"WEBHOOK_DELIVERY_STORAGE_DAYS must be a delivery retention class (3, 7, 30, 90, 180 or 365)"
21292129
),
21302130

2131-
// Ingest hot-path cache for the endpoint + resolved signing secret (keyed by opaqueId). 0 disables.
2132-
WEBHOOK_ENDPOINT_CACHE_TTL_MS: z.coerce.number().int().default(30_000),
2133-
WEBHOOK_ENDPOINT_CACHE_MAX_SIZE: z.coerce.number().int().default(10_000),
2134-
21352131
WEBHOOK_INGRESS_ENABLED: z.string().default("1"),
21362132

21372133
// Public origin for the webhook ingress URL shown to users / returned by the API. Defaults to the

‎apps/webapp/app/v3/webhookEngine.server.ts‎

Lines changed: 3 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@ import type { WebhookDeliverTaskErrorType, WebhookRedisOptions } from "@internal
77
import { tryCatch } from "@trigger.dev/core/utils";
88
import { resolve4 } from "node:dns/promises";
99
import { z } from "zod";
10-
import { prisma, webhookPartitionPrisma, webhookPrisma } from "~/db.server";
10+
import { $replica, webhookPartitionPrisma, webhookPrisma, webhookReplica } from "~/db.server";
1111
import { env } from "~/env.server";
1212
import { findEnvironmentById } from "~/models/runtimeEnvironment.server";
1313
import { logger } from "~/services/logger.server";
@@ -89,10 +89,11 @@ export function webhookVerifyTokenKey(endpointId: string): string {
8989
function createWebhookEngine() {
9090
// The engine owns the webhook tables, so it runs on the webhook DB client. The signing-secret
9191
// store stays on the main client below (SecretStore is control-plane, not part of the split).
92-
const secretStore = getSecretStore("DATABASE", { prismaClient: prisma });
92+
const secretStore = getSecretStore("DATABASE", { prismaClient: $replica });
9393

9494
const engine = new WebhookEngine({
9595
prisma: webhookPrisma,
96+
replica: webhookReplica,
9697
partitionPrisma: webhookPartitionPrisma,
9798
logLevel: env.WEBHOOK_ENGINE_LOG_LEVEL,
9899
disabled: env.WEBHOOK_ENABLED !== "1",
@@ -121,10 +122,6 @@ function createWebhookEngine() {
121122
defaultTtlSeconds: env.WEBHOOK_FRONT_GATE_DEFAULT_TTL_SECONDS,
122123
maxTtlSeconds: env.WEBHOOK_FRONT_GATE_MAX_TTL_SECONDS,
123124
},
124-
endpointCache: {
125-
ttlMs: env.WEBHOOK_ENDPOINT_CACHE_TTL_MS,
126-
maxSize: env.WEBHOOK_ENDPOINT_CACHE_MAX_SIZE,
127-
},
128125
waiters: {
129126
limits: {
130127
perEnvironment: env.WEBHOOK_WAITER_MAX_PER_ENVIRONMENT,

‎internal-packages/webhook-engine/src/engine/cache.ts‎

Lines changed: 0 additions & 51 deletions
This file was deleted.

‎internal-packages/webhook-engine/src/engine/fairness.test.ts‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -80,7 +80,6 @@ containerTestWithIsolatedRedis(
8080
tenantConcurrency: async (environmentId) => (environmentId === "env_noisy" ? 2 : 4),
8181
pollIntervalMs: 20,
8282
},
83-
endpointCache: { ttlMs: 0 },
8483
triggerTask,
8584
resolveSigningSecret: async (key) => (key === SECRET_KEY ? SECRET : undefined),
8685
logLevel: "error",
@@ -117,7 +116,6 @@ containerTestWithIsolatedRedis(
117116
const common = {
118117
prisma,
119118
redis: redisOptions,
120-
endpointCache: { ttlMs: 0 },
121119
resolveSigningSecret: async (key: string) => (key === SECRET_KEY ? SECRET : undefined),
122120
logLevel: "error" as const,
123121
};

‎internal-packages/webhook-engine/src/engine/index.ts‎

Lines changed: 15 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import { Logger } from "@trigger.dev/core/logger";
44
import type {
55
Prisma,
66
WebhookDatabase,
7+
WebhookReplicaDatabase,
78
WebhookDelivery,
89
WebhookDeliveryStatus,
910
WebhookEndpoint,
@@ -37,7 +38,6 @@ import {
3738
import { isInfrastructureError, WebhookInfrastructureError } from "./errors.js";
3839
import { QueueMemoryGuard } from "./memoryGuard.js";
3940
import { ensurePartitions, listDatedPartitions, partitionsCoveredUntil } from "./partitions.js";
40-
import { type CachedEndpoint, TtlCache } from "./cache.js";
4141
import { evaluateFilter, parseFilter } from "./filter/index.js";
4242
import { verifyWithSecrets } from "./verification/index.js";
4343
import { sha256Hex } from "./verification/util.js";
@@ -122,6 +122,13 @@ type TargetOutcome =
122122
type TaskTarget = Extract<WebhookRoutingTarget, { type: "task" }>;
123123
type SessionTarget = Extract<WebhookRoutingTarget, { type: "session" }>;
124124

125+
/** An ACTIVE endpoint with its signing secrets (current first) and parsed routing targets. */
126+
type ResolvedEndpoint = {
127+
endpoint: WebhookEndpoint;
128+
secrets: [string, ...string[]];
129+
targets: ParsedRoutingTarget[];
130+
};
131+
125132
/** The response contract declared on a verifier artifact, if any (bundles carry none). */
126133
function artifactResponseContract(artifact: WebhookVerifierArtifact) {
127134
return "response" in artifact ? artifact.response : undefined;
@@ -202,7 +209,6 @@ export class WebhookEngine {
202209
private deliveryExecutionDuration: Histogram;
203210
private deliveryExecutionFailureCounter: Counter;
204211
private ensurePartitionsCounter: Counter;
205-
private endpointCacheCounter: Counter;
206212
private ingestCounter: Counter;
207213
private ingestRefusedCounter: Counter;
208214
private frontGateRequeuedCounter: Counter;
@@ -217,19 +223,17 @@ export class WebhookEngine {
217223
private partitionCoverageTimer?: NodeJS.Timeout;
218224

219225
prisma: WebhookDatabase;
226+
private readonly replica: WebhookReplicaDatabase;
220227

221228
private triggerTask: TriggerWebhookTaskCallback;
222229
private waiterStore!: WebhookWaiterStore;
223230
private readonly waiterLimits: WebhookWaiterLimits;
224-
// Caches the endpoint + resolved signing secret per opaqueId so the ingest hot path skips two
225-
// Postgres reads (endpoint lookup + secret decrypt). Both are immutable per endpoint within the
226-
// TTL; a status change or secret rotation takes effect after at most the TTL.
227-
private endpointCache: TtlCache<CachedEndpoint>;
228231

229232
constructor(private readonly options: WebhookEngineOptions) {
230233
this.logger =
231234
options.logger ?? new Logger("WebhookEngine", (this.options.logLevel ?? "info") as any);
232235
this.prisma = options.prisma;
236+
this.replica = options.replica ?? options.prisma;
233237
this.triggerTask = options.triggerTask;
234238
this.waiterLimits = { ...DEFAULT_WAITER_LIMITS, ...options.waiters?.limits };
235239

@@ -257,9 +261,6 @@ export class WebhookEngine {
257261
this.ensurePartitionsCounter = this.meter.createCounter("webhook_ensure_partitions_total", {
258262
description: "Total number of ensurePartitions cron runs",
259263
});
260-
this.endpointCacheCounter = this.meter.createCounter("webhook_endpoint_cache_total", {
261-
description: "Endpoint+secret cache lookups on the ingest hot path, by result (hit/miss)",
262-
});
263264
this.ingestCounter = this.meter.createCounter("webhook_ingest_total", {
264265
description:
265266
"Webhook requests received, by path (endpoint or a waiter's own URL) and outcome (accepted, duplicate, verification_failed, ...)",
@@ -300,11 +301,6 @@ export class WebhookEngine {
300301
result.observe((this.partitionsCoveredUntil.getTime() - Date.now()) / DAY_MS);
301302
});
302303

303-
this.endpointCache = new TtlCache<CachedEndpoint>(
304-
options.endpointCache?.ttlMs ?? 30_000,
305-
options.endpointCache?.maxSize ?? 10_000
306-
);
307-
308304
if (options.disabled) {
309305
this.logger.info("Webhook engine disabled; skipping Redis and worker setup");
310306
return;
@@ -385,9 +381,6 @@ export class WebhookEngine {
385381
return startSpan(this.tracer, "webhook.ingest", async (span) => {
386382
span.setAttribute("opaqueId", input.opaqueId);
387383

388-
// 1+2. Resolve the endpoint + signing secret (cached per opaqueId). Fail-closed outcomes
389-
// (not found / inactive / secret missing) are never cached, so verify only ever runs with a
390-
// non-empty secret.
391384
const resolved = await this.#resolveEndpoint(input.opaqueId);
392385
if (!resolved.ok) return resolved.result;
393386
const { endpoint, secrets } = resolved;
@@ -661,7 +654,7 @@ export class WebhookEngine {
661654
this.#assertEnabled();
662655
return startSpan(this.tracer, "webhook.unsupportedMethod", async (span) => {
663656
span.setAttribute("opaqueId", opaqueId);
664-
const endpoint = await this.prisma.webhookEndpoint.findFirst({
657+
const endpoint = await this.replica.webhookEndpoint.findFirst({
665658
where: { opaqueId },
666659
select: { status: true, verifierArtifact: true },
667660
});
@@ -693,7 +686,7 @@ export class WebhookEngine {
693686
return startSpan(this.tracer, "webhook.getHandshake", async (span) => {
694687
span.setAttribute("opaqueId", input.opaqueId);
695688

696-
const endpoint = await this.prisma.webhookEndpoint.findFirst({
689+
const endpoint = await this.replica.webhookEndpoint.findFirst({
697690
where: { opaqueId: input.opaqueId },
698691
});
699692
if (!endpoint) return { outcome: "endpoint_not_found" };
@@ -1239,20 +1232,11 @@ export class WebhookEngine {
12391232
});
12401233
}
12411234

1242-
// Resolve the endpoint + plaintext signing secret for an opaqueId, cached. Only fully-resolved
1243-
// (ACTIVE + secret present) endpoints are cached; fail-closed outcomes always re-read.
12441235
async #resolveEndpoint(
12451236
opaqueId: string
1246-
): Promise<({ ok: true } & CachedEndpoint) | { ok: false; result: IngestResult }> {
1247-
const cached = this.endpointCache.get(opaqueId);
1248-
if (cached) {
1249-
this.endpointCacheCounter.add(1, { result: "hit" });
1250-
return { ok: true, ...cached };
1251-
}
1252-
this.endpointCacheCounter.add(1, { result: "miss" });
1253-
1237+
): Promise<({ ok: true } & ResolvedEndpoint) | { ok: false; result: IngestResult }> {
12541238
// Single-row via the global @unique on opaqueId (Q2).
1255-
const endpoint = await this.prisma.webhookEndpoint.findFirst({ where: { opaqueId } });
1239+
const endpoint = await this.replica.webhookEndpoint.findFirst({ where: { opaqueId } });
12561240
if (!endpoint) return { ok: false, result: { outcome: "endpoint_not_found" } };
12571241
if (endpoint.status !== "ACTIVE")
12581242
return { ok: false, result: { outcome: "endpoint_inactive" } };
@@ -1277,9 +1261,7 @@ export class WebhookEngine {
12771261

12781262
const targets = parseRoutingTargets(endpoint, this.logger);
12791263

1280-
const secrets: CachedEndpoint["secrets"] = [secret, ...previousSecrets];
1281-
this.endpointCache.set(opaqueId, { endpoint, secrets, targets });
1282-
return { ok: true, endpoint, secrets, targets };
1264+
return { ok: true, endpoint, secrets: [secret, ...previousSecrets], targets };
12831265
}
12841266

12851267
// Evaluate a session routing target's `startOn` against the event. Absent => start allowed. A parse or

‎internal-packages/webhook-engine/src/engine/ingest.test.ts‎

Lines changed: 53 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@ import {
44
createStandalonePostgresContainer,
55
} from "@internal/testcontainers";
66
import { createRedisClient } from "@internal/redis";
7-
import { type Prisma, PrismaClient } from "@trigger.dev/database";
7+
import { type Prisma, PrismaClient, type WebhookReplicaDatabase } from "@trigger.dev/database";
88
import type { RedisOptions } from "@internal/redis";
99
import type { Meter } from "@internal/tracing";
1010
import {
@@ -161,27 +161,26 @@ function buildEngine(
161161
triggerTask: TriggerWebhookTaskCallback,
162162
over?: {
163163
workerDisabled?: boolean;
164-
endpointCacheTtlMs?: number;
165164
resolveSigningSecret?: (key: string) => Promise<string | readonly string[] | undefined>;
166165
resolveVerifyToken?: (endpointId: string) => Promise<string | undefined>;
167166
deliverToSession?: DeliverWebhookToSessionCallback;
168167
meter?: Meter;
169168
retentionDays?: (environmentId: string) => Promise<number>;
169+
replica?: WebhookReplicaDatabase;
170170
}
171171
) {
172172
return new WebhookEngine({
173173
meter: over?.meter,
174174
resolveVerifyToken: over?.resolveVerifyToken,
175175
prisma,
176+
replica: over?.replica,
176177
redis: redisOptions,
177178
worker: {
178179
concurrency: 1,
179180
pollIntervalMs: 50,
180181
disabled: over?.workerDisabled,
181182
retryBaseDelayMs: 20,
182183
},
183-
endpointCache:
184-
over?.endpointCacheTtlMs !== undefined ? { ttlMs: over.endpointCacheTtlMs } : undefined,
185184
triggerTask,
186185
deliverToSession: over?.deliverToSession,
187186
retention: over?.retentionDays ? { forEnvironment: over.retentionDays } : undefined,
@@ -773,35 +772,73 @@ containerTestWithIsolatedRedis(
773772
);
774773

775774
containerTestWithIsolatedRedis(
776-
"caches the endpoint + secret across ingests and re-reads after the TTL",
775+
"resolves the endpoint and secret on every ingest, so a new secret applies to the next delivery",
777776
async ({ prisma, redisOptions }) => {
778777
const endpoint = await createEndpoint(prisma);
779778
const { triggerTask } = makeTriggerTaskStub();
779+
let secret = SECRET;
780780
let secretReads = 0;
781781
const engine = buildEngine(prisma, redisOptions, triggerTask, {
782-
endpointCacheTtlMs: 200,
783782
resolveSigningSecret: async (key) => {
784783
secretReads++;
785-
return key === SECRET_KEY ? SECRET : undefined;
784+
return key === SECRET_KEY ? secret : undefined;
786785
},
787786
});
788787

789788
try {
790-
expect((await engine.ingest(signedInput("evt_cache_1", endpoint.opaqueId))).outcome).toBe(
789+
expect((await engine.ingest(signedInput("evt_fresh_1", endpoint.opaqueId))).outcome).toBe(
791790
"accepted"
792791
);
793-
expect((await engine.ingest(signedInput("evt_cache_2", endpoint.opaqueId))).outcome).toBe(
792+
expect((await engine.ingest(signedInput("evt_fresh_2", endpoint.opaqueId))).outcome).toBe(
794793
"accepted"
795794
);
796-
// Second ingest hit the cache, so the secret was resolved only once.
797-
expect(secretReads).toBe(1);
795+
expect(secretReads).toBe(2);
798796

799-
// After the TTL the entry expires and the next ingest re-resolves.
800-
await new Promise((r) => setTimeout(r, 250));
801-
expect((await engine.ingest(signedInput("evt_cache_3", endpoint.opaqueId))).outcome).toBe(
802-
"accepted"
797+
secret = "whsec_rotated_secret";
798+
expect((await engine.ingest(signedInput("evt_fresh_3", endpoint.opaqueId))).outcome).toBe(
799+
"verification_failed"
803800
);
804-
expect(secretReads).toBe(2);
801+
802+
await prisma.webhookEndpoint.update({
803+
where: { id: endpoint.id },
804+
data: { status: "INACTIVE" },
805+
});
806+
expect((await engine.ingest(signedInput("evt_fresh_4", endpoint.opaqueId))).outcome).toBe(
807+
"endpoint_inactive"
808+
);
809+
} finally {
810+
await engine.quit();
811+
}
812+
}
813+
);
814+
815+
containerTestWithIsolatedRedis(
816+
"reads the endpoint for public ingress from the replica, not the writer",
817+
async ({ prisma, redisOptions }) => {
818+
const endpoint = await createEndpoint(prisma);
819+
const { triggerTask } = makeTriggerTaskStub();
820+
const replicaReads: string[] = [];
821+
const laggingReplica = {
822+
webhookEndpoint: {
823+
findFirst: async (args: { where: { opaqueId: string } }) => {
824+
replicaReads.push(args.where.opaqueId);
825+
return null;
826+
},
827+
},
828+
} as unknown as WebhookReplicaDatabase;
829+
const engine = buildEngine(prisma, redisOptions, triggerTask, { replica: laggingReplica });
830+
831+
try {
832+
expect((await engine.ingest(signedInput("evt_replica_1", endpoint.opaqueId))).outcome).toBe(
833+
"endpoint_not_found"
834+
);
835+
expect((await engine.rejectUnsupportedMethod(endpoint.opaqueId)).outcome).toBe(
836+
"endpoint_not_found"
837+
);
838+
expect(
839+
(await engine.verifyGetHandshake({ opaqueId: endpoint.opaqueId, query: {} })).outcome
840+
).toBe("endpoint_not_found");
841+
expect(replicaReads).toEqual([endpoint.opaqueId, endpoint.opaqueId, endpoint.opaqueId]);
805842
} finally {
806843
await engine.quit();
807844
}
@@ -815,7 +852,6 @@ containerTestWithIsolatedRedis(
815852
const { triggerTask } = makeTriggerTaskStub();
816853
let secrets: string[] = ["whsec_rotated_secret", SECRET];
817854
const engine = buildEngine(prisma, redisOptions, triggerTask, {
818-
endpointCacheTtlMs: 0,
819855
resolveSigningSecret: async (key) => (key === SECRET_KEY ? secrets : undefined),
820856
});
821857

0 commit comments

Comments
 (0)