fix(run-engine): debounce hot-key lock contention and 5xx feedback loop (#3453)
## Changes
Three changes in
`internal-packages/run-engine/src/engine/systems/debounceSystem.ts`, in
order of impact:
1. **Fast-path skip before the lock.** In `handleExistingRun`, do an
unlocked read of `delayUntil` (and `createdAt` for the max-duration
check) from the run row before entering `runLock.lock("handleDebounce",
...)`. If `newDelayUntil <= currentDelayUntil` and the run is still
within its max-duration window, return the existing run immediately
without taking the lock. Safe because debounce is monotonic-forward only
— a stale read either matches reality or undershoots, both of which
decay correctly (re-checked properly inside the lock by whichever caller
is actually pushing forward). Trailing-mode triggers carrying
`updateData` still take the lock so the data update is applied.
2. **Quantize `newDelayUntil`.** Round the computed `newDelayUntil` to
1-second buckets (configurable via `quantizeNewDelayUntilMs`, set to 0
to disable). Without quantization, every call has a slightly larger
`newDelayUntil` than the last and they all pass the fast-path check.
With it, concurrent callers on the same key share a target time and ~95%
short-circuit. User-visible effect: a debounced run might fire up to 1s
earlier than the strict spec — non-issue for typical debounce use cases
(chat summarization, batched notifications, etc.).
3. **Graceful lock-contention fallback.** Wrap the `runLock.lock(...)`
call so `LockAcquisitionTimeoutError` and Redlock `ExecutionError` /
`ResourceLockedError` return the existing run id with success instead of
propagating a 5xx. Debounce is best-effort: if we can't take the lock,
the herd is already updating it for us; fall in line. This kills the 5xx
→ SDK-retry feedback loop. With (1)+(2) this rarely fires; without them
it's the difference between 5xx and 200.
Defaults preserve current behaviour aside from quantization (1s) and
fast-path (on). Both are configurable via `RunEngineOptions.debounce`.
## ✅ Checklist
- [x] I have followed every step in the [contributing
guide](https://github.com/triggerdotdev/trigger.dev/blob/main/CONTRIBUTING.md)
- [x] The PR title follows the convention.
- [x] I ran and tested the code works
---
## Changelog
Reduce 5xx feedback loops on hot debounce keys by quantizing
`delayUntil`, adding an unlocked fast-path skip before the redlock, and
gracefully handling redlock contention in `handleDebounce` so the SDK no
longer retries into a herd.
---------
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Eric Allam
@@ -324,6 +324,9 @@ export class RunEngine {
|
||||
executionSnapshotSystem: this.executionSnapshotSystem,
|
||||
delayedRunSystem: this.delayedRunSystem,
|
||||
maxDebounceDurationMs: options.debounce?.maxDebounceDurationMs ?? 60 * 60 * 1000, // Default 1 hour
|
||||
quantizeNewDelayUntilMs: options.debounce?.quantizeNewDelayUntilMs ?? 1000,
|
||||
fastPathSkipEnabled: options.debounce?.fastPathSkipEnabled ?? true,
|
||||
useReplicaForFastPathRead: options.debounce?.useReplicaForFastPathRead ?? false,
|
||||
});
|
||||
|
||||
this.pendingVersionSystem = new PendingVersionSystem({
|
||||
|
||||
@@ -10,11 +10,17 @@ import {
|
||||
parseNaturalLanguageDuration,
|
||||
parseNaturalLanguageDurationInMs,
|
||||
} from "@trigger.dev/core/v3/isomorphic";
|
||||
import { PrismaClientOrTransaction, TaskRun, Waitpoint } from "@trigger.dev/database";
|
||||
import {
|
||||
PrismaClientOrTransaction,
|
||||
PrismaReplicaClient,
|
||||
TaskRun,
|
||||
Waitpoint,
|
||||
} from "@trigger.dev/database";
|
||||
import { nanoid } from "nanoid";
|
||||
import { SystemResources } from "./systems.js";
|
||||
import { ExecutionSnapshotSystem, getLatestExecutionSnapshot } from "./executionSnapshotSystem.js";
|
||||
import { DelayedRunSystem } from "./delayedRunSystem.js";
|
||||
import { LockAcquisitionTimeoutError } from "../locking.js";
|
||||
|
||||
export type DebounceOptions = {
|
||||
key: string;
|
||||
@@ -45,6 +51,22 @@ export type DebounceSystemOptions = {
|
||||
executionSnapshotSystem: ExecutionSnapshotSystem;
|
||||
delayedRunSystem: DelayedRunSystem;
|
||||
maxDebounceDurationMs: number;
|
||||
/**
|
||||
* Bucket size in milliseconds used to quantize the newly computed `delayUntil`.
|
||||
* Set to 0 to disable quantization.
|
||||
*/
|
||||
quantizeNewDelayUntilMs?: number;
|
||||
/**
|
||||
* When true, read the existing run's `delayUntil` outside the redlock and
|
||||
* short-circuit if the new (quantized) `delayUntil` is not later than the
|
||||
* current one.
|
||||
*/
|
||||
fastPathSkipEnabled?: boolean;
|
||||
/**
|
||||
* When true, route the unlocked fast-path reads (probe + full-run fetch)
|
||||
* through `readOnlyPrisma` (e.g. an Aurora reader) instead of the writer.
|
||||
*/
|
||||
useReplicaForFastPathRead?: boolean;
|
||||
};
|
||||
|
||||
export type DebounceResult =
|
||||
@@ -89,6 +111,9 @@ export class DebounceSystem {
|
||||
private readonly executionSnapshotSystem: ExecutionSnapshotSystem;
|
||||
private readonly delayedRunSystem: DelayedRunSystem;
|
||||
private readonly maxDebounceDurationMs: number;
|
||||
private readonly quantizeNewDelayUntilMs: number;
|
||||
private readonly fastPathSkipEnabled: boolean;
|
||||
private readonly useReplicaForFastPathRead: boolean;
|
||||
|
||||
constructor(options: DebounceSystemOptions) {
|
||||
this.$ = options.resources;
|
||||
@@ -106,6 +131,9 @@ export class DebounceSystem {
|
||||
this.executionSnapshotSystem = options.executionSnapshotSystem;
|
||||
this.delayedRunSystem = options.delayedRunSystem;
|
||||
this.maxDebounceDurationMs = options.maxDebounceDurationMs;
|
||||
this.quantizeNewDelayUntilMs = Math.max(0, options.quantizeNewDelayUntilMs ?? 1000);
|
||||
this.fastPathSkipEnabled = options.fastPathSkipEnabled ?? true;
|
||||
this.useReplicaForFastPathRead = options.useReplicaForFastPathRead ?? false;
|
||||
|
||||
this.#registerCommands();
|
||||
}
|
||||
@@ -450,9 +478,264 @@ return 0
|
||||
debounce: DebounceOptions;
|
||||
tx?: PrismaClientOrTransaction;
|
||||
}): Promise<DebounceResult> {
|
||||
return await this.$.runLock.lock("handleDebounce", [existingRunId], async () => {
|
||||
const prisma = tx ?? this.$.prisma;
|
||||
const prisma = tx ?? this.$.prisma;
|
||||
// Reads in the unlocked fast-path can run on `readOnlyPrisma` when
|
||||
// configured (e.g. an Aurora reader). Replica lag is fine: debounce is
|
||||
// best-effort and a stale read either falls through to the locked path
|
||||
// (when delayUntil hasn't replicated yet) or returns the existing run
|
||||
// (when the run's status is stale). The latter is the same outcome the
|
||||
// caller would see if their trigger had simply landed a few hundred ms
|
||||
// earlier, which is within the natural debounce race. Only divert reads
|
||||
// when the caller isn't inside a tx (where the read needs to see the
|
||||
// tx's writes).
|
||||
const fastPathReadPrisma =
|
||||
tx ?? (this.useReplicaForFastPathRead ? this.$.readOnlyPrisma : this.$.prisma);
|
||||
|
||||
// Compute the (quantized) target delayUntil up-front, before taking any lock.
|
||||
// Quantizing to e.g. 1s buckets collapses many concurrent triggers on the same
|
||||
// hot debounce key onto the same target time, so the unlocked fast-path skip
|
||||
// below becomes effective and the redlock is not contended.
|
||||
const newDelayUntil = this.#computeQuantizedDelayUntil(debounce.delay);
|
||||
|
||||
// Fast-path: read the current delayUntil outside the redlock and short-circuit
|
||||
// if our (quantized) newDelayUntil isn't later than what's already scheduled.
|
||||
// Safe because debounce is monotonic-forward only: a stale read either matches
|
||||
// reality or undershoots, both of which decay correctly (re-checked properly
|
||||
// inside the lock by whoever is actually pushing forward).
|
||||
if (this.fastPathSkipEnabled && newDelayUntil) {
|
||||
const fastPathResult = await this.#tryFastPathSkip({
|
||||
existingRunId,
|
||||
newDelayUntil,
|
||||
debounce,
|
||||
prisma: fastPathReadPrisma,
|
||||
});
|
||||
if (fastPathResult) {
|
||||
return fastPathResult;
|
||||
}
|
||||
}
|
||||
|
||||
try {
|
||||
return await this.$.runLock.lock("handleDebounce", [existingRunId], async () => {
|
||||
return await this.#handleExistingRunLocked({
|
||||
existingRunId,
|
||||
redisKey,
|
||||
environmentId,
|
||||
taskIdentifier,
|
||||
debounce,
|
||||
newDelayUntil,
|
||||
prisma,
|
||||
tx,
|
||||
});
|
||||
});
|
||||
} catch (error) {
|
||||
// Lock contention safety net: if we couldn't take the lock (redlock quorum
|
||||
// failure or our retry budget exhausted), fall in line with whoever is
|
||||
// actually updating the run instead of bubbling a 5xx to the SDK and
|
||||
// amplifying the herd via SDK retries. Debounce is best-effort - dropping
|
||||
// our contribution to delayUntil here is fine, the herd is updating it for
|
||||
// us.
|
||||
if (this.#isLockContentionError(error)) {
|
||||
return await this.#handleLockContentionFallback({
|
||||
existingRunId,
|
||||
debounce,
|
||||
error,
|
||||
prisma,
|
||||
});
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Parses the debounce delay and (optionally) quantizes it to a bucket boundary
|
||||
* by flooring the absolute timestamp. Quantization makes concurrent triggers on
|
||||
* the same key share a target time, which is what makes the unlocked fast-path
|
||||
* skip effective.
|
||||
*/
|
||||
#computeQuantizedDelayUntil(delay: string): Date | null {
|
||||
const parsed = parseNaturalLanguageDuration(delay);
|
||||
if (!parsed) {
|
||||
return null;
|
||||
}
|
||||
if (this.quantizeNewDelayUntilMs <= 0) {
|
||||
return parsed;
|
||||
}
|
||||
const bucket = this.quantizeNewDelayUntilMs;
|
||||
const quantized = Math.floor(parsed.getTime() / bucket) * bucket;
|
||||
return new Date(quantized);
|
||||
}
|
||||
|
||||
#isLockContentionError(error: unknown): boolean {
|
||||
if (!(error instanceof Error)) return false;
|
||||
return (
|
||||
error instanceof LockAcquisitionTimeoutError ||
|
||||
error.name === "LockAcquisitionTimeoutError" ||
|
||||
error.name === "ExecutionError" ||
|
||||
error.name === "ResourceLockedError"
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Reads `delayUntil`/`status`/`createdAt` outside the redlock and
|
||||
* short-circuits if the existing scheduled time already covers our target.
|
||||
* Skips trailing-mode triggers that carry `updateData` since those still need
|
||||
* the lock to apply their data update. Also falls through when the run has
|
||||
* already exceeded its max debounce duration so the locked path can return
|
||||
* `max_duration_exceeded` and let the caller create a new run.
|
||||
*
|
||||
* `prisma` may be a read replica - replica lag is acceptable because
|
||||
* debounce is best-effort. A stale `delayUntil` either matches reality or
|
||||
* undershoots (we fall through to the locked path); a stale `status` at
|
||||
* worst returns the existing run, which is the same outcome the caller
|
||||
* would see if their trigger had landed a few hundred ms earlier.
|
||||
*/
|
||||
async #tryFastPathSkip({
|
||||
existingRunId,
|
||||
newDelayUntil,
|
||||
debounce,
|
||||
prisma,
|
||||
}: {
|
||||
existingRunId: string;
|
||||
newDelayUntil: Date;
|
||||
debounce: DebounceOptions;
|
||||
prisma: PrismaClientOrTransaction | PrismaReplicaClient;
|
||||
}): Promise<DebounceResult | null> {
|
||||
// Trailing mode with updateData still needs the lock so the data update is
|
||||
// applied; only short-circuit when there's nothing to update.
|
||||
if (debounce.mode === "trailing" && debounce.updateData) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const probe = await prisma.taskRun.findFirst({
|
||||
where: { id: existingRunId },
|
||||
select: { status: true, delayUntil: true, createdAt: true },
|
||||
});
|
||||
if (!probe || probe.status !== "DELAYED" || !probe.delayUntil) {
|
||||
return null;
|
||||
}
|
||||
if (newDelayUntil.getTime() > probe.delayUntil.getTime()) {
|
||||
return null;
|
||||
}
|
||||
|
||||
// Fall through to the lock path when newDelayUntil would exceed the run's
|
||||
// max debounce window so the caller can return max_duration_exceeded and
|
||||
// create a fresh run.
|
||||
let maxDurationMs = this.maxDebounceDurationMs;
|
||||
if (debounce.maxDelay) {
|
||||
const parsedMaxDelay = parseNaturalLanguageDurationInMs(debounce.maxDelay);
|
||||
if (parsedMaxDelay !== undefined) {
|
||||
maxDurationMs = parsedMaxDelay;
|
||||
}
|
||||
}
|
||||
const maxDelayUntilMs = probe.createdAt.getTime() + maxDurationMs;
|
||||
if (newDelayUntil.getTime() > maxDelayUntilMs) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const fullRun = await prisma.taskRun.findFirst({
|
||||
where: { id: existingRunId },
|
||||
include: { associatedWaitpoint: true },
|
||||
});
|
||||
if (!fullRun || fullRun.status !== "DELAYED") {
|
||||
return null;
|
||||
}
|
||||
|
||||
this.$.logger.debug("handleExistingRun: fast-path skip, existing delayUntil already covers", {
|
||||
existingRunId,
|
||||
debounceKey: debounce.key,
|
||||
newDelayUntil,
|
||||
currentDelayUntil: fullRun.delayUntil,
|
||||
});
|
||||
|
||||
return {
|
||||
status: "existing",
|
||||
run: fullRun,
|
||||
waitpoint: fullRun.associatedWaitpoint,
|
||||
};
|
||||
}
|
||||
|
||||
async #handleLockContentionFallback({
|
||||
existingRunId,
|
||||
debounce,
|
||||
error,
|
||||
prisma,
|
||||
}: {
|
||||
existingRunId: string;
|
||||
debounce: DebounceOptions;
|
||||
error: unknown;
|
||||
prisma: PrismaClientOrTransaction;
|
||||
}): Promise<DebounceResult> {
|
||||
const fullRun = await prisma.taskRun.findFirst({
|
||||
where: { id: existingRunId },
|
||||
include: { associatedWaitpoint: true },
|
||||
});
|
||||
|
||||
if (!fullRun || fullRun.status !== "DELAYED") {
|
||||
// The run is no longer in a state we can safely return as "existing" -
|
||||
// re-throw so the caller surfaces the failure rather than silently
|
||||
// succeeding on a stale/terminated run.
|
||||
this.$.logger.warn(
|
||||
"handleExistingRun: lock contention, but existing run no longer DELAYED - rethrowing",
|
||||
{
|
||||
existingRunId,
|
||||
debounceKey: debounce.key,
|
||||
status: fullRun?.status,
|
||||
}
|
||||
);
|
||||
throw error;
|
||||
}
|
||||
|
||||
// Trailing-mode triggers carrying updateData fall through to the same
|
||||
// "return existing" path as everything else. Under lock contention some
|
||||
// other concurrent caller is winning the lock right now and applying
|
||||
// *its* updateData (which is, by wall-clock, ms-different from ours and
|
||||
// indistinguishable to the user). Re-throwing here would just produce a
|
||||
// 5xx that the SDK retries with our now-older payload - more likely to
|
||||
// result in stale data landing than letting the herd's winner stand.
|
||||
this.$.logger.warn(
|
||||
"handleExistingRun: lock contention, returning existing run without rescheduling",
|
||||
{
|
||||
existingRunId,
|
||||
debounceKey: debounce.key,
|
||||
currentDelayUntil: fullRun.delayUntil,
|
||||
mode: debounce.mode,
|
||||
hasUpdateData: !!debounce.updateData,
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
errorName: error instanceof Error ? error.name : undefined,
|
||||
}
|
||||
);
|
||||
|
||||
return {
|
||||
status: "existing",
|
||||
run: fullRun,
|
||||
waitpoint: fullRun.associatedWaitpoint,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Body of `handleExistingRun` that runs while holding the redlock on the run.
|
||||
* Receives the (possibly quantized) `newDelayUntil` precomputed by the caller.
|
||||
*/
|
||||
async #handleExistingRunLocked({
|
||||
existingRunId,
|
||||
redisKey,
|
||||
environmentId,
|
||||
taskIdentifier,
|
||||
debounce,
|
||||
newDelayUntil,
|
||||
prisma,
|
||||
tx,
|
||||
}: {
|
||||
existingRunId: string;
|
||||
redisKey: string;
|
||||
environmentId: string;
|
||||
taskIdentifier: string;
|
||||
debounce: DebounceOptions;
|
||||
newDelayUntil: Date | null;
|
||||
prisma: PrismaClientOrTransaction;
|
||||
tx?: PrismaClientOrTransaction;
|
||||
}): Promise<DebounceResult> {
|
||||
{
|
||||
// Get the latest execution snapshot
|
||||
let snapshot;
|
||||
try {
|
||||
@@ -514,8 +797,6 @@ return 0
|
||||
});
|
||||
}
|
||||
|
||||
// Calculate new delay - parseNaturalLanguageDuration returns a Date (now + duration)
|
||||
const newDelayUntil = parseNaturalLanguageDuration(debounce.delay);
|
||||
if (!newDelayUntil) {
|
||||
this.$.logger.error("handleExistingRun: invalid delay duration", {
|
||||
delay: debounce.delay,
|
||||
@@ -619,7 +900,7 @@ return 0
|
||||
run: updatedRun,
|
||||
waitpoint: existingRun.associatedWaitpoint,
|
||||
};
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -4,6 +4,7 @@ import { expect } from "vitest";
|
||||
import { RunEngine } from "../index.js";
|
||||
import { setTimeout } from "timers/promises";
|
||||
import { setupAuthenticatedEnvironment, setupBackgroundWorker } from "./setup.js";
|
||||
import { createRedisClient } from "@internal/redis";
|
||||
|
||||
vi.setConfig({ testTimeout: 60_000 });
|
||||
|
||||
@@ -240,6 +241,8 @@ describe("RunEngine debounce", () => {
|
||||
},
|
||||
debounce: {
|
||||
maxDebounceDurationMs: 60_000,
|
||||
// Disable quantization so this test can observe sub-second extensions.
|
||||
quantizeNewDelayUntilMs: 0,
|
||||
},
|
||||
tracer: trace.getTracer("test", "0.0.0"),
|
||||
});
|
||||
@@ -2497,5 +2500,700 @@ describe("RunEngine debounce", () => {
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
containerTest(
|
||||
"Debounce fast-path: subsequent triggers within the same quantization bucket skip the lock",
|
||||
async ({ prisma, redisOptions }) => {
|
||||
const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
|
||||
|
||||
const engine = new RunEngine({
|
||||
prisma,
|
||||
worker: {
|
||||
redis: redisOptions,
|
||||
workers: 1,
|
||||
tasksPerWorker: 10,
|
||||
pollIntervalMs: 100,
|
||||
},
|
||||
queue: {
|
||||
redis: redisOptions,
|
||||
},
|
||||
runLock: {
|
||||
redis: redisOptions,
|
||||
},
|
||||
machines: {
|
||||
defaultMachine: "small-1x",
|
||||
machines: {
|
||||
"small-1x": {
|
||||
name: "small-1x" as const,
|
||||
cpu: 0.5,
|
||||
memory: 0.5,
|
||||
centsPerMs: 0.0001,
|
||||
},
|
||||
},
|
||||
baseCostInCents: 0.0001,
|
||||
},
|
||||
debounce: {
|
||||
maxDebounceDurationMs: 60_000,
|
||||
// Wide bucket so the second trigger is guaranteed to land in the same one.
|
||||
quantizeNewDelayUntilMs: 60_000,
|
||||
},
|
||||
tracer: trace.getTracer("test", "0.0.0"),
|
||||
});
|
||||
|
||||
try {
|
||||
const taskIdentifier = "test-task";
|
||||
|
||||
await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);
|
||||
|
||||
const run1 = await engine.trigger(
|
||||
{
|
||||
number: 1,
|
||||
friendlyId: "run_fp1",
|
||||
environment: authenticatedEnvironment,
|
||||
taskIdentifier,
|
||||
payload: '{"data": "first"}',
|
||||
payloadType: "application/json",
|
||||
context: {},
|
||||
traceContext: {},
|
||||
traceId: "t_fp_1",
|
||||
spanId: "s_fp_1",
|
||||
workerQueue: "main",
|
||||
queue: "task/test-task",
|
||||
isTest: false,
|
||||
tags: [],
|
||||
delayUntil: new Date(Date.now() + 5000),
|
||||
debounce: {
|
||||
key: "fast-path-key",
|
||||
delay: "5s",
|
||||
},
|
||||
},
|
||||
prisma
|
||||
);
|
||||
|
||||
const originalDelayUntil = run1.delayUntil;
|
||||
assertNonNullable(originalDelayUntil);
|
||||
|
||||
// Update the delayUntil directly to a far-future value, simulating a
|
||||
// previous trigger that already pushed the bucket forward. The next
|
||||
// call's quantized newDelayUntil will land at-or-before this, so the
|
||||
// fast-path should skip the lock and leave delayUntil untouched.
|
||||
const farFuture = new Date(Date.now() + 10 * 60_000);
|
||||
await prisma.taskRun.update({
|
||||
where: { id: run1.id },
|
||||
data: { delayUntil: farFuture },
|
||||
});
|
||||
|
||||
const run2 = await engine.trigger(
|
||||
{
|
||||
number: 2,
|
||||
friendlyId: "run_fp2",
|
||||
environment: authenticatedEnvironment,
|
||||
taskIdentifier,
|
||||
payload: '{"data": "second"}',
|
||||
payloadType: "application/json",
|
||||
context: {},
|
||||
traceContext: {},
|
||||
traceId: "t_fp_2",
|
||||
spanId: "s_fp_2",
|
||||
workerQueue: "main",
|
||||
queue: "task/test-task",
|
||||
isTest: false,
|
||||
tags: [],
|
||||
delayUntil: new Date(Date.now() + 5000),
|
||||
debounce: {
|
||||
key: "fast-path-key",
|
||||
delay: "5s",
|
||||
},
|
||||
},
|
||||
prisma
|
||||
);
|
||||
|
||||
expect(run2.id).toBe(run1.id);
|
||||
|
||||
const updatedRun = await prisma.taskRun.findFirst({
|
||||
where: { id: run1.id },
|
||||
});
|
||||
assertNonNullable(updatedRun);
|
||||
assertNonNullable(updatedRun.delayUntil);
|
||||
|
||||
// delayUntil must NOT have been bumped backward by the second trigger,
|
||||
// proving we short-circuited before taking the lock or rescheduling.
|
||||
expect(updatedRun.delayUntil.getTime()).toBe(farFuture.getTime());
|
||||
} finally {
|
||||
await engine.quit();
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
containerTest(
|
||||
"Debounce fast-path: trailing mode with updateData still takes the lock",
|
||||
async ({ prisma, redisOptions }) => {
|
||||
const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
|
||||
|
||||
const engine = new RunEngine({
|
||||
prisma,
|
||||
worker: {
|
||||
redis: redisOptions,
|
||||
workers: 1,
|
||||
tasksPerWorker: 10,
|
||||
pollIntervalMs: 100,
|
||||
},
|
||||
queue: {
|
||||
redis: redisOptions,
|
||||
},
|
||||
runLock: {
|
||||
redis: redisOptions,
|
||||
},
|
||||
machines: {
|
||||
defaultMachine: "small-1x",
|
||||
machines: {
|
||||
"small-1x": {
|
||||
name: "small-1x" as const,
|
||||
cpu: 0.5,
|
||||
memory: 0.5,
|
||||
centsPerMs: 0.0001,
|
||||
},
|
||||
},
|
||||
baseCostInCents: 0.0001,
|
||||
},
|
||||
debounce: {
|
||||
maxDebounceDurationMs: 60_000,
|
||||
quantizeNewDelayUntilMs: 60_000,
|
||||
},
|
||||
tracer: trace.getTracer("test", "0.0.0"),
|
||||
});
|
||||
|
||||
try {
|
||||
const taskIdentifier = "test-task";
|
||||
|
||||
await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);
|
||||
|
||||
const run1 = await engine.trigger(
|
||||
{
|
||||
number: 1,
|
||||
friendlyId: "run_trfp1",
|
||||
environment: authenticatedEnvironment,
|
||||
taskIdentifier,
|
||||
payload: '{"data": "first"}',
|
||||
payloadType: "application/json",
|
||||
context: {},
|
||||
traceContext: {},
|
||||
traceId: "t_trfp_1",
|
||||
spanId: "s_trfp_1",
|
||||
workerQueue: "main",
|
||||
queue: "task/test-task",
|
||||
isTest: false,
|
||||
tags: [],
|
||||
delayUntil: new Date(Date.now() + 5000),
|
||||
debounce: {
|
||||
key: "trailing-fast-path-key",
|
||||
delay: "5s",
|
||||
mode: "trailing",
|
||||
},
|
||||
},
|
||||
prisma
|
||||
);
|
||||
|
||||
// Push delayUntil far forward so the fast-path *would* short-circuit
|
||||
// for leading mode. Trailing-mode triggers with updateData must still
|
||||
// take the lock so the data update is applied.
|
||||
const farFuture = new Date(Date.now() + 10 * 60_000);
|
||||
await prisma.taskRun.update({
|
||||
where: { id: run1.id },
|
||||
data: { delayUntil: farFuture },
|
||||
});
|
||||
|
||||
const run2 = await engine.trigger(
|
||||
{
|
||||
number: 2,
|
||||
friendlyId: "run_trfp2",
|
||||
environment: authenticatedEnvironment,
|
||||
taskIdentifier,
|
||||
payload: '{"data": "second"}',
|
||||
payloadType: "application/json",
|
||||
context: {},
|
||||
traceContext: {},
|
||||
traceId: "t_trfp_2",
|
||||
spanId: "s_trfp_2",
|
||||
workerQueue: "main",
|
||||
queue: "task/test-task",
|
||||
isTest: false,
|
||||
tags: [],
|
||||
delayUntil: new Date(Date.now() + 5000),
|
||||
debounce: {
|
||||
key: "trailing-fast-path-key",
|
||||
delay: "5s",
|
||||
mode: "trailing",
|
||||
updateData: {
|
||||
payload: '{"data": "second"}',
|
||||
payloadType: "application/json",
|
||||
},
|
||||
},
|
||||
},
|
||||
prisma
|
||||
);
|
||||
|
||||
expect(run2.id).toBe(run1.id);
|
||||
|
||||
const updatedRun = await prisma.taskRun.findFirst({
|
||||
where: { id: run1.id },
|
||||
});
|
||||
assertNonNullable(updatedRun);
|
||||
// Trailing-mode update went through the lock and rewrote the payload.
|
||||
expect(updatedRun.payload).toBe('{"data": "second"}');
|
||||
} finally {
|
||||
await engine.quit();
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
containerTest(
|
||||
"Debounce: quantized newDelayUntil falls on a bucket boundary",
|
||||
async ({ prisma, redisOptions }) => {
|
||||
const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
|
||||
|
||||
const engine = new RunEngine({
|
||||
prisma,
|
||||
worker: {
|
||||
redis: redisOptions,
|
||||
workers: 1,
|
||||
tasksPerWorker: 10,
|
||||
pollIntervalMs: 100,
|
||||
},
|
||||
queue: {
|
||||
redis: redisOptions,
|
||||
},
|
||||
runLock: {
|
||||
redis: redisOptions,
|
||||
},
|
||||
machines: {
|
||||
defaultMachine: "small-1x",
|
||||
machines: {
|
||||
"small-1x": {
|
||||
name: "small-1x" as const,
|
||||
cpu: 0.5,
|
||||
memory: 0.5,
|
||||
centsPerMs: 0.0001,
|
||||
},
|
||||
},
|
||||
baseCostInCents: 0.0001,
|
||||
},
|
||||
debounce: {
|
||||
maxDebounceDurationMs: 60_000,
|
||||
quantizeNewDelayUntilMs: 1000,
|
||||
},
|
||||
tracer: trace.getTracer("test", "0.0.0"),
|
||||
});
|
||||
|
||||
try {
|
||||
const taskIdentifier = "test-task";
|
||||
|
||||
await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);
|
||||
|
||||
const run1 = await engine.trigger(
|
||||
{
|
||||
number: 1,
|
||||
friendlyId: "run_q1",
|
||||
environment: authenticatedEnvironment,
|
||||
taskIdentifier,
|
||||
payload: '{"data": "first"}',
|
||||
payloadType: "application/json",
|
||||
context: {},
|
||||
traceContext: {},
|
||||
traceId: "t_q_1",
|
||||
spanId: "s_q_1",
|
||||
workerQueue: "main",
|
||||
queue: "task/test-task",
|
||||
isTest: false,
|
||||
tags: [],
|
||||
delayUntil: new Date(Date.now() + 5000),
|
||||
debounce: {
|
||||
key: "quantize-key",
|
||||
delay: "5s",
|
||||
},
|
||||
},
|
||||
prisma
|
||||
);
|
||||
|
||||
// Force a meaningfully-later bucket so the second trigger pushes
|
||||
// delayUntil forward through the lock.
|
||||
await prisma.taskRun.update({
|
||||
where: { id: run1.id },
|
||||
data: { delayUntil: new Date(Date.now() - 1000) },
|
||||
});
|
||||
|
||||
await engine.trigger(
|
||||
{
|
||||
number: 2,
|
||||
friendlyId: "run_q2",
|
||||
environment: authenticatedEnvironment,
|
||||
taskIdentifier,
|
||||
payload: '{"data": "second"}',
|
||||
payloadType: "application/json",
|
||||
context: {},
|
||||
traceContext: {},
|
||||
traceId: "t_q_2",
|
||||
spanId: "s_q_2",
|
||||
workerQueue: "main",
|
||||
queue: "task/test-task",
|
||||
isTest: false,
|
||||
tags: [],
|
||||
delayUntil: new Date(Date.now() + 5000),
|
||||
debounce: {
|
||||
key: "quantize-key",
|
||||
delay: "5s",
|
||||
},
|
||||
},
|
||||
prisma
|
||||
);
|
||||
|
||||
const updatedRun = await prisma.taskRun.findFirst({
|
||||
where: { id: run1.id },
|
||||
});
|
||||
assertNonNullable(updatedRun);
|
||||
assertNonNullable(updatedRun.delayUntil);
|
||||
|
||||
// The new delayUntil should be aligned to a 1s bucket boundary.
|
||||
expect(updatedRun.delayUntil.getTime() % 1000).toBe(0);
|
||||
} finally {
|
||||
await engine.quit();
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
containerTest(
|
||||
"Debounce: lock contention falls back to returning existing run",
|
||||
async ({ prisma, redisOptions }) => {
|
||||
const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
|
||||
|
||||
const engine = new RunEngine({
|
||||
prisma,
|
||||
worker: {
|
||||
redis: redisOptions,
|
||||
workers: 1,
|
||||
tasksPerWorker: 10,
|
||||
pollIntervalMs: 100,
|
||||
},
|
||||
queue: {
|
||||
redis: redisOptions,
|
||||
},
|
||||
runLock: {
|
||||
redis: redisOptions,
|
||||
// Force lock acquisition to fail almost immediately so we can
|
||||
// exercise the contention safety net deterministically.
|
||||
retryConfig: {
|
||||
maxAttempts: 0,
|
||||
baseDelay: 1,
|
||||
maxDelay: 1,
|
||||
maxTotalWaitTime: 1,
|
||||
},
|
||||
duration: 30_000,
|
||||
},
|
||||
machines: {
|
||||
defaultMachine: "small-1x",
|
||||
machines: {
|
||||
"small-1x": {
|
||||
name: "small-1x" as const,
|
||||
cpu: 0.5,
|
||||
memory: 0.5,
|
||||
centsPerMs: 0.0001,
|
||||
},
|
||||
},
|
||||
baseCostInCents: 0.0001,
|
||||
},
|
||||
debounce: {
|
||||
maxDebounceDurationMs: 60_000,
|
||||
// Disable fast-path so the request is forced through the lock and
|
||||
// we can prove the contention fallback handles 5xx prevention.
|
||||
fastPathSkipEnabled: false,
|
||||
quantizeNewDelayUntilMs: 0,
|
||||
},
|
||||
tracer: trace.getTracer("test", "0.0.0"),
|
||||
});
|
||||
|
||||
try {
|
||||
const taskIdentifier = "test-task";
|
||||
|
||||
await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);
|
||||
|
||||
const run1 = await engine.trigger(
|
||||
{
|
||||
number: 1,
|
||||
friendlyId: "run_lc1",
|
||||
environment: authenticatedEnvironment,
|
||||
taskIdentifier,
|
||||
payload: '{"data": "first"}',
|
||||
payloadType: "application/json",
|
||||
context: {},
|
||||
traceContext: {},
|
||||
traceId: "t_lc_1",
|
||||
spanId: "s_lc_1",
|
||||
workerQueue: "main",
|
||||
queue: "task/test-task",
|
||||
isTest: false,
|
||||
tags: [],
|
||||
delayUntil: new Date(Date.now() + 5000),
|
||||
debounce: {
|
||||
key: "contention-key",
|
||||
delay: "5s",
|
||||
},
|
||||
},
|
||||
prisma
|
||||
);
|
||||
|
||||
// Hold the underlying redlock key from a separate Redis connection so
|
||||
// the engine's runLock cannot acquire it. Since we configured
|
||||
// `retryConfig.maxAttempts: 0` and `maxTotalWaitTime: 1`, the second
|
||||
// trigger should hit the contention fallback rather than bubble a 5xx.
|
||||
// Note: the prefix template here intentionally matches what the engine
|
||||
// builds at index.ts:120 (no `?? ""` fallback) so that the keys line up
|
||||
// even when redisOptions.keyPrefix is undefined.
|
||||
const blockingRedis = createRedisClient({
|
||||
...redisOptions,
|
||||
keyPrefix: `${redisOptions.keyPrefix}runlock:`,
|
||||
});
|
||||
|
||||
const originalDelayUntil = run1.delayUntil;
|
||||
assertNonNullable(originalDelayUntil);
|
||||
|
||||
try {
|
||||
const blockResult = await blockingRedis.set(
|
||||
run1.id,
|
||||
"test-blocker",
|
||||
"PX",
|
||||
30_000,
|
||||
"NX"
|
||||
);
|
||||
expect(blockResult).toBe("OK");
|
||||
|
||||
const run2 = await engine.trigger(
|
||||
{
|
||||
number: 2,
|
||||
friendlyId: "run_lc2",
|
||||
environment: authenticatedEnvironment,
|
||||
taskIdentifier,
|
||||
payload: '{"data": "second"}',
|
||||
payloadType: "application/json",
|
||||
context: {},
|
||||
traceContext: {},
|
||||
traceId: "t_lc_2",
|
||||
spanId: "s_lc_2",
|
||||
workerQueue: "main",
|
||||
queue: "task/test-task",
|
||||
isTest: false,
|
||||
tags: [],
|
||||
delayUntil: new Date(Date.now() + 5000),
|
||||
debounce: {
|
||||
key: "contention-key",
|
||||
delay: "5s",
|
||||
},
|
||||
},
|
||||
prisma
|
||||
);
|
||||
|
||||
// We did NOT 5xx; we returned the existing run.
|
||||
expect(run2.id).toBe(run1.id);
|
||||
|
||||
// Prove the fallback actually ran rather than the lock being acquired
|
||||
// normally: the second trigger could not push delayUntil forward
|
||||
// because rescheduling is skipped on contention.
|
||||
const updatedRun = await prisma.taskRun.findFirst({
|
||||
where: { id: run1.id },
|
||||
});
|
||||
assertNonNullable(updatedRun);
|
||||
assertNonNullable(updatedRun.delayUntil);
|
||||
expect(updatedRun.delayUntil.getTime()).toBe(originalDelayUntil.getTime());
|
||||
} finally {
|
||||
await blockingRedis.del(run1.id);
|
||||
await blockingRedis.quit();
|
||||
}
|
||||
} finally {
|
||||
await engine.quit();
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
// Reproduces the hot-key contention from TRI-8758: fires N concurrent
|
||||
// triggers on the same debounce key after the run is already DELAYED.
|
||||
//
|
||||
// - fixed=true: fast-path skip + 1s quantization on. The herd collapses on
|
||||
// the unlocked read and onto the same quantized newDelayUntil, so almost
|
||||
// every call short-circuits and `taskRun.update` is barely written.
|
||||
// - fixed=false: fast-path off and quantization off (closer to the
|
||||
// pre-fix behaviour). The lock-contention fallback (also part of this
|
||||
// PR) still catches herd lock failures; this case validates that even
|
||||
// without the fast-path the system stays correct under stress, just at
|
||||
// higher Redlock cost.
|
||||
for (const fixed of [true, false]) {
|
||||
containerTest(
|
||||
`Debounce hot-key stress (fixed=${fixed}): N concurrent triggers stay correct`,
|
||||
async ({ prisma, redisOptions }) => {
|
||||
const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
|
||||
|
||||
const engine = new RunEngine({
|
||||
prisma,
|
||||
worker: {
|
||||
redis: redisOptions,
|
||||
workers: 1,
|
||||
tasksPerWorker: 10,
|
||||
pollIntervalMs: 100,
|
||||
},
|
||||
queue: {
|
||||
redis: redisOptions,
|
||||
},
|
||||
runLock: {
|
||||
redis: redisOptions,
|
||||
},
|
||||
machines: {
|
||||
defaultMachine: "small-1x",
|
||||
machines: {
|
||||
"small-1x": {
|
||||
name: "small-1x" as const,
|
||||
cpu: 0.5,
|
||||
memory: 0.5,
|
||||
centsPerMs: 0.0001,
|
||||
},
|
||||
},
|
||||
baseCostInCents: 0.0001,
|
||||
},
|
||||
debounce: {
|
||||
maxDebounceDurationMs: 10 * 60_000,
|
||||
fastPathSkipEnabled: fixed,
|
||||
// 1s buckets - same as the real default - or 0 to mimic the
|
||||
// pre-fix behaviour where every concurrent trigger has a slightly
|
||||
// larger newDelayUntil than the last.
|
||||
quantizeNewDelayUntilMs: fixed ? 1000 : 0,
|
||||
},
|
||||
tracer: trace.getTracer("test", "0.0.0"),
|
||||
});
|
||||
|
||||
try {
|
||||
const taskIdentifier = "test-task";
|
||||
await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);
|
||||
|
||||
// Seed the debounce key with an initial run, then push delayUntil far
|
||||
// forward so the herd lands well inside the existing window.
|
||||
const seed = await engine.trigger(
|
||||
{
|
||||
number: 0,
|
||||
friendlyId: "run_stress0",
|
||||
environment: authenticatedEnvironment,
|
||||
taskIdentifier,
|
||||
payload: '{"data": "seed"}',
|
||||
payloadType: "application/json",
|
||||
context: {},
|
||||
traceContext: {},
|
||||
traceId: "t_stress_seed",
|
||||
spanId: "s_stress_seed",
|
||||
workerQueue: "main",
|
||||
queue: "task/test-task",
|
||||
isTest: false,
|
||||
tags: [],
|
||||
delayUntil: new Date(Date.now() + 30_000),
|
||||
debounce: {
|
||||
key: "stress-key",
|
||||
delay: "30s",
|
||||
},
|
||||
},
|
||||
prisma
|
||||
);
|
||||
|
||||
// Move delayUntil to a small but safe future offset. The herd's
|
||||
// newDelayUntil (now + 30s) will be meaningfully later than the
|
||||
// current value, so the fast-path-off branch reschedules. The
|
||||
// ~2s buffer keeps the run DELAYED long enough to absorb startup
|
||||
// jitter before the first trigger writes delayUntil = now + 30s.
|
||||
await prisma.taskRun.update({
|
||||
where: { id: seed.id },
|
||||
data: { delayUntil: new Date(Date.now() + 2_000) },
|
||||
});
|
||||
|
||||
// Subscribe to `runDelayRescheduled` so we can count how many times
|
||||
// the herd actually pushed `delayUntil` forward. Each event corresponds
|
||||
// to a successful reschedule under the lock - the fast-path/contention
|
||||
// fallback paths skip the reschedule entirely. We use the engine's
|
||||
// public eventBus, which is the same observable interface other tests
|
||||
// in this repo (ttl, trigger, cancelling, waitpoints) use.
|
||||
let rescheduleCount = 0;
|
||||
engine.eventBus.on("runDelayRescheduled", () => {
|
||||
rescheduleCount++;
|
||||
});
|
||||
|
||||
const N = 40;
|
||||
const triggers = Array.from({ length: N }, (_, i) =>
|
||||
engine.trigger(
|
||||
{
|
||||
number: i + 1,
|
||||
friendlyId: `run_stress${i + 1}`,
|
||||
environment: authenticatedEnvironment,
|
||||
taskIdentifier,
|
||||
payload: `{"data": "stress-${i}"}`,
|
||||
payloadType: "application/json",
|
||||
context: {},
|
||||
traceContext: {},
|
||||
traceId: `t_stress_${i}`,
|
||||
spanId: `s_stress_${i}`,
|
||||
workerQueue: "main",
|
||||
queue: "task/test-task",
|
||||
isTest: false,
|
||||
tags: [],
|
||||
delayUntil: new Date(Date.now() + 30_000),
|
||||
debounce: {
|
||||
key: "stress-key",
|
||||
delay: "30s",
|
||||
},
|
||||
},
|
||||
prisma
|
||||
)
|
||||
);
|
||||
|
||||
const start = performance.now();
|
||||
const settled = await Promise.allSettled(triggers);
|
||||
const durationMs = performance.now() - start;
|
||||
|
||||
const fulfilled = settled.filter(
|
||||
(r): r is PromiseFulfilledResult<{ id: string }> => r.status === "fulfilled"
|
||||
);
|
||||
const rejected = settled.filter((r) => r.status === "rejected");
|
||||
|
||||
// No 5xx feedback loop: every concurrent trigger succeeds and
|
||||
// returns the existing run id.
|
||||
expect(rejected).toHaveLength(0);
|
||||
expect(fulfilled).toHaveLength(N);
|
||||
for (const r of fulfilled) {
|
||||
expect(r.value.id).toBe(seed.id);
|
||||
}
|
||||
|
||||
// Only one row, regardless of contention path.
|
||||
const runs = await prisma.taskRun.findMany({
|
||||
where: { taskIdentifier, runtimeEnvironmentId: authenticatedEnvironment.id },
|
||||
});
|
||||
expect(runs.length).toBe(1);
|
||||
|
||||
// Wait briefly for any in-flight reschedule events to flush before
|
||||
// asserting on the count. EventBus emit is synchronous here but
|
||||
// settle a microtask just to be safe.
|
||||
await new Promise((resolve) => setImmediate(resolve));
|
||||
|
||||
console.log(
|
||||
`[stress fixed=${fixed}] N=${N} duration=${durationMs.toFixed(
|
||||
0
|
||||
)}ms reschedules=${rescheduleCount}`
|
||||
);
|
||||
|
||||
if (fixed) {
|
||||
// With fast-path + quantization: the herd collapses onto the
|
||||
// same quantized newDelayUntil. Trigger #1 takes the lock and
|
||||
// pushes delayUntil; every subsequent trigger sees a covering
|
||||
// delayUntil on the unlocked read and short-circuits without
|
||||
// emitting a reschedule. So at most one reschedule fires.
|
||||
expect(rescheduleCount).toBeLessThanOrEqual(1);
|
||||
}
|
||||
} finally {
|
||||
await engine.quit();
|
||||
}
|
||||
}
|
||||
);
|
||||
}
|
||||
});
|
||||
|
||||
|
||||
@@ -129,6 +129,42 @@ export type RunEngineOptions = {
|
||||
redis?: RedisOptions;
|
||||
/** Maximum duration in milliseconds that a run can be debounced. Default: 1 hour */
|
||||
maxDebounceDurationMs?: number;
|
||||
/**
|
||||
* Bucket size in milliseconds used to quantize the newly computed `delayUntil`.
|
||||
* Quantization collapses many concurrent triggers on the same hot debounce key
|
||||
* into the same target time, so that the unlocked fast-path skip becomes
|
||||
* effective and the redlock on `handleDebounce` is not contended.
|
||||
*
|
||||
* A run might fire up to `quantizeNewDelayUntilMs` earlier than the strict
|
||||
* `now + delay` spec.
|
||||
*
|
||||
* Set to 0 to disable quantization.
|
||||
*
|
||||
* Default: 1000 (1s).
|
||||
*/
|
||||
quantizeNewDelayUntilMs?: number;
|
||||
/**
|
||||
* Whether to read the existing run's `delayUntil` outside of the redlock and
|
||||
* short-circuit when the new (quantized) `delayUntil` is not later than the
|
||||
* current one. Trailing-mode triggers carrying `updateData` always bypass
|
||||
* this fast path and take the lock so payload/metadata/tag updates still
|
||||
* land on the run.
|
||||
*
|
||||
* Default: true.
|
||||
*/
|
||||
fastPathSkipEnabled?: boolean;
|
||||
/**
|
||||
* Whether to route the unlocked fast-path reads (probe + full-run fetch)
|
||||
* through `readOnlyPrisma` (e.g. an Aurora reader) instead of the writer.
|
||||
* Safe because debounce is best-effort: a stale `delayUntil` falls
|
||||
* through to the locked path (the locked path re-checks under the lock),
|
||||
* and a stale `status` at worst returns the existing run, which is the
|
||||
* same outcome the caller would see if their trigger had landed a few
|
||||
* hundred ms earlier.
|
||||
*
|
||||
* Default: false.
|
||||
*/
|
||||
useReplicaForFastPathRead?: boolean;
|
||||
};
|
||||
/** If not set then checkpoints won't ever be used */
|
||||
retryWarmStartThresholdMs?: number;
|
||||
|
||||
Reference in New Issue
Block a user