feat(run-engine): trigger tasks pinned to an external deployment id (#4664)

The SDK discovers an external deployment id at runtime (explicit
TRIGGER_EXTERNAL_DEPLOYMENT_ID always; platform commit-SHA variables and
generic fallbacks when TRIGGER_AUTOMATIC_SKEW_VERSION_PROTECTION=1) and
sends it alongside lockToVersion; the server resolves precedence
(version > external id > current). An id held by a deployed deployment
pins the run to that worker; an in-flight or unknown id parks the run in
PENDING_VERSION with the id in TaskRun.annotations, wakes it pinned when
a deployment carrying the id finalizes (ClickHouse candidates, Postgres
authoritative), and expires it after a deadline that re-checks Postgres
before acting. Parking outranks delaying and preserves delayUntil. The
id is projected to ClickHouse task_runs_v2.external_deployment_id during
replication. Redis cache for id-to-worker resolution, guarded
version-aware writes.

Ids are not unique. Several deployments can hold one id - a --force
rebuild is the ordinary way to get there - so resolution always picks
the highest version among the candidates, never the newest by timestamp.
The rule is applied identically on both paths that can bind a run to a
worker: resolveExternalDeployment at trigger time, and
PendingVersionSystem when a landing deployment wakes a parked run.
Version comparison is numeric on the counter half, so 20260807.10
outranks 20260807.9.

A run whose id never lands expires at the deadline with
EXTERNAL_DEPLOYMENT_NOT_FOUND and an error naming the id it waited for,
which is what a failed build or a typo looks like from the caller.
Default deadline is one hour (EXTERNAL_DEPLOYMENT_PARK_DEADLINE_MS).

Debounce registration happens in both the parked and the delayed branch
through one helper, so a debounced run that parks still binds its
debounce key; without it every later trigger for the same key created
another parked run, and all of them executed when the deployment landed.
The two DELAYED-only status checks in DebounceSystem also accept
PENDING_VERSION, without which the lock-contention fallback would
rethrow a 5xx the SDK retries and amplifies, and the fast path would
push every trigger on a parked key through the redlock.

Resolution is skipped in development. A dev environment cannot hold a
WorkerDeployment - trigger dev registers a BackgroundWorker with nothing
behind it, and deploy --env refuses dev - so an external deployment id
there could only ever park, and the parked run then expired against the
dev TTL while a connected dev worker sat idle. The id is still annotated
so the dashboard shows what the app sent (TRI-13000).
This commit is contained in:
Oskar Otwinowski
2026-08-19 17:43:53 +02:00
committed by GitHub
parent 6bfce6387d
commit 8b0385c429
37 changed files with 3005 additions and 72 deletions
@@ -0,0 +1,6 @@
---
"@trigger.dev/core": patch
"@trigger.dev/sdk": patch
---
Pin runs to the deployment your calling code came from, so an old release never triggers tasks from a new one: set `TRIGGER_EXTERNAL_DEPLOYMENT_ID` to the id you deployed with, or `TRIGGER_AUTOMATIC_SKEW_VERSION_PROTECTION=1` to detect the commit automatically on Vercel and most CI systems. Runs triggered before that deployment finishes building wait for it, then start pinned.
+25
View File
@@ -488,6 +488,31 @@ const EnvironmentSchema = z
.default(process.env.REDIS_TLS_DISABLED ?? "false"),
TASK_META_CACHE_CURRENT_ENV_TTL_SECONDS: z.coerce.number().default(86400),
EXTERNAL_DEPLOYMENT_CACHE_REDIS_HOST: z
.string()
.optional()
.transform((v) => v ?? process.env.REDIS_HOST),
EXTERNAL_DEPLOYMENT_CACHE_REDIS_PORT: z.coerce
.number()
.optional()
.transform(
(v) => v ?? (process.env.REDIS_PORT ? parseInt(process.env.REDIS_PORT) : undefined)
),
EXTERNAL_DEPLOYMENT_CACHE_REDIS_USERNAME: z
.string()
.optional()
.transform((v) => v ?? process.env.REDIS_USERNAME),
EXTERNAL_DEPLOYMENT_CACHE_REDIS_PASSWORD: z
.string()
.optional()
.transform((v) => v ?? process.env.REDIS_PASSWORD),
EXTERNAL_DEPLOYMENT_CACHE_REDIS_TLS_DISABLED: z
.string()
.default(process.env.REDIS_TLS_DISABLED ?? "false"),
EXTERNAL_DEPLOYMENT_CACHE_TTL_SECONDS: z.coerce.number().default(2592000),
EXTERNAL_DEPLOYMENT_CACHE_MISSING_TTL_SECONDS: z.coerce.number().default(20),
EXTERNAL_DEPLOYMENT_PARK_DEADLINE_MS: z.coerce.number().default(3600000),
// Runs-list empty-state check: how far back the ClickHouse "does this env have any run"
// probe looks. Bounds the prove-absence partition scan. 0 = unbounded ("any run ever").
RUN_LIST_HAS_RUNS_LOOKBACK_DAYS: z.coerce.number().default(30),
@@ -74,6 +74,9 @@ import {
import { mollifyTrigger } from "~/v3/mollifier/mollifierMollify.server";
import { QueueSizeLimitExceededError, ServiceValidationError } from "~/v3/services/common.server";
import { runStore } from "~/v3/runStore.server";
import type { ExternalDeploymentCache } from "~/services/externalDeploymentCache.server";
import { externalDeploymentCacheInstance } from "~/services/externalDeploymentCacheInstance.server";
import { resolveExternalDeployment } from "~/v3/services/resolveExternalDeployment.server";
class NoopTriggerRacepointSystem implements TriggerRacepointSystem {
async waitForRacepoint(options: { racepoint: TriggerRacepoints; id: string }): Promise<void> {
@@ -101,6 +104,7 @@ export class RunEngineTriggerTaskService {
private readonly evaluateGate: MollifierEvaluateGate;
private readonly getMollifierBuffer: MollifierGetBuffer;
private readonly isMollifierGloballyEnabled: () => boolean;
private readonly externalDeploymentCache: ExternalDeploymentCache;
constructor(opts: {
prisma: PrismaClientOrTransaction;
@@ -117,6 +121,7 @@ export class RunEngineTriggerTaskService {
evaluateGate?: MollifierEvaluateGate;
getMollifierBuffer?: MollifierGetBuffer;
isMollifierGloballyEnabled?: () => boolean;
externalDeploymentCache?: ExternalDeploymentCache;
}) {
this.prisma = opts.prisma;
this.engine = opts.engine;
@@ -134,6 +139,7 @@ export class RunEngineTriggerTaskService {
this.getMollifierBuffer = opts.getMollifierBuffer ?? defaultGetMollifierBuffer;
this.isMollifierGloballyEnabled =
opts.isMollifierGloballyEnabled ?? (() => env.TRIGGER_MOLLIFIER_ENABLED === "1");
this.externalDeploymentCache = opts.externalDeploymentCache ?? externalDeploymentCacheInstance;
}
/**
@@ -394,7 +400,7 @@ export class RunEngineTriggerTaskService {
});
}
const lockedToBackgroundWorker = body.options?.lockToVersion
const explicitlyLockedToBackgroundWorker = body.options?.lockToVersion
? await this.prisma.backgroundWorker.findFirst({
where: {
projectId: environment.projectId,
@@ -410,6 +416,34 @@ export class RunEngineTriggerTaskService {
})
: undefined;
const externalDeploymentId = body.options?.lockToVersion
? undefined
: body.options?.externalDeploymentId;
const externalDeploymentResolution =
externalDeploymentId && environment.type !== "DEVELOPMENT"
? await resolveExternalDeployment({
prisma: this.prisma,
environmentId: environment.id,
externalDeploymentId,
cache: this.externalDeploymentCache,
})
: undefined;
const lockedToBackgroundWorker =
explicitlyLockedToBackgroundWorker ??
(externalDeploymentResolution?.outcome === "deployed"
? {
id: externalDeploymentResolution.worker.workerId,
version: externalDeploymentResolution.worker.version,
sdkVersion: externalDeploymentResolution.worker.sdkVersion,
cliVersion: externalDeploymentResolution.worker.cliVersion,
}
: undefined);
const parkedOnExternalDeploymentId =
externalDeploymentResolution?.outcome === "park" ? externalDeploymentId : undefined;
const { queueName, lockedQueueId, taskTtl, taskKind } =
await this.queueConcern.resolveQueueProperties(
triggerRequest,
@@ -503,6 +537,7 @@ export class RunEngineTriggerTaskService {
rootTriggerSource: parentAnnotations?.rootTriggerSource ?? triggerSource,
rootScheduleId: parentAnnotations?.rootScheduleId || options.scheduleId || undefined,
taskKind: taskKind ?? "STANDARD",
externalDeploymentId,
};
// Route runs in a scheduled lineage (the scheduled run itself and every
@@ -638,6 +673,7 @@ export class RunEngineTriggerTaskService {
depth,
parentRun: parentRun ?? undefined,
annotations,
parkedOnExternalDeploymentId,
planType,
taskId,
payloadPacket,
@@ -717,6 +753,7 @@ export class RunEngineTriggerTaskService {
depth,
parentRun: parentRun ?? undefined,
annotations,
parkedOnExternalDeploymentId,
planType,
taskId,
payloadPacket,
@@ -892,7 +929,9 @@ export class RunEngineTriggerTaskService {
triggerAction: string;
rootTriggerSource: string;
rootScheduleId?: string | undefined;
externalDeploymentId?: string | undefined;
};
parkedOnExternalDeploymentId?: string;
planType?: string;
taskId: string;
payloadPacket: { data?: string; dataType: string };
@@ -974,6 +1013,7 @@ export class RunEngineTriggerTaskService {
streamBasinName: args.environment.organization.streamBasinName,
debounce: removeNullBytesFromKey(args.body.options?.debounce),
annotations: args.annotations,
parkedOnExternalDeploymentId: args.parkedOnExternalDeploymentId,
};
}
@@ -0,0 +1,214 @@
import type { Callback, Redis, Result } from "ioredis";
import { logger } from "./logger.server";
export type ExternalDeploymentCacheEntry = {
workerId: string;
version: string;
sdkVersion: string;
cliVersion: string;
};
export type ExternalDeploymentCacheResult =
| { outcome: "deployed"; entry: ExternalDeploymentCacheEntry }
| { outcome: "missing" };
export interface ExternalDeploymentCache {
get(environmentId: string, externalId: string): Promise<ExternalDeploymentCacheResult | null>;
setIfNewer(
environmentId: string,
externalId: string,
entry: ExternalDeploymentCacheEntry
): Promise<void>;
setMissing(environmentId: string, externalId: string): Promise<void>;
}
const KEY_PREFIX = "skewid:";
const DEFAULT_TTL_SECONDS = 30 * 24 * 60 * 60;
const DEFAULT_MISSING_TTL_SECONDS = 20;
const MISSING_ENTRY = JSON.stringify({ m: 1 });
function buildKey(environmentId: string, externalId: string): string {
return `${KEY_PREFIX}${environmentId}:${externalId}`;
}
type CachedEntry = {
w: string;
v: string;
s: string;
c: string;
};
function encode(entry: ExternalDeploymentCacheEntry): string {
return JSON.stringify({
w: entry.workerId,
v: entry.version,
s: entry.sdkVersion,
c: entry.cliVersion,
} satisfies CachedEntry);
}
function decode(raw: string): ExternalDeploymentCacheResult | null {
const parsed: unknown = JSON.parse(raw);
if (typeof parsed !== "object" || parsed === null) {
return null;
}
const { w, v, s, c, m } = parsed as Partial<CachedEntry> & { m?: unknown };
if (m === 1) {
return { outcome: "missing" };
}
if (typeof w !== "string" || typeof v !== "string") {
return null;
}
return {
outcome: "deployed",
entry: {
workerId: w,
version: v,
sdkVersion: typeof s === "string" ? s : "",
cliVersion: typeof c === "string" ? c : "",
},
};
}
const SET_IF_NEWER_LUA = `
local existing = redis.call("GET", KEYS[1])
if existing then
local ok, decoded = pcall(cjson.decode, existing)
if ok and type(decoded) == "table" and type(decoded.v) == "string" then
local existingDate, existingCounter = string.match(decoded.v, "^([^.]*)%.?(.*)$")
local incomingDate, incomingCounter = string.match(ARGV[2], "^([^.]*)%.?(.*)$")
if existingDate > incomingDate then
return 0
end
if existingDate == incomingDate then
if (tonumber(existingCounter) or 0) >= (tonumber(incomingCounter) or 0) then
return 0
end
end
end
end
redis.call("SET", KEYS[1], ARGV[1], "EX", tonumber(ARGV[3]))
return 1
`;
declare module "ioredis" {
interface RedisCommander<Context> {
skewIdSetIfNewer(
key: string,
entry: string,
version: string,
ttlSeconds: string,
callback?: Callback<number>
): Result<number, Context>;
}
}
export type RedisExternalDeploymentCacheOptions = {
redis: Redis;
ttlSeconds?: number;
missingTtlSeconds?: number;
};
export class RedisExternalDeploymentCache implements ExternalDeploymentCache {
private readonly redis: Redis;
private readonly ttlSeconds: number;
private readonly missingTtlSeconds: number;
constructor(options: RedisExternalDeploymentCacheOptions) {
this.redis = options.redis;
this.ttlSeconds = options.ttlSeconds ?? DEFAULT_TTL_SECONDS;
this.missingTtlSeconds = options.missingTtlSeconds ?? DEFAULT_MISSING_TTL_SECONDS;
this.redis.defineCommand("skewIdSetIfNewer", { numberOfKeys: 1, lua: SET_IF_NEWER_LUA });
}
async get(
environmentId: string,
externalId: string
): Promise<ExternalDeploymentCacheResult | null> {
try {
const raw = await this.redis.get(buildKey(environmentId, externalId));
if (!raw) return null;
return decode(raw);
} catch (error) {
logger.error("Failed to read external deployment resolution from cache", {
environmentId,
externalId,
error,
});
return null;
}
}
async setIfNewer(
environmentId: string,
externalId: string,
entry: ExternalDeploymentCacheEntry
): Promise<void> {
try {
await this.redis.skewIdSetIfNewer(
buildKey(environmentId, externalId),
encode(entry),
entry.version,
String(this.ttlSeconds)
);
} catch (error) {
logger.error("Failed to write external deployment resolution to cache", {
environmentId,
externalId,
version: entry.version,
error,
});
try {
await this.redis.del(buildKey(environmentId, externalId));
} catch (deleteError) {
logger.error("Failed to evict stale external deployment resolution after write failure", {
environmentId,
externalId,
error: deleteError,
});
}
}
}
async setMissing(environmentId: string, externalId: string): Promise<void> {
try {
await this.redis.set(
buildKey(environmentId, externalId),
MISSING_ENTRY,
"EX",
this.missingTtlSeconds,
"NX"
);
} catch (error) {
logger.error("Failed to write missing external deployment marker to cache", {
environmentId,
externalId,
error,
});
}
}
}
export class NoopExternalDeploymentCache implements ExternalDeploymentCache {
async get(): Promise<ExternalDeploymentCacheResult | null> {
return null;
}
async setIfNewer(): Promise<void> {}
async setMissing(): Promise<void> {}
}
@@ -0,0 +1,38 @@
import { defaultReconnectOnError } from "@internal/redis";
import Redis from "ioredis";
import { env } from "~/env.server";
import { singleton } from "~/utils/singleton";
import {
type ExternalDeploymentCache,
NoopExternalDeploymentCache,
RedisExternalDeploymentCache,
} from "./externalDeploymentCache.server";
export const externalDeploymentCacheInstance: ExternalDeploymentCache = singleton(
"externalDeploymentCacheInstance",
initializeExternalDeploymentCache
);
function initializeExternalDeploymentCache(): ExternalDeploymentCache {
if (!env.EXTERNAL_DEPLOYMENT_CACHE_REDIS_HOST) {
return new NoopExternalDeploymentCache();
}
const redis = new Redis({
connectionName: "externalDeploymentCache",
host: env.EXTERNAL_DEPLOYMENT_CACHE_REDIS_HOST,
port: env.EXTERNAL_DEPLOYMENT_CACHE_REDIS_PORT,
username: env.EXTERNAL_DEPLOYMENT_CACHE_REDIS_USERNAME,
password: env.EXTERNAL_DEPLOYMENT_CACHE_REDIS_PASSWORD,
keyPrefix: "tr:",
enableAutoPipelining: true,
reconnectOnError: defaultReconnectOnError,
...(env.EXTERNAL_DEPLOYMENT_CACHE_REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
});
return new RedisExternalDeploymentCache({
redis,
ttlSeconds: env.EXTERNAL_DEPLOYMENT_CACHE_TTL_SECONDS,
missingTtlSeconds: env.EXTERNAL_DEPLOYMENT_CACHE_MISSING_TTL_SECONDS,
});
}
@@ -36,6 +36,7 @@ import {
} from "@trigger.dev/core/v3/serverOnly";
import { RunAnnotations } from "@trigger.dev/core/v3";
import { type TaskRun } from "@trigger.dev/database";
import { PARKED_ON_EXTERNAL_DEPLOYMENT_STATUS_REASON } from "@internal/run-engine";
import { nanoid } from "nanoid";
import EventEmitter from "node:events";
import pLimit from "p-limit";
@@ -1356,6 +1357,7 @@ export class RunsReplicationService {
annotations?.rootTriggerSource ?? "", // root_trigger_source
annotations?.taskKind ?? "", // task_kind
run.isWarmStart ?? null, // is_warm_start
this.#readExternalDeploymentId(run),
];
}
@@ -1363,6 +1365,22 @@ export class RunsReplicationService {
return RunAnnotations.safeParse(annotations).data;
}
#readExternalDeploymentId(run: TaskRun): string {
if (run.statusReason !== PARKED_ON_EXTERNAL_DEPLOYMENT_STATUS_REASON) {
return "";
}
const annotations = run.annotations;
if (typeof annotations !== "object" || annotations === null) {
return "";
}
const value = (annotations as Record<string, unknown>).externalDeploymentId;
return typeof value === "string" ? value : "";
}
async #preparePayloadInsert(run: TaskRun, _version: bigint): Promise<PayloadInsertArray> {
const payload = await this.#prepareJson(run.payload, run.payloadType);
+1
View File
@@ -160,6 +160,7 @@ function createRunEngine() {
},
retryWarmStartThresholdMs: env.RUN_ENGINE_RETRY_WARM_START_THRESHOLD_MS,
pendingVersionRunIdLookup: runEnginePendingVersionLookup,
externalDeploymentParkDeadlineMs: env.EXTERNAL_DEPLOYMENT_PARK_DEADLINE_MS,
billing: {
getCurrentPlan: async (orgId: string) => {
const plan = await getCurrentPlan(orgId);
@@ -76,9 +76,18 @@ export class ClickhousePendingVersionLookup implements PendingVersionRunIdLookup
taskIdentifiers: options.taskIdentifiers,
})
.where("queue IN {queues: Array(String)}", { queues: options.queues })
.where("_is_deleted = 0")
.orderBy("created_at ASC")
.limit(options.limit);
.where("_is_deleted = 0");
if (options.externalDeploymentId) {
builder.where(
"(external_deployment_id = '' OR external_deployment_id = {externalDeploymentId: String})",
{ externalDeploymentId: options.externalDeploymentId }
);
} else {
builder.where("external_deployment_id = ''");
}
builder.orderBy("created_at ASC").limit(options.limit);
const [queryError, rows] = await builder.execute();
@@ -12,6 +12,7 @@ import { DeploymentService } from "./deployment.server";
import { recordDeploymentOutcome } from "./recordDeploymentOutcome.server";
import { engine } from "../runEngine.server";
import { tryCatch } from "@trigger.dev/core";
import { externalDeploymentCacheInstance } from "~/services/externalDeploymentCacheInstance.server";
export class FinalizeDeploymentService extends BaseService {
public async call(
@@ -128,6 +129,21 @@ export class FinalizeDeploymentService extends BaseService {
logger.error("Failed to publish WORKER_CREATED event", { err });
}
if (deployment.externalId) {
const [cacheError] = await tryCatch(
externalDeploymentCacheInstance.setIfNewer(authenticatedEnv.id, deployment.externalId, {
workerId: deployment.worker.id,
version: deployment.worker.version,
sdkVersion: deployment.worker.sdkVersion ?? "",
cliVersion: deployment.worker.cliVersion ?? "",
})
);
if (cacheError) {
logger.error("Error caching external deployment resolution", { error: cacheError });
}
}
if (deployment.worker.engine === "V2") {
const [schedulePendingVersionsError] = await tryCatch(
engine.scheduleEnqueueRunsForBackgroundWorker(deployment.worker.id)
@@ -0,0 +1,108 @@
import type { PrismaClientOrTransaction } from "@trigger.dev/database";
import type {
ExternalDeploymentCache,
ExternalDeploymentCacheEntry,
} from "~/services/externalDeploymentCache.server";
import { compareDeploymentVersions } from "../utils/deploymentVersions";
const MAX_CANDIDATES = 20;
export type ExternalDeploymentResolution =
| { outcome: "deployed"; worker: ExternalDeploymentCacheEntry }
| { outcome: "park" };
export type ResolveExternalDeploymentOptions = {
prisma: PrismaClientOrTransaction;
environmentId: string;
externalDeploymentId: string;
cache: ExternalDeploymentCache;
};
export async function resolveExternalDeployment({
prisma,
environmentId,
externalDeploymentId,
cache,
}: ResolveExternalDeploymentOptions): Promise<ExternalDeploymentResolution> {
const cached = await cache.get(environmentId, externalDeploymentId);
if (cached?.outcome === "deployed") {
return { outcome: "deployed", worker: cached.entry };
}
if (cached?.outcome === "missing") {
return { outcome: "park" };
}
const worker = await findDeployedWorkerForExternalId({
prisma,
environmentId,
externalDeploymentId,
});
if (!worker) {
await cache.setMissing(environmentId, externalDeploymentId);
return { outcome: "park" };
}
await cache.setIfNewer(environmentId, externalDeploymentId, worker);
return { outcome: "deployed", worker };
}
type FindDeployedWorkerOptions = {
prisma: PrismaClientOrTransaction;
environmentId: string;
externalDeploymentId: string;
};
async function findDeployedWorkerForExternalId({
prisma,
environmentId,
externalDeploymentId,
}: FindDeployedWorkerOptions): Promise<ExternalDeploymentCacheEntry | undefined> {
const candidates = await prisma.workerDeployment.findMany({
where: {
environmentId,
externalId: externalDeploymentId,
status: "DEPLOYED",
},
select: {
version: true,
worker: {
select: {
id: true,
version: true,
sdkVersion: true,
cliVersion: true,
},
},
},
orderBy: { id: "desc" },
take: MAX_CANDIDATES,
});
let highest: { version: string; worker: ExternalDeploymentCacheEntry } | undefined;
for (const candidate of candidates) {
if (!candidate.worker) {
continue;
}
if (highest && compareDeploymentVersions(candidate.version, highest.version) <= 0) {
continue;
}
highest = {
version: candidate.version,
worker: {
workerId: candidate.worker.id,
version: candidate.worker.version,
sdkVersion: candidate.worker.sdkVersion ?? "",
cliVersion: candidate.worker.cliVersion ?? "",
},
};
}
return highest?.worker;
}
+1 -28
View File
@@ -1,28 +1 @@
// Compares two versions of a deployment, like 20250208.1 and 20250208.2
// Returns -1 if versionA is older than versionB, 0 if they are the same, and 1 if versionA is newer than versionB
export function compareDeploymentVersions(versionA: string, versionB: string) {
const [dateA, numberA] = versionA.split(".");
const [dateB, numberB] = versionB.split(".");
if (dateA < dateB) {
return -1;
}
if (dateA > dateB) {
return 1;
}
// Convert to numbers before comparing
const numA = Number(numberA);
const numB = Number(numberB);
if (numA < numB) {
return -1;
}
if (numA > numB) {
return 1;
}
return 0;
}
export { compareDeploymentVersions } from "@trigger.dev/core/v3/isomorphic";
@@ -0,0 +1,484 @@
import { describe, expect, onTestFinished, vi } from "vitest";
vi.mock("~/db.server", () => ({
prisma: {},
$replica: {},
runOpsNewPrisma: {},
runOpsLegacyPrisma: {},
}));
vi.mock("~/v3/runOpsMigration/splitMode.server", () => ({ isSplitEnabled: async () => false }));
vi.mock("~/services/platform.v3.server", async (importOriginal) => {
const actual = (await importOriginal()) as Record<string, unknown>;
return {
...actual,
getEntitlement: vi.fn(),
};
});
import { RunEngine } from "@internal/run-engine";
import { setupAuthenticatedEnvironment, setupBackgroundWorker } from "@internal/run-engine/tests";
import { assertNonNullable, containerTest } from "@internal/testcontainers";
import { trace } from "@opentelemetry/api";
import type { PrismaClient } from "@trigger.dev/database";
import { IdempotencyKeyConcern } from "~/runEngine/concerns/idempotencyKeys.server";
import { DefaultQueueManager } from "~/runEngine/concerns/queues.server";
import {
type ExternalDeploymentCache,
type ExternalDeploymentCacheEntry,
NoopExternalDeploymentCache,
} from "~/services/externalDeploymentCache.server";
import { RunEngineTriggerTaskService } from "../../app/runEngine/services/triggerTask.server";
import {
MockPayloadProcessor,
MockTraceEventConcern,
MockTriggerTaskValidator,
} from "./triggerTaskTestHelpers";
vi.setConfig({ testTimeout: 60_000, hookTimeout: 60_000 });
class RecordingExternalDeploymentCache implements ExternalDeploymentCache {
readonly gets: Array<{ environmentId: string; externalId: string }> = [];
readonly writes: Array<{ externalId: string; entry: ExternalDeploymentCacheEntry }> = [];
constructor(private readonly entries = new Map<string, ExternalDeploymentCacheEntry>()) {}
readonly missing: string[] = [];
async get(environmentId: string, externalId: string) {
this.gets.push({ environmentId, externalId });
const entry = this.entries.get(externalId);
if (entry) {
return { outcome: "deployed" as const, entry };
}
return this.missing.includes(externalId) ? { outcome: "missing" as const } : null;
}
async setIfNewer(
_environmentId: string,
externalId: string,
entry: ExternalDeploymentCacheEntry
) {
this.writes.push({ externalId, entry });
this.entries.set(externalId, entry);
}
async setMissing(_environmentId: string, externalId: string) {
this.missing.push(externalId);
}
}
function createEngine(prisma: PrismaClient, redisOptions: unknown) {
const engine = new RunEngine({
prisma,
// eslint-disable-next-line @typescript-eslint/no-explicit-any
worker: { redis: redisOptions as any, workers: 1, tasksPerWorker: 10, pollIntervalMs: 100 },
// eslint-disable-next-line @typescript-eslint/no-explicit-any
queue: { redis: redisOptions as any },
// eslint-disable-next-line @typescript-eslint/no-explicit-any
runLock: { redis: redisOptions as any },
machines: {
defaultMachine: "small-1x",
machines: {
"small-1x": { name: "small-1x", cpu: 0.5, memory: 0.5, centsPerMs: 0.0001 },
},
baseCostInCents: 0.0005,
},
tracer: trace.getTracer("test", "0.0.0"),
});
return engine;
}
function createService(
prisma: PrismaClient,
engine: RunEngine,
externalDeploymentCache: ExternalDeploymentCache
) {
return new RunEngineTriggerTaskService({
engine,
prisma,
payloadProcessor: new MockPayloadProcessor(),
queueConcern: new DefaultQueueManager(prisma, engine),
idempotencyKeyConcern: new IdempotencyKeyConcern(prisma, engine, new MockTraceEventConcern()),
validator: new MockTriggerTaskValidator(),
traceEventConcern: new MockTraceEventConcern(),
tracer: trace.getTracer("test", "0.0.0"),
metadataMaximumSize: 1024 * 1024,
externalDeploymentCache,
});
}
async function nameDeploymentWithExternalId(
prisma: PrismaClient,
workerId: string,
externalId: string
) {
await prisma.workerDeployment.update({ where: { workerId }, data: { externalId } });
}
describe("triggerTask external deployment id", () => {
containerTest(
"pins the run to the deployment holding the id, not to whatever is current",
async ({ prisma, redisOptions }) => {
const engine = createEngine(prisma, redisOptions);
onTestFinished(() => engine.quit());
const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
const taskIdentifier = "pinned-task";
const targetWorker = await setupBackgroundWorker(engine, environment, taskIdentifier);
await nameDeploymentWithExternalId(prisma, targetWorker.worker.id, "commit-target");
const currentWorker = await setupBackgroundWorker(engine, environment, taskIdentifier);
const cache = new RecordingExternalDeploymentCache();
const service = createService(prisma, engine, cache);
const result = await service.call({
taskId: taskIdentifier,
environment,
body: { payload: { test: "x" }, options: { externalDeploymentId: "commit-target" } },
});
assertNonNullable(result);
const run = await prisma.taskRun.findFirstOrThrow({ where: { id: result.run.id } });
expect(run.status).toBe("PENDING");
expect(run.lockedToVersionId).toBe(targetWorker.worker.id);
expect(run.taskVersion).toBe(targetWorker.worker.version);
expect(run.lockedToVersionId).not.toBe(currentWorker.worker.id);
expect((run.annotations as Record<string, unknown>).externalDeploymentId).toBe(
"commit-target"
);
expect(cache.writes.map((w) => w.externalId)).toEqual(["commit-target"]);
}
);
containerTest(
"an explicit version wins over an external deployment id, and the id is not even resolved",
async ({ prisma, redisOptions }) => {
const engine = createEngine(prisma, redisOptions);
onTestFinished(() => engine.quit());
const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
const taskIdentifier = "precedence-task";
const versionWorker = await setupBackgroundWorker(engine, environment, taskIdentifier);
const idWorker = await setupBackgroundWorker(engine, environment, taskIdentifier);
await nameDeploymentWithExternalId(prisma, idWorker.worker.id, "commit-loser");
const cache = new RecordingExternalDeploymentCache();
const service = createService(prisma, engine, cache);
const result = await service.call({
taskId: taskIdentifier,
environment,
body: {
payload: { test: "x" },
options: {
lockToVersion: versionWorker.worker.version,
externalDeploymentId: "commit-loser",
},
},
});
assertNonNullable(result);
const run = await prisma.taskRun.findFirstOrThrow({ where: { id: result.run.id } });
expect(run.status).toBe("PENDING");
expect(run.lockedToVersionId).toBe(versionWorker.worker.id);
expect(run.taskVersion).toBe(versionWorker.worker.version);
expect(cache.gets).toEqual([]);
expect((run.annotations as Record<string, unknown>).externalDeploymentId).toBeUndefined();
}
);
containerTest(
"a trigger carrying no id runs on current, exactly as before",
async ({ prisma, redisOptions }) => {
const engine = createEngine(prisma, redisOptions);
onTestFinished(() => engine.quit());
const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
const taskIdentifier = "plain-task";
await setupBackgroundWorker(engine, environment, taskIdentifier);
const cache = new RecordingExternalDeploymentCache();
const service = createService(prisma, engine, cache);
const result = await service.call({
taskId: taskIdentifier,
environment,
body: { payload: { test: "x" } },
});
assertNonNullable(result);
const run = await prisma.taskRun.findFirstOrThrow({ where: { id: result.run.id } });
expect(run.status).toBe("PENDING");
expect(run.lockedToVersionId).toBeNull();
expect(cache.gets).toEqual([]);
}
);
containerTest(
"parks a run whose id nothing holds, recording the id in annotations",
async ({ prisma, redisOptions }) => {
const engine = createEngine(prisma, redisOptions);
onTestFinished(() => engine.quit());
const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
const taskIdentifier = "parked-task";
await setupBackgroundWorker(engine, environment, taskIdentifier);
const service = createService(prisma, engine, new NoopExternalDeploymentCache());
const result = await service.call({
taskId: taskIdentifier,
environment,
body: { payload: { test: "x" }, options: { externalDeploymentId: "commit-unknown" } },
});
assertNonNullable(result);
const run = await prisma.taskRun.findFirstOrThrow({ where: { id: result.run.id } });
expect(run.status).toBe("PENDING_VERSION");
expect(run.statusReason).toBe("EXTERNAL_DEPLOYMENT_PENDING");
expect(run.lockedToVersionId).toBeNull();
expect((run.annotations as Record<string, unknown>).externalDeploymentId).toBe(
"commit-unknown"
);
}
);
containerTest(
"never parks in development, where no deployment can ever hold the id",
async ({ prisma, redisOptions }) => {
const engine = createEngine(prisma, redisOptions);
onTestFinished(() => engine.quit());
const environment = await setupAuthenticatedEnvironment(prisma, "DEVELOPMENT");
const taskIdentifier = "dev-task";
await setupBackgroundWorker(engine, environment, taskIdentifier);
const cache = new RecordingExternalDeploymentCache();
const service = createService(prisma, engine, cache);
const result = await service.call({
taskId: taskIdentifier,
environment,
body: { payload: { test: "x" }, options: { externalDeploymentId: "commit-unknown" } },
});
assertNonNullable(result);
const run = await prisma.taskRun.findFirstOrThrow({ where: { id: result.run.id } });
expect(run.status).toBe("PENDING");
expect(run.statusReason).toBeNull();
expect(cache.gets).toEqual([]);
expect((run.annotations as Record<string, unknown>).externalDeploymentId).toBe(
"commit-unknown"
);
}
);
containerTest(
"parks a run whose id is held only by an in-flight deployment",
async ({ prisma, redisOptions }) => {
const engine = createEngine(prisma, redisOptions);
onTestFinished(() => engine.quit());
const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
const taskIdentifier = "inflight-task";
const worker = await setupBackgroundWorker(engine, environment, taskIdentifier);
await prisma.workerDeployment.update({
where: { workerId: worker.worker.id },
data: { externalId: "commit-building", status: "BUILDING" },
});
const cache = new RecordingExternalDeploymentCache();
const service = createService(prisma, engine, cache);
const result = await service.call({
taskId: taskIdentifier,
environment,
body: { payload: { test: "x" }, options: { externalDeploymentId: "commit-building" } },
});
assertNonNullable(result);
const run = await prisma.taskRun.findFirstOrThrow({ where: { id: result.run.id } });
expect(run.status).toBe("PENDING_VERSION");
expect(run.lockedToVersionId).toBeNull();
expect(cache.writes).toEqual([]);
}
);
containerTest(
"trusts a cache hit without querying Postgres",
async ({ prisma, redisOptions }) => {
const engine = createEngine(prisma, redisOptions);
onTestFinished(() => engine.quit());
const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
const taskIdentifier = "cached-pin-task";
const worker = await setupBackgroundWorker(engine, environment, taskIdentifier);
const cache = new RecordingExternalDeploymentCache(
new Map([
[
"commit-cached",
{
workerId: worker.worker.id,
version: worker.worker.version,
sdkVersion: "",
cliVersion: "",
},
],
])
);
const service = createService(prisma, engine, cache);
const result = await service.call({
taskId: taskIdentifier,
environment,
body: { payload: { test: "x" }, options: { externalDeploymentId: "commit-cached" } },
});
assertNonNullable(result);
const run = await prisma.taskRun.findFirstOrThrow({ where: { id: result.run.id } });
expect(run.status).toBe("PENDING");
expect(run.lockedToVersionId).toBe(worker.worker.id);
expect(cache.gets).toEqual([{ environmentId: environment.id, externalId: "commit-cached" }]);
expect(cache.writes).toEqual([]);
}
);
containerTest(
"resolves to the highest version when several deployed deployments hold the id",
async ({ prisma, redisOptions }) => {
const engine = createEngine(prisma, redisOptions);
onTestFinished(() => engine.quit());
const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
const taskIdentifier = "forced-task";
const older = await setupBackgroundWorker(engine, environment, taskIdentifier);
await prisma.backgroundWorker.update({
where: { id: older.worker.id },
data: { version: "20260807.9" },
});
await prisma.workerDeployment.update({
where: { workerId: older.worker.id },
data: {
externalId: "commit-forced",
version: "20260807.9",
shortCode: "short_code_20260807.9",
},
});
const newer = await setupBackgroundWorker(engine, environment, taskIdentifier);
await prisma.backgroundWorker.update({
where: { id: newer.worker.id },
data: { version: "20260807.10" },
});
await prisma.workerDeployment.update({
where: { workerId: newer.worker.id },
data: {
externalId: "commit-forced",
version: "20260807.10",
shortCode: "short_code_20260807.10",
},
});
const service = createService(prisma, engine, new NoopExternalDeploymentCache());
const result = await service.call({
taskId: taskIdentifier,
environment,
body: { payload: { test: "x" }, options: { externalDeploymentId: "commit-forced" } },
});
assertNonNullable(result);
const run = await prisma.taskRun.findFirstOrThrow({ where: { id: result.run.id } });
expect(run.lockedToVersionId).toBe(newer.worker.id);
expect(run.taskVersion).toBe("20260807.10");
}
);
containerTest(
"an id is environment-scoped, so a deployment in another environment never resolves it",
async ({ prisma, redisOptions }) => {
const engine = createEngine(prisma, redisOptions);
onTestFinished(() => engine.quit());
const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
const taskIdentifier = "scoped-task";
const worker = await setupBackgroundWorker(engine, environment, taskIdentifier);
const otherEnvironment = await prisma.runtimeEnvironment.create({
data: {
slug: "staging-scoped",
type: "STAGING",
projectId: environment.project.id,
organizationId: environment.organization.id,
apiKey: "tr_stg_scoped",
pkApiKey: "pk_stg_scoped",
shortcode: "stg-scoped",
},
});
await prisma.workerDeployment.create({
data: {
friendlyId: "deployment_elsewhere",
contentHash: "hash",
shortCode: "sc_elsewhere",
version: worker.worker.version,
status: "DEPLOYED",
externalId: "commit-elsewhere",
projectId: environment.project.id,
environmentId: otherEnvironment.id,
},
});
const service = createService(prisma, engine, new NoopExternalDeploymentCache());
const result = await service.call({
taskId: taskIdentifier,
environment,
body: { payload: { test: "x" }, options: { externalDeploymentId: "commit-elsewhere" } },
});
assertNonNullable(result);
const run = await prisma.taskRun.findFirstOrThrow({ where: { id: result.run.id } });
expect(run.status).toBe("PENDING_VERSION");
expect(run.lockedToVersionId).toBeNull();
}
);
});
@@ -0,0 +1,188 @@
import { redisTest } from "@internal/testcontainers";
import { Redis } from "ioredis";
import { describe, expect } from "vitest";
import { RedisExternalDeploymentCache } from "~/services/externalDeploymentCache.server";
vi.setConfig({ testTimeout: 30_000 });
function entry(version: string, workerId = `worker_${version}`) {
return { workerId, version, sdkVersion: "4.0.0", cliVersion: "4.0.0" };
}
function deployed(version: string, workerId = `worker_${version}`) {
return { outcome: "deployed", entry: entry(version, workerId) };
}
describe("RedisExternalDeploymentCache", () => {
redisTest("stores and reads back a resolution", async ({ redisOptions }) => {
const redis = new Redis({ ...redisOptions, maxRetriesPerRequest: null });
const cache = new RedisExternalDeploymentCache({ redis });
try {
expect(await cache.get("env_1", "commit-a")).toBeNull();
await cache.setIfNewer("env_1", "commit-a", entry("20260807.1"));
expect(await cache.get("env_1", "commit-a")).toEqual(deployed("20260807.1"));
} finally {
await redis.quit();
}
});
redisTest("scopes entries by environment", async ({ redisOptions }) => {
const redis = new Redis({ ...redisOptions, maxRetriesPerRequest: null });
const cache = new RedisExternalDeploymentCache({ redis });
try {
await cache.setIfNewer("env_1", "commit-a", entry("20260807.1"));
expect(await cache.get("env_2", "commit-a")).toBeNull();
} finally {
await redis.quit();
}
});
redisTest("overwrites when the finalising version is higher", async ({ redisOptions }) => {
const redis = new Redis({ ...redisOptions, maxRetriesPerRequest: null });
const cache = new RedisExternalDeploymentCache({ redis });
try {
await cache.setIfNewer("env_1", "commit-a", entry("20260807.1"));
await cache.setIfNewer("env_1", "commit-a", entry("20260807.2"));
expect(await cache.get("env_1", "commit-a")).toEqual(deployed("20260807.2"));
} finally {
await redis.quit();
}
});
redisTest(
"refuses a lower version — the slower earlier build finalising last must not reinstate the version the operator was replacing",
async ({ redisOptions }) => {
const redis = new Redis({ ...redisOptions, maxRetriesPerRequest: null });
const cache = new RedisExternalDeploymentCache({ redis });
try {
await cache.setIfNewer("env_1", "commit-a", entry("20260807.2"));
await cache.setIfNewer("env_1", "commit-a", entry("20260807.1"));
expect(await cache.get("env_1", "commit-a")).toEqual(deployed("20260807.2"));
} finally {
await redis.quit();
}
}
);
redisTest("refuses an equal version", async ({ redisOptions }) => {
const redis = new Redis({ ...redisOptions, maxRetriesPerRequest: null });
const cache = new RedisExternalDeploymentCache({ redis });
try {
await cache.setIfNewer("env_1", "commit-a", entry("20260807.1", "worker_first"));
await cache.setIfNewer("env_1", "commit-a", entry("20260807.1", "worker_second"));
expect(await cache.get("env_1", "commit-a")).toEqual(deployed("20260807.1", "worker_first"));
} finally {
await redis.quit();
}
});
redisTest(
"compares the counter numerically, not lexicographically — .10 beats .9",
async ({ redisOptions }) => {
const redis = new Redis({ ...redisOptions, maxRetriesPerRequest: null });
const cache = new RedisExternalDeploymentCache({ redis });
try {
await cache.setIfNewer("env_1", "commit-a", entry("20260807.9"));
await cache.setIfNewer("env_1", "commit-a", entry("20260807.10"));
expect(await cache.get("env_1", "commit-a")).toEqual(deployed("20260807.10"));
await cache.setIfNewer("env_1", "commit-a", entry("20260807.9"));
expect(await cache.get("env_1", "commit-a")).toEqual(deployed("20260807.10"));
} finally {
await redis.quit();
}
}
);
redisTest("a later date wins regardless of counter", async ({ redisOptions }) => {
const redis = new Redis({ ...redisOptions, maxRetriesPerRequest: null });
const cache = new RedisExternalDeploymentCache({ redis });
try {
await cache.setIfNewer("env_1", "commit-a", entry("20260807.50"));
await cache.setIfNewer("env_1", "commit-a", entry("20260808.1"));
expect(await cache.get("env_1", "commit-a")).toEqual(deployed("20260808.1"));
await cache.setIfNewer("env_1", "commit-a", entry("20260806.99"));
expect(await cache.get("env_1", "commit-a")).toEqual(deployed("20260808.1"));
} finally {
await redis.quit();
}
});
redisTest("caches a missing resolution with a short TTL", async ({ redisOptions }) => {
const redis = new Redis({ ...redisOptions, maxRetriesPerRequest: null });
const cache = new RedisExternalDeploymentCache({ redis, missingTtlSeconds: 30 });
try {
await cache.setMissing("env_1", "commit-a");
expect(await cache.get("env_1", "commit-a")).toEqual({ outcome: "missing" });
const ttl = await redis.ttl("skewid:env_1:commit-a");
expect(ttl).toBeGreaterThan(0);
expect(ttl).toBeLessThanOrEqual(30);
} finally {
await redis.quit();
}
});
redisTest("a landed deployment replaces a missing marker", async ({ redisOptions }) => {
const redis = new Redis({ ...redisOptions, maxRetriesPerRequest: null });
const cache = new RedisExternalDeploymentCache({ redis });
try {
await cache.setMissing("env_1", "commit-a");
await cache.setIfNewer("env_1", "commit-a", entry("20260807.1"));
expect(await cache.get("env_1", "commit-a")).toEqual(deployed("20260807.1"));
} finally {
await redis.quit();
}
});
redisTest("a missing marker never clobbers a resolution", async ({ redisOptions }) => {
const redis = new Redis({ ...redisOptions, maxRetriesPerRequest: null });
const cache = new RedisExternalDeploymentCache({ redis });
try {
await cache.setIfNewer("env_1", "commit-a", entry("20260807.1"));
await cache.setMissing("env_1", "commit-a");
expect(await cache.get("env_1", "commit-a")).toEqual(deployed("20260807.1"));
} finally {
await redis.quit();
}
});
redisTest("sets a TTL on the entry", async ({ redisOptions }) => {
const redis = new Redis({ ...redisOptions, maxRetriesPerRequest: null });
const cache = new RedisExternalDeploymentCache({ redis, ttlSeconds: 120 });
try {
await cache.setIfNewer("env_1", "commit-a", entry("20260807.1"));
const ttl = await redis.ttl("skewid:env_1:commit-a");
expect(ttl).toBeGreaterThan(0);
expect(ttl).toBeLessThanOrEqual(120);
} finally {
await redis.quit();
}
});
});
@@ -118,6 +118,9 @@ class RoutingRunStore implements RunStore {
promotePendingVersionRuns(runId: string, ...a: any[]): any {
return (this.#resolveById(runId).promotePendingVersionRuns as any)(runId, ...a);
}
expireParkedRun(runId: string, ...a: any[]): any {
return (this.#resolveById(runId).expireParkedRun as any)(runId, ...a);
}
suspendForCheckpoint(runId: string, ...a: any[]): any {
return (this.#resolveById(runId).suspendForCheckpoint as any)(runId, ...a);
}
@@ -0,0 +1,176 @@
import { ClickHouse } from "@internal/clickhouse";
import { replicationContainerTest } from "@internal/testcontainers";
import { z } from "zod";
import { RunsReplicationService } from "~/services/runsReplicationService.server";
import { TestReplicationClickhouseFactory } from "./utils/testReplicationClickhouseFactory";
import { createInMemoryTracing } from "./utils/tracing";
vi.setConfig({ testTimeout: 60_000 });
describe("RunsReplicationService external_deployment_id", () => {
replicationContainerTest(
"projects the external deployment id only for runs parked on it, including when the annotations blob fails RunAnnotations validation",
async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
await prisma.$executeRawUnsafe(`ALTER TABLE public."TaskRun" REPLICA IDENTITY FULL;`);
const clickhouse = new ClickHouse({
url: clickhouseContainer.getConnectionUrl(),
name: "runs-replication",
compression: { request: true },
logLevel: "warn",
});
const { tracer } = createInMemoryTracing();
const runsReplicationService = new RunsReplicationService({
clickhouseFactory: new TestReplicationClickhouseFactory(clickhouse),
pgConnectionUrl: postgresContainer.getConnectionUri(),
serviceName: "runs-replication",
slotName: "task_runs_to_clickhouse_v1",
publicationName: "task_runs_to_clickhouse_v1_publication",
redisOptions,
maxFlushConcurrency: 1,
flushIntervalMs: 100,
flushBatchSize: 10,
leaderLockTimeoutMs: 5000,
leaderLockExtendIntervalMs: 1000,
ackIntervalSeconds: 5,
tracer,
logLevel: "warn",
});
await runsReplicationService.start();
const organization = await prisma.organization.create({
data: { title: "test", slug: "test" },
});
const project = await prisma.project.create({
data: {
name: "test",
slug: "test",
organizationId: organization.id,
externalRef: "test",
},
});
const runtimeEnvironment = await prisma.runtimeEnvironment.create({
data: {
slug: "test",
type: "DEVELOPMENT",
projectId: project.id,
organizationId: organization.id,
apiKey: "test",
pkApiKey: "test",
shortcode: "test",
},
});
const baseRun = {
taskIdentifier: "my-task",
payload: JSON.stringify({ foo: "bar" }),
queue: "test",
runtimeEnvironmentId: runtimeEnvironment.id,
projectId: project.id,
organizationId: organization.id,
environmentType: "DEVELOPMENT" as const,
engine: "V2" as const,
};
const withId = await prisma.taskRun.create({
data: {
...baseRun,
friendlyId: "run_1234",
traceId: "1234",
spanId: "1234",
status: "PENDING_VERSION",
statusReason: "EXTERNAL_DEPLOYMENT_PENDING",
annotations: {
triggerSource: "sdk",
triggerAction: "trigger",
rootTriggerSource: "sdk",
externalDeploymentId: "fa1eade47b73733d6312d5abfad33ce9e4068081",
},
},
});
const withoutId = await prisma.taskRun.create({
data: {
...baseRun,
friendlyId: "run_1235",
traceId: "1235",
spanId: "1235",
annotations: {
triggerSource: "sdk",
triggerAction: "trigger",
rootTriggerSource: "sdk",
},
},
});
const withUnparseableAnnotations = await prisma.taskRun.create({
data: {
...baseRun,
friendlyId: "run_1236",
traceId: "1236",
spanId: "1236",
status: "PENDING_VERSION",
statusReason: "EXTERNAL_DEPLOYMENT_PENDING",
annotations: {
triggerSource: "sdk",
externalDeploymentId: "commit-on-a-broken-blob",
},
},
});
const reportedButParkedForAnotherReason = await prisma.taskRun.create({
data: {
...baseRun,
friendlyId: "run_1237",
traceId: "1237",
spanId: "1237",
status: "PENDING_VERSION",
statusReason: "NO_WORKER",
annotations: {
triggerSource: "sdk",
triggerAction: "trigger",
rootTriggerSource: "sdk",
externalDeploymentId: "commit-reported-not-parked",
},
},
});
const queryRuns = clickhouse.reader.query({
name: "runs-replication",
query: "SELECT run_id, external_deployment_id FROM trigger_dev.task_runs_v2",
schema: z.any(),
});
const rows = await vi.waitFor(
async () => {
const [queryError, result] = await queryRuns({});
expect(queryError).toBeNull();
expect(result?.length).toBe(4);
return result;
},
{ timeout: 30_000, interval: 250 }
);
const byRunId = new Map<string, string>(
(rows ?? []).map((row: { run_id: string; external_deployment_id: string }) => [
row.run_id,
row.external_deployment_id,
])
);
expect(byRunId.get(withId.id)).toBe("fa1eade47b73733d6312d5abfad33ce9e4068081");
expect(byRunId.get(withoutId.id)).toBe("");
expect(byRunId.get(withUnparseableAnnotations.id)).toBe("commit-on-a-broken-blob");
expect(byRunId.get(reportedButParkedForAnotherReason.id)).toBe("");
await runsReplicationService.stop();
}
);
});
@@ -186,8 +186,11 @@ class RoutingRunStore implements RunStore {
parkPendingVersion(runId: string, data: any, args: any, _tx?: unknown): any {
return (this.#resolveById(runId).parkPendingVersion as any)(runId, data, args);
}
promotePendingVersionRuns(runId: string, _tx?: unknown): any {
return this.#resolveById(runId).promotePendingVersionRuns(runId);
promotePendingVersionRuns(runId: string, args?: any, _tx?: unknown): any {
return this.#resolveById(runId).promotePendingVersionRuns(runId, args);
}
expireParkedRun(runId: string, data: any, _tx?: unknown): any {
return (this.#resolveById(runId).expireParkedRun as any)(runId, data);
}
suspendForCheckpoint(runId: string, args: any, _tx?: unknown): any {
return (this.#resolveById(runId).suspendForCheckpoint as any)(runId, args);
@@ -93,6 +93,7 @@ describe("Task Runs V2", () => {
"", // root_trigger_source
"", // task_kind
null, // is_warm_start
"", // external_deployment_id
];
const [insertError, insertResult] = await insert([taskRunData]);
@@ -236,6 +237,7 @@ describe("Task Runs V2", () => {
"", // root_trigger_source
"", // task_kind
null, // is_warm_start
"", // external_deployment_id
];
const [insertError, insertResult] = await insert([taskRunData]);
@@ -336,6 +338,7 @@ describe("Task Runs V2", () => {
"", // root_trigger_source
"", // task_kind
null, // is_warm_start
"", // external_deployment_id
];
const run2: TaskRunInsertArray = [
@@ -394,6 +397,7 @@ describe("Task Runs V2", () => {
"", // root_trigger_source
"", // task_kind
null, // is_warm_start
"", // external_deployment_id
];
const [insertError, insertResult] = await insert([run1, run2]);
@@ -499,6 +503,7 @@ describe("Task Runs V2", () => {
"", // root_trigger_source
"", // task_kind
null, // is_warm_start
"", // external_deployment_id
];
const [_insertError, _insertResult] = await insert([taskRun]);
@@ -612,6 +617,7 @@ describe("Task Runs V2", () => {
"", // root_trigger_source
"", // task_kind
null, // is_warm_start
"", // external_deployment_id
];
const childA_v1: TaskRunInsertArray = [
@@ -670,6 +676,7 @@ describe("Task Runs V2", () => {
"",
"",
null,
"", // external_deployment_id
];
const childA_v2: TaskRunInsertArray = [...childA_v1];
@@ -732,6 +739,7 @@ describe("Task Runs V2", () => {
"",
"",
null,
"", // external_deployment_id
];
const childDeleted_v1: TaskRunInsertArray = [
@@ -790,6 +798,7 @@ describe("Task Runs V2", () => {
"",
"",
null,
"", // external_deployment_id
];
const childDeleted_v2: TaskRunInsertArray = [...childDeleted_v1];
@@ -968,6 +977,7 @@ describe("Task Runs V2", () => {
"",
"",
null,
"", // external_deployment_id
];
const rdsSnapshot: TaskRunInsertArray = [...base];
@@ -1072,6 +1082,7 @@ describe("Task Runs V2", () => {
"",
"",
null,
"", // external_deployment_id
];
const earlier: TaskRunInsertArray = [...base];
@@ -1176,6 +1187,7 @@ describe("Task Runs V2", () => {
"",
"",
null,
"", // external_deployment_id
];
const rdsSnapshot: TaskRunInsertArray = [...base];
@@ -56,6 +56,7 @@ export const TaskRunV2 = z.object({
root_trigger_source: z.string().default(""),
task_kind: z.string().default(""),
is_warm_start: z.boolean().nullish(),
external_deployment_id: z.string().default(""),
_version: z.string(),
_is_deleted: z.number().int().default(0),
});
@@ -119,6 +120,7 @@ export const TASK_RUN_COLUMNS = [
"root_trigger_source",
"task_kind",
"is_warm_start",
"external_deployment_id",
] as const;
export type TaskRunColumnName = (typeof TASK_RUN_COLUMNS)[number];
@@ -189,6 +191,7 @@ export type TaskRunFieldTypes = {
root_trigger_source: string;
task_kind: string;
is_warm_start: boolean | null;
external_deployment_id: string;
};
/**
@@ -358,6 +361,7 @@ export type TaskRunInsertArray = [
root_trigger_source: string,
task_kind: string,
is_warm_start: boolean | null,
external_deployment_id: string,
];
/**
+107 -19
View File
@@ -77,7 +77,10 @@ import {
getExecutionSnapshotsSince,
getLatestExecutionSnapshot,
} from "./systems/executionSnapshotSystem.js";
import { PendingVersionSystem } from "./systems/pendingVersionSystem.js";
import {
PARKED_ON_EXTERNAL_DEPLOYMENT_STATUS_REASON,
PendingVersionSystem,
} from "./systems/pendingVersionSystem.js";
import { RaceSimulationSystem } from "./systems/raceSimulationSystem.js";
import { RunAttemptSystem } from "./systems/runAttemptSystem.js";
import { NoopPendingVersionRunIdLookup } from "./services/pendingVersionLookup.js";
@@ -286,6 +289,12 @@ export class RunEngine {
payload.attempt
);
},
expireParkedExternalDeploymentRun: async ({ payload }) => {
await this.pendingVersionSystem.expireParkedExternalDeploymentRun({
runId: payload.runId,
externalDeploymentId: payload.externalDeploymentId,
});
},
tryCompleteBatch: async ({ payload }) => {
await this.batchSystem.performCompleteBatch({ batchId: payload.batchId });
},
@@ -397,6 +406,7 @@ export class RunEngine {
queueRunsPendingVersionBatchSize: options.queueRunsWaitingForWorkerBatchSize,
lagRetryDelayMs: options.pendingVersionLagRetryDelayMs,
lagMaxRetries: options.pendingVersionLagMaxRetries,
externalDeploymentParkDeadlineMs: options.externalDeploymentParkDeadlineMs,
});
this.waitpointSystem = new WaitpointSystem({
@@ -855,6 +865,7 @@ export class RunEngine {
streamBasinName,
debounce,
annotations,
parkedOnExternalDeploymentId,
onDebounced,
}: TriggerParams,
tx?: PrismaClientOrTransaction
@@ -946,7 +957,11 @@ export class RunEngine {
}
}
const status = delayUntil ? "DELAYED" : "PENDING";
const status = parkedOnExternalDeploymentId
? "PENDING_VERSION"
: delayUntil
? "DELAYED"
: "PENDING";
// Apply defaultMaxTtl: use as default when no TTL is provided, clamp when larger
const resolvedTtl = this.#resolveMaxTtl(ttl);
@@ -966,6 +981,9 @@ export class RunEngine {
id: taskRunId,
engine: "V2",
status,
statusReason: parkedOnExternalDeploymentId
? PARKED_ON_EXTERNAL_DEPLOYMENT_STATUS_REASON
: undefined,
friendlyId,
runtimeEnvironmentId: environment.id,
environmentType: environment.type,
@@ -1041,8 +1059,16 @@ export class RunEngine {
snapshot: {
id: initialSnapshotId,
engine: "V2",
executionStatus: delayUntil ? "DELAYED" : QUEUED_SNAPSHOT_STATUS,
description: delayUntil ? "Run is delayed" : QUEUED_SNAPSHOT_DESCRIPTION,
executionStatus: parkedOnExternalDeploymentId
? "RUN_CREATED"
: delayUntil
? "DELAYED"
: QUEUED_SNAPSHOT_STATUS,
description: parkedOnExternalDeploymentId
? `Run is waiting for a deployment of '${parkedOnExternalDeploymentId}'`
: delayUntil
? "Run is delayed"
: QUEUED_SNAPSHOT_DESCRIPTION,
runStatus: status,
environmentId: environment.id,
environmentType: environment.type,
@@ -1130,7 +1156,44 @@ export class RunEngine {
}
}
if (taskRun.delayUntil) {
if (parkedOnExternalDeploymentId) {
await this.pendingVersionSystem.scheduleExternalDeploymentParkDeadline({
runId: taskRun.id,
externalDeploymentId: parkedOnExternalDeploymentId,
ttl: taskRun.ttl,
delayUntil: taskRun.delayUntil,
});
this.eventBus.emit("executionSnapshotCreated", {
time: new Date(),
run: {
id: taskRun.id,
},
snapshot: {
id: initialSnapshotId,
executionStatus: "RUN_CREATED",
description: `Run is waiting for a deployment of '${parkedOnExternalDeploymentId}'`,
runStatus: taskRun.status,
attemptNumber: taskRun.attemptNumber ?? null,
checkpointId: null,
workerId: workerId ?? null,
runnerId: runnerId ?? null,
isValid: true,
error: null,
completedWaitpointIds: [],
},
});
if (debounce) {
await this.#registerDebouncedRun({
taskRun,
environmentId: environment.id,
taskIdentifier,
debounce,
debounceClaimId,
});
}
} else if (taskRun.delayUntil) {
// Schedule the run to be enqueued at the delayUntil time
await this.delayedRunSystem.scheduleDelayedRunEnqueuing({
runId: taskRun.id,
@@ -1139,23 +1202,13 @@ export class RunEngine {
// Register debounced run in Redis for future lookups
if (debounce) {
const registered = await this.debounceSystem.registerDebouncedRun({
runId: taskRun.id,
await this.#registerDebouncedRun({
taskRun,
environmentId: environment.id,
taskIdentifier,
debounceKey: debounce.key,
delayUntil: taskRun.delayUntil,
claimId: debounceClaimId,
debounce,
debounceClaimId,
});
if (!registered) {
// We lost the claim - this shouldn't normally happen, but log it
this.logger.warn("trigger: lost debounce claim after creating run", {
runId: taskRun.id,
debounceKey: debounce.key,
claimId: debounceClaimId,
});
}
}
} else {
try {
@@ -2925,6 +2978,41 @@ export class RunEngine {
* - No TTL on the run → use the max as the default.
* - Both exist → clamp to the smaller value.
*/
async #registerDebouncedRun({
taskRun,
environmentId,
taskIdentifier,
debounce,
debounceClaimId,
}: {
taskRun: { id: string; delayUntil: Date | null };
environmentId: string;
taskIdentifier: string;
debounce: { key: string };
debounceClaimId: string | undefined;
}) {
if (!taskRun.delayUntil) {
return;
}
const registered = await this.debounceSystem.registerDebouncedRun({
runId: taskRun.id,
environmentId,
taskIdentifier,
debounceKey: debounce.key,
delayUntil: taskRun.delayUntil,
claimId: debounceClaimId,
});
if (!registered) {
this.logger.warn("trigger: lost debounce claim after creating run", {
runId: taskRun.id,
debounceKey: debounce.key,
claimId: debounceClaimId,
});
}
}
#resolveMaxTtl(ttl: string | undefined): string | undefined {
const maxTtl = this.options.defaultMaxTtl;
@@ -17,6 +17,7 @@ export type PendingVersionRunIdLookupOptions = {
environmentId: string;
taskIdentifiers: string[];
queues: string[];
externalDeploymentId?: string;
/** Maximum number of ids to return. Implementations must respect this cap. */
limit: number;
};
@@ -14,6 +14,7 @@ import type {
PrismaClientOrTransaction,
PrismaReplicaClient,
TaskRun,
TaskRunStatus,
Waitpoint,
} from "@trigger.dev/database";
import { nanoid } from "nanoid";
@@ -91,6 +92,8 @@ export type DebounceResult =
status: "max_duration_exceeded";
};
const DEBOUNCEABLE_RUN_STATUSES: TaskRunStatus[] = ["DELAYED", "PENDING_VERSION"];
// TTL for the pending claim state (30 seconds)
const CLAIM_TTL_MS = 30_000;
// Max retries when waiting for another server to complete its claim
@@ -643,7 +646,7 @@ return 0
{ select: { status: true, delayUntil: true, createdAt: true } },
prisma
);
if (!probe || probe.status !== "DELAYED" || !probe.delayUntil) {
if (!probe || !DEBOUNCEABLE_RUN_STATUSES.includes(probe.status) || !probe.delayUntil) {
return null;
}
if (newDelayUntil.getTime() > probe.delayUntil.getTime()) {
@@ -666,7 +669,7 @@ return 0
{ include: { associatedWaitpoint: true } },
prisma
);
if (!fullRun || fullRun.status !== "DELAYED") {
if (!fullRun || !DEBOUNCEABLE_RUN_STATUSES.includes(fullRun.status)) {
return null;
}
@@ -701,12 +704,12 @@ return 0
prisma
);
if (!fullRun || fullRun.status !== "DELAYED") {
if (!fullRun || !DEBOUNCEABLE_RUN_STATUSES.includes(fullRun.status)) {
// 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",
"handleExistingRun: lock contention, but existing run is no longer debounceable - rethrowing",
{
existingRunId,
debounceKey: debounce.key,
@@ -465,8 +465,12 @@ describe("pendingVersionSystem store routing (cross-version / cross-DB)", () =>
expect(oldHydrate.map((r) => r.id)).toEqual(oldIds);
// Promotion flips identically across versions.
const newPromote = await newStore.promotePendingVersionRuns(newIds[0], prisma17 as any);
const oldPromote = await legacyStore.promotePendingVersionRuns(oldIds[0], prisma14 as any);
const newPromote = await newStore.promotePendingVersionRuns(newIds[0], undefined, prisma17);
const oldPromote = await legacyStore.promotePendingVersionRuns(
oldIds[0],
undefined,
prisma14
);
expect(newPromote.count).toBe(oldPromote.count);
expect(newPromote.count).toBe(1);
@@ -518,7 +522,7 @@ describe("pendingVersionSystem store routing (cross-version / cross-DB)", () =>
expect(hydrated.map((r) => r.id)).toEqual([newId]);
// Promote on NEW.
const promote = await newStore.promotePendingVersionRuns(newId, prisma17 as any);
const promote = await newStore.promotePendingVersionRuns(newId, undefined, prisma17);
expect(promote.count).toBe(1);
// NEW flipped; LEGACY row untouched.
@@ -1,3 +1,9 @@
import {
compareDeploymentVersions,
parseNaturalLanguageDuration,
} from "@trigger.dev/core/v3/isomorphic";
import type { TaskRunError } from "@trigger.dev/core/v3/schemas";
import type { MinimalAuthenticatedEnvironment } from "../../shared/index.js";
import type { EnqueueSystem } from "./enqueueSystem.js";
import type { SystemResources } from "./systems.js";
@@ -21,10 +27,35 @@ export type PendingVersionSystemOptions = {
* disable lag-aware retries entirely.
*/
lagMaxRetries?: number;
externalDeploymentParkDeadlineMs?: number;
};
const DEFAULT_LAG_RETRY_DELAY_MS = 5_000;
const DEFAULT_LAG_MAX_RETRIES = 1;
const DEFAULT_EXTERNAL_DEPLOYMENT_PARK_DEADLINE_MS = 60 * 60 * 1000;
export const PARKED_ON_EXTERNAL_DEPLOYMENT_STATUS_REASON = "EXTERNAL_DEPLOYMENT_PENDING";
const EXPIRED_ON_EXTERNAL_DEPLOYMENT_STATUS_REASON = "EXTERNAL_DEPLOYMENT_NOT_FOUND";
const MAX_DEPLOYMENT_CANDIDATES = 20;
type ExternalDeploymentWorker = {
id: string;
version: string;
sdkVersion?: string;
cliVersion?: string;
};
function readExternalDeploymentIdAnnotation(annotations: unknown): string | undefined {
if (typeof annotations !== "object" || annotations === null) {
return undefined;
}
const value = (annotations as Record<string, unknown>).externalDeploymentId;
return typeof value === "string" && value.length > 0 ? value : undefined;
}
export class PendingVersionSystem {
private readonly $: SystemResources;
@@ -53,6 +84,9 @@ export class PendingVersionSystem {
},
tasks: true,
queues: true,
deployment: {
select: { externalId: true },
},
},
});
@@ -65,11 +99,13 @@ export class PendingVersionSystem {
const taskIdentifiers = backgroundWorker.tasks.map((task) => task.slug);
const queues = backgroundWorker.queues.map((queue) => queue.name);
const externalDeploymentId = backgroundWorker.deployment?.externalId ?? undefined;
this.$.logger.debug("Finding PENDING_VERSION runs for background worker", {
workerId: backgroundWorker.id,
taskIdentifiers,
queues,
externalDeploymentId,
});
// Step 1: ask the injected lookup (typically ClickHouse-backed) for
@@ -81,6 +117,7 @@ export class PendingVersionSystem {
environmentId: backgroundWorker.runtimeEnvironmentId,
taskIdentifiers,
queues,
externalDeploymentId,
limit: maxCount + 1,
});
@@ -111,6 +148,11 @@ export class PendingVersionSystem {
// CH returned candidates but all of them have already moved past
// PENDING_VERSION (typically because a concurrent deploy or retry
// beat us to them). Don't reschedule — there's no work to wait for.
await this.#maybeScheduleExternalDeploymentLagRetry(
backgroundWorkerId,
attempt,
externalDeploymentId
);
return;
}
@@ -119,6 +161,7 @@ export class PendingVersionSystem {
lookupName: this.$.pendingVersionRunIdLookup.name,
candidateCount: candidateIds.length,
pendingRunCount: pendingRuns.length,
externalDeploymentId,
runs: pendingRuns.map((run) => ({
id: run.id,
taskIdentifier: run.taskIdentifier,
@@ -128,7 +171,45 @@ export class PendingVersionSystem {
})),
});
const now = new Date();
let promotedCount = 0;
let skippedForOtherId = 0;
const externalDeploymentPin = externalDeploymentId
? ((await this.#findDeployedWorkerForExternalId(
backgroundWorker.runtimeEnvironmentId,
externalDeploymentId
)) ?? {
id: backgroundWorker.id,
version: backgroundWorker.version,
sdkVersion: backgroundWorker.sdkVersion ?? undefined,
cliVersion: backgroundWorker.cliVersion ?? undefined,
})
: undefined;
for (const run of pendingRuns) {
const runExternalDeploymentId =
run.statusReason === PARKED_ON_EXTERNAL_DEPLOYMENT_STATUS_REASON
? readExternalDeploymentIdAnnotation(run.annotations)
: undefined;
if (runExternalDeploymentId && runExternalDeploymentId !== externalDeploymentId) {
skippedForOtherId++;
continue;
}
const pin =
runExternalDeploymentId && externalDeploymentPin
? {
lockedToVersionId: externalDeploymentPin.id,
taskVersion: externalDeploymentPin.version,
sdkVersion: externalDeploymentPin.sdkVersion ?? undefined,
cliVersion: externalDeploymentPin.cliVersion ?? undefined,
}
: {};
const stillDelayed = run.delayUntil !== null && run.delayUntil > now;
// Atomic unit: the status promotion and the new QUEUED snapshot must commit together
// or a crash between them leaves the run promoted-to-PENDING with no snapshot. Under the run-ops
// split these route to the run's owning DB but, as two router calls, would each auto-commit.
@@ -139,12 +220,20 @@ export class PendingVersionSystem {
// Idempotency guard: only flips PENDING_VERSION → PENDING. If another
// worker already promoted this run between our findMany and the
// update, count is 0 and we skip the enqueue.
const updateResult = await store.promotePendingVersionRuns(run.id, tx);
const updateResult = await store.promotePendingVersionRuns(
run.id,
{ ...pin, status: stillDelayed ? "DELAYED" : "PENDING" },
tx
);
if (updateResult.count === 0) {
return false;
}
if (stillDelayed) {
return true;
}
const updatedRun = await store.findRunOrThrow({ id: run.id }, tx);
await this.enqueueSystem.enqueueRun({
@@ -164,11 +253,22 @@ export class PendingVersionSystem {
if (!promoted) continue;
promotedCount++;
if (stillDelayed && run.delayUntil) {
await this.$.worker.enqueue({
id: `enqueueDelayedRun:${run.id}`,
job: "enqueueDelayedRun",
payload: { runId: run.id },
availableAt: run.delayUntil,
});
}
this.$.eventBus.emit("runStatusChanged", {
time: new Date(),
run: {
id: run.id,
status: "PENDING",
status: stillDelayed ? "DELAYED" : "PENDING",
updatedAt: run.updatedAt,
createdAt: run.createdAt,
runTags: run.runTags,
@@ -186,14 +286,38 @@ export class PendingVersionSystem {
});
}
// Reschedule when the lookup returned a full-plus-one batch — that's
// the signal there are more candidates to drain. Use `candidateIds`
// (the raw lookup result) rather than `pendingRuns` (post-status-guard)
// because runs that already left PENDING_VERSION shouldn't suppress
// the next batch.
if (candidateIds.length > maxCount) {
if (candidateIds.length > maxCount && (promotedCount > 0 || skippedForOtherId === 0)) {
await this.scheduleResolvePendingVersionRuns(backgroundWorkerId);
return;
}
await this.#maybeScheduleExternalDeploymentLagRetry(
backgroundWorkerId,
attempt,
externalDeploymentId
);
}
// A run parked on an external deployment id is typically created moments before the
// deployment finalizes, so replication lag can hide it from the candidate lookup. The
// `lookup_empty` retry does not cover that: any other visible parked run in the
// environment makes the lookup non-empty and suppresses it. Arm one bounded follow-up
// whenever the landing deployment carries an id, so a just-parked run is picked up in
// seconds instead of waiting for the park deadline.
async #maybeScheduleExternalDeploymentLagRetry(
backgroundWorkerId: string,
attempt: number,
externalDeploymentId: string | undefined
): Promise<void> {
if (!externalDeploymentId) {
return;
}
await this.#maybeScheduleLagRetry(
backgroundWorkerId,
attempt,
"external_deployment_replication_lag"
);
}
async scheduleResolvePendingVersionRuns(
@@ -208,6 +332,332 @@ export class PendingVersionSystem {
});
}
async scheduleExternalDeploymentParkDeadline({
runId,
externalDeploymentId,
ttl,
delayUntil,
}: {
runId: string;
externalDeploymentId: string;
ttl?: string | null;
delayUntil?: Date | null;
}): Promise<void> {
const deadlineMs =
this.options.externalDeploymentParkDeadlineMs ?? DEFAULT_EXTERNAL_DEPLOYMENT_PARK_DEADLINE_MS;
const now = Date.now();
const anchor = delayUntil && delayUntil.getTime() > now ? delayUntil.getTime() : now;
const defaultDeadline = new Date(anchor + deadlineMs);
const ttlDeadline = ttl ? parseNaturalLanguageDuration(ttl) : undefined;
const availableAt =
ttlDeadline && ttlDeadline < defaultDeadline ? ttlDeadline : defaultDeadline;
await this.$.worker.enqueue({
id: `expireParkedExternalDeploymentRun:${runId}`,
job: "expireParkedExternalDeploymentRun",
payload: { runId, externalDeploymentId },
availableAt,
});
}
async expireParkedExternalDeploymentRun({
runId,
externalDeploymentId,
}: {
runId: string;
externalDeploymentId: string;
}): Promise<void> {
const run = await this.$.runStore.findRun(
{ id: runId },
{
select: {
id: true,
status: true,
annotations: true,
runtimeEnvironmentId: true,
organizationId: true,
projectId: true,
spanId: true,
ttl: true,
taskEventStore: true,
createdAt: true,
updatedAt: true,
delayUntil: true,
runTags: true,
batchId: true,
associatedWaitpoint: { select: { id: true } },
},
},
this.$.prisma
);
if (!run) {
this.$.logger.debug("expireParkedExternalDeploymentRun: run not found", { runId });
return;
}
if (!run.organizationId) {
this.$.logger.error("expireParkedExternalDeploymentRun: run has no organization", { runId });
return;
}
if (run.status !== "PENDING_VERSION") {
return;
}
if (readExternalDeploymentIdAnnotation(run.annotations) !== externalDeploymentId) {
this.$.logger.debug(
"expireParkedExternalDeploymentRun: run no longer parked on this external deployment id",
{ runId, externalDeploymentId }
);
return;
}
const env = await this.$.controlPlaneResolver.resolveEnv(run.runtimeEnvironmentId);
if (!env) {
this.$.logger.error("expireParkedExternalDeploymentRun: environment not found", {
runId,
environmentId: run.runtimeEnvironmentId,
});
return;
}
const worker = await this.#findDeployedWorkerForExternalId(
run.runtimeEnvironmentId,
externalDeploymentId
);
if (worker) {
this.$.logger.info(
"expireParkedExternalDeploymentRun: deployment landed after all, releasing run",
{ runId, externalDeploymentId, workerId: worker.id, version: worker.version }
);
const released = await this.#promoteParkedRun({
run: {
id: run.id,
delayUntil: run.delayUntil,
createdAt: run.createdAt,
updatedAt: run.updatedAt,
runTags: run.runTags,
batchId: run.batchId,
},
env,
pin: worker,
});
if (released) {
return;
}
const stillParked = await this.$.runStore.findRun(
{ id: runId },
{ select: { status: true } },
this.$.prisma
);
if (stillParked?.status !== "PENDING_VERSION") {
return;
}
this.$.logger.warn(
"expireParkedExternalDeploymentRun: deployment holds the id but the run could not be released, expiring",
{ runId, externalDeploymentId, workerId: worker.id }
);
}
const error: TaskRunError = {
type: "STRING_ERROR",
raw: `Run expired because no deployment with external id '${externalDeploymentId}' became available`,
};
const now = new Date();
const result = await this.$.runStore.expireParkedRun(
runId,
{
error,
completedAt: now,
expiredAt: now,
statusReason: EXPIRED_ON_EXTERNAL_DEPLOYMENT_STATUS_REASON,
snapshot: {
engine: "V2",
executionStatus: "FINISHED",
description: `Run was expired because no deployment with external id '${externalDeploymentId}' became available`,
runStatus: "EXPIRED",
environmentId: run.runtimeEnvironmentId,
environmentType: env.type,
projectId: run.projectId,
organizationId: run.organizationId,
},
},
this.$.prisma
);
if (result.count === 0) {
return;
}
if (run.associatedWaitpoint) {
await this.$.worker.enqueue({
id: `finishWaitpoint.externalDeploymentPark.${run.associatedWaitpoint.id}`,
job: "finishWaitpoint",
payload: {
waitpointId: run.associatedWaitpoint.id,
error: JSON.stringify(error),
},
});
}
this.$.eventBus.emit("runExpired", {
time: now,
run: {
id: runId,
status: "EXPIRED",
spanId: run.spanId,
ttl: run.ttl,
taskEventStore: run.taskEventStore,
createdAt: run.createdAt,
completedAt: now,
expiredAt: now,
updatedAt: now,
},
organization: {
id: run.organizationId,
},
project: {
id: run.projectId,
},
environment: {
id: run.runtimeEnvironmentId,
},
});
}
async #promoteParkedRun({
run,
env,
pin,
}: {
run: {
id: string;
delayUntil: Date | null;
createdAt: Date;
updatedAt: Date;
runTags: string[];
batchId: string | null;
};
env: MinimalAuthenticatedEnvironment & { organizationId?: string; projectId?: string };
pin: ExternalDeploymentWorker;
}): Promise<boolean> {
const stillDelayed = run.delayUntil !== null && run.delayUntil > new Date();
const promoted = await this.$.runStore.runInTransaction(run.id, async (store, tx) => {
const updateResult = await store.promotePendingVersionRuns(
run.id,
{
status: stillDelayed ? "DELAYED" : "PENDING",
lockedToVersionId: pin.id,
taskVersion: pin.version,
sdkVersion: pin.sdkVersion ?? undefined,
cliVersion: pin.cliVersion ?? undefined,
},
tx
);
if (updateResult.count === 0) {
return false;
}
if (stillDelayed) {
return true;
}
const updatedRun = await store.findRunOrThrow({ id: run.id }, tx);
await this.enqueueSystem.enqueueRun({
run: updatedRun,
env,
store,
tx,
includeTtl: true,
});
return true;
});
if (!promoted) {
return false;
}
if (stillDelayed && run.delayUntil) {
await this.$.worker.enqueue({
id: `enqueueDelayedRun:${run.id}`,
job: "enqueueDelayedRun",
payload: { runId: run.id },
availableAt: run.delayUntil,
});
}
this.$.eventBus.emit("runStatusChanged", {
time: new Date(),
run: {
id: run.id,
status: stillDelayed ? "DELAYED" : "PENDING",
updatedAt: run.updatedAt,
createdAt: run.createdAt,
runTags: run.runTags,
batchId: run.batchId,
},
organization: { id: env.organization.id },
project: { id: env.project.id },
environment: { id: env.id },
});
return true;
}
async #findDeployedWorkerForExternalId(
environmentId: string,
externalDeploymentId: string
): Promise<ExternalDeploymentWorker | undefined> {
const candidates = await this.$.prisma.workerDeployment.findMany({
where: {
environmentId,
externalId: externalDeploymentId,
status: "DEPLOYED",
},
select: {
version: true,
worker: { select: { id: true, version: true, sdkVersion: true, cliVersion: true } },
},
orderBy: { id: "desc" },
take: MAX_DEPLOYMENT_CANDIDATES,
});
let highest: { version: string; worker: ExternalDeploymentWorker } | undefined;
for (const candidate of candidates) {
if (!candidate.worker) continue;
if (highest && compareDeploymentVersions(candidate.version, highest.version) <= 0) continue;
highest = {
version: candidate.version,
worker: {
id: candidate.worker.id,
version: candidate.worker.version,
sdkVersion: candidate.worker.sdkVersion ?? undefined,
cliVersion: candidate.worker.cliVersion ?? undefined,
},
};
}
return highest?.worker;
}
/**
* Schedule one more lookup attempt when the first found zero candidates,
* to cover ClickHouse replication lag against `task_runs_v2`. Bounded by
@@ -216,7 +666,7 @@ export class PendingVersionSystem {
async #maybeScheduleLagRetry(
backgroundWorkerId: string,
attempt: number,
reason: "lookup_empty"
reason: "lookup_empty" | "external_deployment_replication_lag"
): Promise<void> {
const maxRetries = this.options.lagMaxRetries ?? DEFAULT_LAG_MAX_RETRIES;
@@ -0,0 +1,854 @@
import { assertNonNullable, containerTest } from "@internal/testcontainers";
import { trace } from "@internal/tracing";
import type { PrismaClient } from "@trigger.dev/database";
import { RunEngine } from "../index.js";
import { NoopPendingVersionRunIdLookup } from "../services/pendingVersionLookup.js";
import { PostgresPendingVersionRunIdLookup } from "./postgresPendingVersionLookup.js";
import { setupAuthenticatedEnvironment, setupBackgroundWorker } from "./setup.js";
vi.setConfig({ testTimeout: 60_000 });
function createEngine(
prisma: PrismaClient,
redisOptions: any,
overrides?: Record<string, unknown>
) {
return new RunEngine({
prisma,
worker: {
redis: redisOptions,
workers: 1,
tasksPerWorker: 10,
pollIntervalMs: 100,
},
queue: {
redis: redisOptions,
processWorkerQueueDebounceMs: 50,
masterQueueConsumersDisabled: true,
},
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,
},
pendingVersionRunIdLookup: new PostgresPendingVersionRunIdLookup(prisma),
tracer: trace.getTracer("test", "0.0.0"),
...overrides,
});
}
async function nameDeploymentWithExternalId(
prisma: PrismaClient,
workerId: string,
externalId: string
) {
return prisma.workerDeployment.update({
where: { workerId },
data: { externalId },
});
}
describe("RunEngine external deployment parking", () => {
containerTest(
"parks a run whose external deployment id nothing holds, then releases it pinned when a deployment carrying that id lands",
async ({ prisma, redisOptions }) => {
const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
const engine = createEngine(prisma, redisOptions);
try {
const taskIdentifier = "test-task";
const currentWorker = await setupBackgroundWorker(
engine,
authenticatedEnvironment,
taskIdentifier
);
const run = await engine.trigger(
{
number: 1,
friendlyId: "run_1234",
environment: authenticatedEnvironment,
taskIdentifier,
payload: "{}",
payloadType: "application/json",
context: {},
traceContext: {},
traceId: "t1234",
spanId: "s1234",
queue: `task/${taskIdentifier}`,
isTest: false,
tags: [],
annotations: {
triggerSource: "sdk",
triggerAction: "trigger",
rootTriggerSource: "sdk",
externalDeploymentId: "commit-abc",
},
parkedOnExternalDeploymentId: "commit-abc",
},
prisma
);
const parked = await prisma.taskRun.findFirstOrThrow({ where: { id: run.id } });
expect(parked.status).toBe("PENDING_VERSION");
expect(parked.statusReason).toBe("EXTERNAL_DEPLOYMENT_PENDING");
expect(parked.lockedToVersionId).toBeNull();
expect((parked.annotations as Record<string, unknown>).externalDeploymentId).toBe(
"commit-abc"
);
const targetWorker = await setupBackgroundWorker(
engine,
authenticatedEnvironment,
taskIdentifier
);
await nameDeploymentWithExternalId(prisma, targetWorker.worker.id, "commit-abc");
await engine.pendingVersionSystem.enqueueRunsForBackgroundWorker(targetWorker.worker.id);
const released = await prisma.taskRun.findFirstOrThrow({ where: { id: run.id } });
expect(released.status).toBe("PENDING");
expect(released.lockedToVersionId).toBe(targetWorker.worker.id);
expect(released.taskVersion).toBe(targetWorker.worker.version);
expect(released.lockedToVersionId).not.toBe(currentWorker.worker.id);
} finally {
await engine.quit();
}
}
);
containerTest(
"an unrelated deployment landing does not release a run parked on a different id, but still releases runs parked for other reasons",
async ({ prisma, redisOptions }) => {
const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
const engine = createEngine(prisma, redisOptions);
try {
const taskIdentifier = "test-task";
const pinnedRun = await engine.trigger(
{
number: 1,
friendlyId: "run_1234",
environment: authenticatedEnvironment,
taskIdentifier,
payload: "{}",
payloadType: "application/json",
context: {},
traceContext: {},
traceId: "t1234",
spanId: "s1234",
queue: `task/${taskIdentifier}`,
isTest: false,
tags: [],
annotations: {
triggerSource: "sdk",
triggerAction: "trigger",
rootTriggerSource: "sdk",
externalDeploymentId: "commit-wanted",
},
parkedOnExternalDeploymentId: "commit-wanted",
},
prisma
);
const plainRun = await engine.trigger(
{
number: 2,
friendlyId: "run_1235",
environment: authenticatedEnvironment,
taskIdentifier,
payload: "{}",
payloadType: "application/json",
context: {},
traceContext: {},
traceId: "t1235",
spanId: "s1235",
queue: `task/${taskIdentifier}`,
isTest: false,
tags: [],
},
prisma
);
expect((await prisma.taskRun.findFirstOrThrow({ where: { id: plainRun.id } })).status).toBe(
"PENDING"
);
await prisma.taskRun.update({
where: { id: plainRun.id },
data: { status: "PENDING_VERSION", statusReason: "NO_WORKER" },
});
const otherWorker = await setupBackgroundWorker(
engine,
authenticatedEnvironment,
taskIdentifier
);
await nameDeploymentWithExternalId(prisma, otherWorker.worker.id, "commit-unrelated");
await engine.pendingVersionSystem.enqueueRunsForBackgroundWorker(otherWorker.worker.id);
const stillParked = await prisma.taskRun.findFirstOrThrow({ where: { id: pinnedRun.id } });
expect(stillParked.status).toBe("PENDING_VERSION");
expect(stillParked.lockedToVersionId).toBeNull();
const releasedPlain = await prisma.taskRun.findFirstOrThrow({
where: { id: plainRun.id },
});
expect(releasedPlain.status).toBe("PENDING");
expect(releasedPlain.lockedToVersionId).toBeNull();
} finally {
await engine.quit();
}
}
);
containerTest(
"releases a run that reports an external deployment id but is parked for an unrelated reason",
async ({ prisma, redisOptions }) => {
const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
const engine = createEngine(prisma, redisOptions);
try {
const taskIdentifier = "test-task";
const run = await engine.trigger(
{
number: 1,
friendlyId: "run_1234",
environment: authenticatedEnvironment,
taskIdentifier,
payload: "{}",
payloadType: "application/json",
context: {},
traceContext: {},
traceId: "t1234",
spanId: "s1234",
queue: `task/${taskIdentifier}`,
isTest: false,
tags: [],
annotations: {
triggerSource: "sdk",
triggerAction: "trigger",
rootTriggerSource: "sdk",
externalDeploymentId: "commit-reported",
},
},
prisma
);
await prisma.taskRun.update({
where: { id: run.id },
data: { status: "PENDING_VERSION", statusReason: "NO_WORKER" },
});
const worker = await setupBackgroundWorker(
engine,
authenticatedEnvironment,
taskIdentifier
);
await engine.pendingVersionSystem.enqueueRunsForBackgroundWorker(worker.worker.id);
const released = await prisma.taskRun.findFirstOrThrow({ where: { id: run.id } });
expect(released.status).toBe("PENDING");
expect(released.lockedToVersionId).toBeNull();
} finally {
await engine.quit();
}
}
);
containerTest(
"a no-id deployment landing drains ordinary parked runs past a backlog of id-parked ones, without looping",
async ({ prisma, redisOptions }) => {
const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
const engine = createEngine(prisma, redisOptions, {
queueRunsWaitingForWorkerBatchSize: 2,
});
try {
const taskIdentifier = "test-task";
for (let i = 0; i < 3; i++) {
await engine.trigger(
{
number: i + 1,
friendlyId: `run_124${i}`,
environment: authenticatedEnvironment,
taskIdentifier,
payload: "{}",
payloadType: "application/json",
context: {},
traceContext: {},
traceId: `t124${i}`,
spanId: `s124${i}`,
queue: `task/${taskIdentifier}`,
isTest: false,
tags: [],
annotations: {
triggerSource: "sdk",
triggerAction: "trigger",
rootTriggerSource: "sdk",
externalDeploymentId: "commit-never-lands",
},
parkedOnExternalDeploymentId: "commit-never-lands",
},
prisma
);
}
const plainRun = await engine.trigger(
{
number: 4,
friendlyId: "run_1299",
environment: authenticatedEnvironment,
taskIdentifier,
payload: "{}",
payloadType: "application/json",
context: {},
traceContext: {},
traceId: "t1299",
spanId: "s1299",
queue: `task/${taskIdentifier}`,
isTest: false,
tags: [],
},
prisma
);
await prisma.taskRun.update({
where: { id: plainRun.id },
data: { status: "PENDING_VERSION", statusReason: "NO_WORKER" },
});
const worker = await setupBackgroundWorker(
engine,
authenticatedEnvironment,
taskIdentifier
);
const scheduleSpy = vi.spyOn(
engine.pendingVersionSystem,
"scheduleResolvePendingVersionRuns"
);
await engine.pendingVersionSystem.enqueueRunsForBackgroundWorker(worker.worker.id);
const releasedPlain = await prisma.taskRun.findFirstOrThrow({ where: { id: plainRun.id } });
expect(releasedPlain.status).toBe("PENDING");
const stillParked = await prisma.taskRun.findMany({
where: { id: { not: plainRun.id }, runtimeEnvironmentId: authenticatedEnvironment.id },
select: { status: true },
});
expect(stillParked.map((r) => r.status)).toEqual([
"PENDING_VERSION",
"PENDING_VERSION",
"PENDING_VERSION",
]);
expect(scheduleSpy).not.toHaveBeenCalled();
} finally {
await engine.quit();
}
}
);
containerTest(
"pins to the highest deployed version holding the id, even when an older build of that id finalizes last",
async ({ prisma, redisOptions }) => {
const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
const engine = createEngine(prisma, redisOptions);
try {
const taskIdentifier = "test-task";
const run = await engine.trigger(
{
number: 1,
friendlyId: "run_1234",
environment: authenticatedEnvironment,
taskIdentifier,
payload: "{}",
payloadType: "application/json",
context: {},
traceContext: {},
traceId: "t1234",
spanId: "s1234",
queue: `task/${taskIdentifier}`,
isTest: false,
tags: [],
annotations: {
triggerSource: "sdk",
triggerAction: "trigger",
rootTriggerSource: "sdk",
externalDeploymentId: "commit-forced",
},
parkedOnExternalDeploymentId: "commit-forced",
},
prisma
);
const olderWorker = await setupBackgroundWorker(
engine,
authenticatedEnvironment,
taskIdentifier
);
await prisma.backgroundWorker.update({
where: { id: olderWorker.worker.id },
data: { version: "20260807.9" },
});
await prisma.workerDeployment.update({
where: { workerId: olderWorker.worker.id },
data: {
externalId: "commit-forced",
version: "20260807.9",
shortCode: "short_code_20260807.9",
},
});
const newerWorker = await setupBackgroundWorker(
engine,
authenticatedEnvironment,
taskIdentifier
);
await prisma.backgroundWorker.update({
where: { id: newerWorker.worker.id },
data: { version: "20260807.10" },
});
await prisma.workerDeployment.update({
where: { workerId: newerWorker.worker.id },
data: {
externalId: "commit-forced",
version: "20260807.10",
shortCode: "short_code_20260807.10",
},
});
await engine.pendingVersionSystem.enqueueRunsForBackgroundWorker(olderWorker.worker.id);
const released = await prisma.taskRun.findFirstOrThrow({ where: { id: run.id } });
expect(released.status).toBe("PENDING");
expect(released.lockedToVersionId).toBe(newerWorker.worker.id);
expect(released.taskVersion).toBe("20260807.10");
} finally {
await engine.quit();
}
}
);
containerTest(
"the parking deadline is measured from the delay the caller asked for, not from creation",
async ({ prisma, redisOptions }) => {
const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
const engine = createEngine(prisma, redisOptions);
try {
const taskIdentifier = "test-task";
const delayUntil = new Date(Date.now() + 24 * 60 * 60 * 1000);
// eslint-disable-next-line @typescript-eslint/no-explicit-any
const enqueueSpy = vi.spyOn((engine.pendingVersionSystem as any).$.worker, "enqueue");
await engine.trigger(
{
number: 1,
friendlyId: "run_1234",
environment: authenticatedEnvironment,
taskIdentifier,
payload: "{}",
payloadType: "application/json",
context: {},
traceContext: {},
traceId: "t1234",
spanId: "s1234",
queue: `task/${taskIdentifier}`,
isTest: false,
tags: [],
delayUntil,
annotations: {
triggerSource: "sdk",
triggerAction: "trigger",
rootTriggerSource: "sdk",
externalDeploymentId: "commit-delayed",
},
parkedOnExternalDeploymentId: "commit-delayed",
},
prisma
);
const deadlineCall = enqueueSpy.mock.calls.find(
// eslint-disable-next-line @typescript-eslint/no-explicit-any
([args]: any[]) => args?.job === "expireParkedExternalDeploymentRun"
);
assertNonNullable(deadlineCall);
// eslint-disable-next-line @typescript-eslint/no-explicit-any
const availableAt = (deadlineCall[0] as any).availableAt as Date;
expect(availableAt.getTime()).toBeGreaterThan(delayUntil.getTime());
} finally {
await engine.quit();
}
}
);
containerTest(
"a deployment carrying an external id arms a follow-up sweep even when it found candidates",
async ({ prisma, redisOptions }) => {
const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
const engine = createEngine(prisma, redisOptions);
try {
const taskIdentifier = "test-task";
await engine.trigger(
{
number: 1,
friendlyId: "run_1234",
environment: authenticatedEnvironment,
taskIdentifier,
payload: "{}",
payloadType: "application/json",
context: {},
traceContext: {},
traceId: "t1234",
spanId: "s1234",
queue: `task/${taskIdentifier}`,
isTest: false,
tags: [],
annotations: {
triggerSource: "sdk",
triggerAction: "trigger",
rootTriggerSource: "sdk",
externalDeploymentId: "commit-lands",
},
parkedOnExternalDeploymentId: "commit-lands",
},
prisma
);
const worker = await setupBackgroundWorker(
engine,
authenticatedEnvironment,
taskIdentifier
);
await nameDeploymentWithExternalId(prisma, worker.worker.id, "commit-lands");
const scheduleSpy = vi.spyOn(
engine.pendingVersionSystem,
"scheduleResolvePendingVersionRuns"
);
await engine.pendingVersionSystem.enqueueRunsForBackgroundWorker(worker.worker.id);
expect(scheduleSpy).toHaveBeenCalledTimes(1);
expect(scheduleSpy.mock.calls[0]?.[1]?.attempt).toBe(1);
} finally {
await engine.quit();
}
}
);
containerTest(
"the parking deadline expires a run whose deployment never arrived",
async ({ prisma, redisOptions }) => {
const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
const engine = createEngine(prisma, redisOptions);
try {
const taskIdentifier = "test-task";
const run = await engine.trigger(
{
number: 1,
friendlyId: "run_1234",
environment: authenticatedEnvironment,
taskIdentifier,
payload: "{}",
payloadType: "application/json",
context: {},
traceContext: {},
traceId: "t1234",
spanId: "s1234",
queue: `task/${taskIdentifier}`,
isTest: false,
tags: [],
annotations: {
triggerSource: "sdk",
triggerAction: "trigger",
rootTriggerSource: "sdk",
externalDeploymentId: "commit-never",
},
parkedOnExternalDeploymentId: "commit-never",
},
prisma
);
await engine.pendingVersionSystem.expireParkedExternalDeploymentRun({
runId: run.id,
externalDeploymentId: "commit-never",
});
const expired = await prisma.taskRun.findFirstOrThrow({ where: { id: run.id } });
expect(expired.status).toBe("EXPIRED");
expect(expired.statusReason).toBe("EXTERNAL_DEPLOYMENT_NOT_FOUND");
expect(expired.expiredAt).not.toBeNull();
expect(JSON.stringify(expired.error)).toContain("commit-never");
} finally {
await engine.quit();
}
}
);
containerTest(
"the parking deadline re-checks Postgres and releases instead of expiring when the deployment did land",
async ({ prisma, redisOptions }) => {
const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
const engine = createEngine(prisma, redisOptions, {
pendingVersionRunIdLookup: new NoopPendingVersionRunIdLookup(),
});
try {
const taskIdentifier = "test-task";
const run = await engine.trigger(
{
number: 1,
friendlyId: "run_1234",
environment: authenticatedEnvironment,
taskIdentifier,
payload: "{}",
payloadType: "application/json",
context: {},
traceContext: {},
traceId: "t1234",
spanId: "s1234",
queue: `task/${taskIdentifier}`,
isTest: false,
tags: [],
annotations: {
triggerSource: "sdk",
triggerAction: "trigger",
rootTriggerSource: "sdk",
externalDeploymentId: "commit-late",
},
parkedOnExternalDeploymentId: "commit-late",
},
prisma
);
const targetWorker = await setupBackgroundWorker(
engine,
authenticatedEnvironment,
taskIdentifier
);
await nameDeploymentWithExternalId(prisma, targetWorker.worker.id, "commit-late");
await engine.pendingVersionSystem.expireParkedExternalDeploymentRun({
runId: run.id,
externalDeploymentId: "commit-late",
});
const released = await prisma.taskRun.findFirstOrThrow({ where: { id: run.id } });
expect(released.status).toBe("PENDING");
expect(released.lockedToVersionId).toBe(targetWorker.worker.id);
} finally {
await engine.quit();
}
}
);
containerTest(
"the deadline is a no-op once the run has left PENDING_VERSION",
async ({ prisma, redisOptions }) => {
const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
const engine = createEngine(prisma, redisOptions);
try {
const taskIdentifier = "test-task";
const run = await engine.trigger(
{
number: 1,
friendlyId: "run_1234",
environment: authenticatedEnvironment,
taskIdentifier,
payload: "{}",
payloadType: "application/json",
context: {},
traceContext: {},
traceId: "t1234",
spanId: "s1234",
queue: `task/${taskIdentifier}`,
isTest: false,
tags: [],
annotations: {
triggerSource: "sdk",
triggerAction: "trigger",
rootTriggerSource: "sdk",
externalDeploymentId: "commit-raced",
},
parkedOnExternalDeploymentId: "commit-raced",
},
prisma
);
await prisma.taskRun.update({
where: { id: run.id },
data: { status: "PENDING" },
});
await engine.pendingVersionSystem.expireParkedExternalDeploymentRun({
runId: run.id,
externalDeploymentId: "commit-raced",
});
const untouched = await prisma.taskRun.findFirstOrThrow({ where: { id: run.id } });
expect(untouched.status).toBe("PENDING");
expect(untouched.expiredAt).toBeNull();
} finally {
await engine.quit();
}
}
);
containerTest(
"a parked run keeps the delay its caller asked for, and re-delays rather than jumping the queue",
async ({ prisma, redisOptions }) => {
const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
const engine = createEngine(prisma, redisOptions);
try {
const taskIdentifier = "test-task";
const delayUntil = new Date(Date.now() + 60 * 60 * 1000);
const run = await engine.trigger(
{
number: 1,
friendlyId: "run_1234",
environment: authenticatedEnvironment,
taskIdentifier,
payload: "{}",
payloadType: "application/json",
context: {},
traceContext: {},
traceId: "t1234",
spanId: "s1234",
queue: `task/${taskIdentifier}`,
isTest: false,
tags: [],
delayUntil,
annotations: {
triggerSource: "sdk",
triggerAction: "trigger",
rootTriggerSource: "sdk",
externalDeploymentId: "commit-delayed",
},
parkedOnExternalDeploymentId: "commit-delayed",
},
prisma
);
const parked = await prisma.taskRun.findFirstOrThrow({ where: { id: run.id } });
expect(parked.status).toBe("PENDING_VERSION");
assertNonNullable(parked.delayUntil);
const targetWorker = await setupBackgroundWorker(
engine,
authenticatedEnvironment,
taskIdentifier
);
await nameDeploymentWithExternalId(prisma, targetWorker.worker.id, "commit-delayed");
await engine.pendingVersionSystem.enqueueRunsForBackgroundWorker(targetWorker.worker.id);
const released = await prisma.taskRun.findFirstOrThrow({ where: { id: run.id } });
expect(released.status).toBe("DELAYED");
expect(released.lockedToVersionId).toBe(targetWorker.worker.id);
} finally {
await engine.quit();
}
}
);
containerTest(
"a debounced run that parks still collapses later triggers for the same key",
async ({ prisma, redisOptions }) => {
const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
const engine = createEngine(prisma, redisOptions);
try {
const taskIdentifier = "debounced-parked-task";
await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);
const debounce = { key: "digest-user-1", delay: "5m" };
const delayUntil = new Date(Date.now() + 5 * 60 * 1000);
const triggerOnce = (number: number) =>
engine.trigger(
{
number,
friendlyId: `run_123${number}`,
environment: authenticatedEnvironment,
taskIdentifier,
payload: "{}",
payloadType: "application/json",
context: {},
traceContext: {},
traceId: `t_debounce_${number}`,
spanId: `s_debounce_${number}`,
queue: `task/${taskIdentifier}`,
isTest: false,
tags: [],
annotations: {
triggerSource: "sdk",
triggerAction: "trigger",
rootTriggerSource: "sdk",
externalDeploymentId: "commit-abc",
},
delayUntil,
debounce,
parkedOnExternalDeploymentId: "commit-abc",
},
prisma
);
const first = await triggerOnce(1);
const second = await triggerOnce(2);
const third = await triggerOnce(3);
expect(second.id).toBe(first.id);
expect(third.id).toBe(first.id);
const runs = await prisma.taskRun.findMany({
where: { runtimeEnvironmentId: authenticatedEnvironment.id, taskIdentifier },
});
expect(runs).toHaveLength(1);
expect(runs[0]!.status).toBe("PENDING_VERSION");
} finally {
await engine.quit();
}
}
);
});
@@ -1,4 +1,5 @@
import type { PrismaClient } from "@trigger.dev/database";
import { type PrismaClient } from "@trigger.dev/database";
import { PARKED_ON_EXTERNAL_DEPLOYMENT_STATUS_REASON } from "../systems/pendingVersionSystem.js";
import type {
PendingVersionRunIdLookup,
PendingVersionRunIdLookupOptions,
@@ -32,6 +33,21 @@ export class PostgresPendingVersionRunIdLookup implements PendingVersionRunIdLoo
status: "PENDING_VERSION",
taskIdentifier: { in: options.taskIdentifiers },
queue: { in: options.queues },
OR: [
{ statusReason: null },
{ statusReason: { not: PARKED_ON_EXTERNAL_DEPLOYMENT_STATUS_REASON } },
...(options.externalDeploymentId
? [
{
statusReason: PARKED_ON_EXTERNAL_DEPLOYMENT_STATUS_REASON,
annotations: {
path: ["externalDeploymentId"],
equals: options.externalDeploymentId,
},
},
]
: []),
],
},
select: { id: true },
orderBy: { createdAt: "asc" },
@@ -249,6 +249,7 @@ export type RunEngineOptions = {
* to disable lag-aware retries entirely.
*/
pendingVersionLagMaxRetries?: number;
externalDeploymentParkDeadlineMs?: number;
/** Optional maximum TTL for all runs (e.g. "14d"). If set, runs without an explicit TTL
* will use this as their TTL, and runs with a TTL larger than this will be clamped. */
defaultMaxTtl?: string;
@@ -372,7 +373,9 @@ export type TriggerParams = {
triggerAction: string;
rootTriggerSource: string;
rootScheduleId?: string;
externalDeploymentId?: string;
};
parkedOnExternalDeploymentId?: string;
/**
* Called when a run is debounced (existing delayed run found with triggerAndWait).
* Return spanIdToComplete to enable span closing when the run completes.
@@ -52,6 +52,13 @@ export const workerCatalog = {
}),
visibilityTimeoutMs: 60_000,
},
expireParkedExternalDeploymentRun: {
schema: z.object({
runId: z.string(),
externalDeploymentId: z.string(),
}),
visibilityTimeoutMs: 60_000,
},
tryCompleteBatch: {
schema: z.object({
batchId: z.string(),
@@ -5,6 +5,7 @@ export {
ServiceValidationError as EngineServiceValidationError,
} from "./engine/errors.js";
export type { EventBusEventArgs, EventBusEvents } from "./engine/eventBus.js";
export { PARKED_ON_EXTERNAL_DEPLOYMENT_STATUS_REASON } from "./engine/systems/pendingVersionSystem.js";
export type { AuthenticatedEnvironment } from "./shared/index.js";
export type {
PendingVersionRunIdLookup,
@@ -22,6 +22,7 @@ import type {
ForWaitpointCompletionContext,
IdempotencyKeyRunMatch,
LockRunData,
PromotePendingVersionArgs,
ReadClient,
RescheduleSnapshotInput,
RewriteDebouncedRunData,
@@ -1321,18 +1322,77 @@ export class PostgresRunStore implements RunStore {
async promotePendingVersionRuns(
runId: string,
args?: PromotePendingVersionArgs,
tx?: PrismaClientOrTransaction
): Promise<{ count: number }> {
const prisma = tx ?? this.prisma;
const result = await prisma.taskRun.updateMany({
where: { id: runId, status: "PENDING_VERSION" },
data: { status: "PENDING" },
data: {
status: args?.status ?? "PENDING",
lockedToVersionId: args?.lockedToVersionId,
taskVersion: args?.taskVersion,
sdkVersion: args?.sdkVersion,
cliVersion: args?.cliVersion,
},
});
return { count: result.count };
}
async expireParkedRun(
runId: string,
data: {
error: TaskRunError;
completedAt: Date;
expiredAt: Date;
statusReason: string;
snapshot: ExpireSnapshotInput;
},
tx?: PrismaClientOrTransaction
): Promise<{ count: number }> {
const prisma = tx ?? this.prisma;
try {
await prisma.taskRun.update({
where: { id: runId, status: "PENDING_VERSION" },
data: {
status: "EXPIRED",
statusReason: data.statusReason,
completedAt: data.completedAt,
expiredAt: data.expiredAt,
error: data.error as Prisma.InputJsonValue,
executionSnapshots: {
create: {
engine: data.snapshot.engine,
executionStatus: data.snapshot.executionStatus,
description: data.snapshot.description,
runStatus: data.snapshot.runStatus,
environmentId: data.snapshot.environmentId,
environmentType: data.snapshot.environmentType,
projectId: data.snapshot.projectId,
organizationId: data.snapshot.organizationId,
},
},
},
});
} catch (error) {
const normalized = normalizeRunOpsError(error);
if (
normalized instanceof Prisma.PrismaClientKnownRequestError &&
normalized.code === "P2025"
) {
return { count: 0 };
}
throw error;
}
return { count: 1 };
}
async suspendForCheckpoint<I extends Prisma.TaskRunInclude>(
runId: string,
args: { include: I },
+17 -1
View File
@@ -22,6 +22,7 @@ import type {
ForWaitpointCompletionContext,
IdempotencyKeyRunMatch,
LockRunData,
PromotePendingVersionArgs,
ReadClient,
RescheduleSnapshotInput,
RewriteDebouncedRunData,
@@ -654,9 +655,24 @@ export class RoutingRunStore implements RunStore {
async promotePendingVersionRuns(
runId: string,
args?: PromotePendingVersionArgs,
tx?: PrismaClientOrTransaction
): Promise<{ count: number }> {
return (await this.#routeForWrite(runId)).promotePendingVersionRuns(runId);
return (await this.#routeForWrite(runId)).promotePendingVersionRuns(runId, args);
}
async expireParkedRun(
runId: string,
data: {
error: TaskRunError;
completedAt: Date;
expiredAt: Date;
statusReason: string;
snapshot: ExpireSnapshotInput;
},
tx?: PrismaClientOrTransaction
): Promise<{ count: number }> {
return (await this.#routeForWrite(runId)).expireParkedRun(runId, data);
}
async suspendForCheckpoint<I extends Prisma.TaskRunInclude>(
+21
View File
@@ -55,6 +55,14 @@ export type CompletionSnapshotInput = {
runnerId?: string;
};
export type PromotePendingVersionArgs = {
status?: Extract<TaskRunStatus, "PENDING" | "DELAYED">;
lockedToVersionId?: string;
taskVersion?: string;
sdkVersion?: string;
cliVersion?: string;
};
export type ExpireSnapshotInput = {
engine: "V2";
executionStatus: "FINISHED";
@@ -105,6 +113,7 @@ export type CreateRunData = {
id: string;
engine: "V2";
status: TaskRunStatus;
statusReason?: string;
friendlyId: string;
runtimeEnvironmentId: string;
environmentType: RuntimeEnvironmentType;
@@ -486,6 +495,18 @@ export interface RunStore {
): Promise<Prisma.TaskRunGetPayload<{ select: S }>>;
promotePendingVersionRuns(
runId: string,
args?: PromotePendingVersionArgs,
tx?: PrismaClientOrTransaction
): Promise<{ count: number }>;
expireParkedRun(
runId: string,
data: {
error: TaskRunError;
completedAt: Date;
expiredAt: Date;
statusReason: string;
snapshot: ExpireSnapshotInput;
},
tx?: PrismaClientOrTransaction
): Promise<{ count: number }>;
suspendForCheckpoint<I extends Prisma.TaskRunInclude>(
@@ -85,6 +85,14 @@ export class APIClientManagerAPI {
return value ? value : undefined;
}
get externalDeploymentId(): string | undefined {
const scoped = sdkScope.getStore();
if (scoped) {
return scoped.apiClientConfig.externalDeploymentId;
}
return this.#getConfig()?.externalDeploymentId;
}
public resolveApiClientConfig(partial: ApiClientConfiguration = {}): ApiClientConfiguration {
return {
baseURL: partial.baseURL ?? getEnvVar("TRIGGER_API_URL"),
@@ -99,6 +107,7 @@ export class APIClientManagerAPI {
getEnvVar("TRIGGER_PREVIEW_BRANCH") ??
getEnvVar("VERCEL_GIT_COMMIT_REF") ??
getDevBranchEnvVar(),
externalDeploymentId: partial.externalDeploymentId,
requestOptions: partial.requestOptions,
future: partial.future,
};
@@ -14,6 +14,12 @@ export type ApiClientConfiguration = {
* The preview branch name (for preview environments)
*/
previewBranch?: string;
/**
* Pin every run triggered through this client to the deployment deployed under this
* external id. An explicit `externalDeploymentId` on an individual trigger wins over it;
* it in turn wins over `TRIGGER_EXTERNAL_DEPLOYMENT_ID` and platform discovery.
*/
externalDeploymentId?: string;
requestOptions?: ApiRequestOptions;
future?: ApiClientFutureFlags;
};
@@ -0,0 +1,25 @@
export function compareDeploymentVersions(versionA: string, versionB: string): number {
const [dateA = "", numberA] = versionA.split(".");
const [dateB = "", numberB] = versionB.split(".");
if (dateA < dateB) {
return -1;
}
if (dateA > dateB) {
return 1;
}
const numA = Number(numberA);
const numB = Number(numberB);
if (numA < numB) {
return -1;
}
if (numA > numB) {
return 1;
}
return 0;
}
+1
View File
@@ -6,3 +6,4 @@ export * from "./queueName.js";
export * from "./consts.js";
export * from "./traceContext.js";
export * from "./dates.js";
export * from "./deploymentVersions.js";
+22
View File
@@ -917,6 +917,28 @@ export type TriggerOptions = {
*/
version?: string;
/**
* Pin this run to the deployment that was deployed under this external id a commit SHA,
* a CI run id, a release tag matching `trigger.dev deploy --external-id`.
*
* Use this when the code making the call and the tasks it triggers must be the same
* release. If nothing has been deployed under the id yet the run waits for it rather than
* running on the wrong version, and gives up after an hour if it never arrives.
*
* Usually you don't set this by hand: the SDK reads `TRIGGER_EXTERNAL_DEPLOYMENT_ID`, and
* with `TRIGGER_AUTOMATIC_SKEW_VERSION_PROTECTION=1` it discovers your hosting platform's
* commit variable automatically. Setting it here always wins over both.
*
* `version` (and the `TRIGGER_VERSION` environment variable) take precedence over this.
*
* @example
*
* ```ts
* await myTask.trigger({ foo: "bar" }, { externalDeploymentId: process.env.VERCEL_GIT_COMMIT_SHA });
* ```
*/
externalDeploymentId?: string;
/**
* Specify the region to run the task in. This overrides the default region set for your project in the dashboard.
*
+28
View File
@@ -23,6 +23,7 @@ import {
lifecycleHooks,
makeIdempotencyKey,
packetRequiresOffloading,
resolveExternalDeploymentId,
parsePacket,
RateLimitError,
resourceCatalog,
@@ -125,6 +126,14 @@ function scopedEnvVar(name: string): string | undefined {
return getEnvVar(name);
}
function resolveTriggerExternalDeploymentId(explicit?: string): string | undefined {
return resolveExternalDeploymentId({
explicit,
clientConfig: apiClientManager.externalDeploymentId,
read: scopedEnvVar,
});
}
export function queue(options: QueueOptions): Queue {
resourceCatalog.registerQueueMetadata(options);
@@ -737,6 +746,9 @@ export async function batchTriggerById<TTask extends AnyTask>(
priority: item.options?.priority,
region: item.options?.region,
lockToVersion: item.options?.version ?? scopedEnvVar("TRIGGER_VERSION"),
externalDeploymentId: resolveTriggerExternalDeploymentId(
item.options?.externalDeploymentId
),
debounce: item.options?.debounce,
},
} satisfies BatchItemNDJSON;
@@ -1259,6 +1271,9 @@ export async function batchTriggerTasks<TTasks extends readonly AnyTask[]>(
priority: item.options?.priority,
region: item.options?.region,
lockToVersion: item.options?.version ?? scopedEnvVar("TRIGGER_VERSION"),
externalDeploymentId: resolveTriggerExternalDeploymentId(
item.options?.externalDeploymentId
),
debounce: item.options?.debounce,
},
} satisfies BatchItemNDJSON;
@@ -2007,6 +2022,9 @@ async function* transformBatchItemsStream<TTask extends AnyTask>(
priority: item.options?.priority,
region: item.options?.region,
lockToVersion: item.options?.version ?? scopedEnvVar("TRIGGER_VERSION"),
externalDeploymentId: resolveTriggerExternalDeploymentId(
item.options?.externalDeploymentId
),
debounce: item.options?.debounce,
},
} satisfies BatchItemNDJSON;
@@ -2110,6 +2128,9 @@ async function* transformBatchByTaskItemsStream<TTasks extends readonly AnyTask[
priority: item.options?.priority,
region: item.options?.region,
lockToVersion: item.options?.version ?? scopedEnvVar("TRIGGER_VERSION"),
externalDeploymentId: resolveTriggerExternalDeploymentId(
item.options?.externalDeploymentId
),
debounce: item.options?.debounce,
},
} satisfies BatchItemNDJSON;
@@ -2214,6 +2235,9 @@ async function* transformSingleTaskBatchItemsStream<TPayload>(
priority: item.options?.priority,
region: item.options?.region,
lockToVersion: item.options?.version ?? scopedEnvVar("TRIGGER_VERSION"),
externalDeploymentId: resolveTriggerExternalDeploymentId(
item.options?.externalDeploymentId
),
debounce: item.options?.debounce,
},
} satisfies BatchItemNDJSON;
@@ -2327,6 +2351,7 @@ async function trigger_internal<TRunTypes extends AnyRunTypes>(
priority: options?.priority,
region: options?.region,
lockToVersion: options?.version ?? scopedEnvVar("TRIGGER_VERSION"),
externalDeploymentId: resolveTriggerExternalDeploymentId(options?.externalDeploymentId),
debounce: options?.debounce,
},
},
@@ -2413,6 +2438,9 @@ async function batchTrigger_internal<TRunTypes extends AnyRunTypes>(
priority: item.options?.priority,
region: item.options?.region,
lockToVersion: item.options?.version ?? scopedEnvVar("TRIGGER_VERSION"),
externalDeploymentId: resolveTriggerExternalDeploymentId(
item.options?.externalDeploymentId
),
debounce: item.options?.debounce,
},
} satisfies BatchItemNDJSON;