diff --git a/.changeset/trigger-external-deployment-id.md b/.changeset/trigger-external-deployment-id.md new file mode 100644 index 0000000000..11d3daac87 --- /dev/null +++ b/.changeset/trigger-external-deployment-id.md @@ -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. diff --git a/apps/webapp/app/env.server.ts b/apps/webapp/app/env.server.ts index e5dfe86307..07f2122b64 100644 --- a/apps/webapp/app/env.server.ts +++ b/apps/webapp/app/env.server.ts @@ -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), diff --git a/apps/webapp/app/runEngine/services/triggerTask.server.ts b/apps/webapp/app/runEngine/services/triggerTask.server.ts index 6d15c55435..8e9e99d7f0 100644 --- a/apps/webapp/app/runEngine/services/triggerTask.server.ts +++ b/apps/webapp/app/runEngine/services/triggerTask.server.ts @@ -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 { @@ -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, }; } diff --git a/apps/webapp/app/services/externalDeploymentCache.server.ts b/apps/webapp/app/services/externalDeploymentCache.server.ts new file mode 100644 index 0000000000..34270fa7c4 --- /dev/null +++ b/apps/webapp/app/services/externalDeploymentCache.server.ts @@ -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; + setIfNewer( + environmentId: string, + externalId: string, + entry: ExternalDeploymentCacheEntry + ): Promise; + setMissing(environmentId: string, externalId: string): Promise; +} + +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 & { 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 { + skewIdSetIfNewer( + key: string, + entry: string, + version: string, + ttlSeconds: string, + callback?: Callback + ): Result; + } +} + +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 { + 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 { + 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 { + 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 { + return null; + } + + async setIfNewer(): Promise {} + + async setMissing(): Promise {} +} diff --git a/apps/webapp/app/services/externalDeploymentCacheInstance.server.ts b/apps/webapp/app/services/externalDeploymentCacheInstance.server.ts new file mode 100644 index 0000000000..e62518612e --- /dev/null +++ b/apps/webapp/app/services/externalDeploymentCacheInstance.server.ts @@ -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, + }); +} diff --git a/apps/webapp/app/services/runsReplicationService.server.ts b/apps/webapp/app/services/runsReplicationService.server.ts index 5604c107a6..ace3029abf 100644 --- a/apps/webapp/app/services/runsReplicationService.server.ts +++ b/apps/webapp/app/services/runsReplicationService.server.ts @@ -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).externalDeploymentId; + + return typeof value === "string" ? value : ""; + } + async #preparePayloadInsert(run: TaskRun, _version: bigint): Promise { const payload = await this.#prepareJson(run.payload, run.payloadType); diff --git a/apps/webapp/app/v3/runEngine.server.ts b/apps/webapp/app/v3/runEngine.server.ts index 8598693329..76ac35f349 100644 --- a/apps/webapp/app/v3/runEngine.server.ts +++ b/apps/webapp/app/v3/runEngine.server.ts @@ -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); diff --git a/apps/webapp/app/v3/services/clickhousePendingVersionLookup.server.ts b/apps/webapp/app/v3/services/clickhousePendingVersionLookup.server.ts index fde0e9bcaa..ca66ca8e94 100644 --- a/apps/webapp/app/v3/services/clickhousePendingVersionLookup.server.ts +++ b/apps/webapp/app/v3/services/clickhousePendingVersionLookup.server.ts @@ -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(); diff --git a/apps/webapp/app/v3/services/finalizeDeployment.server.ts b/apps/webapp/app/v3/services/finalizeDeployment.server.ts index fd559aff30..0595cee1e2 100644 --- a/apps/webapp/app/v3/services/finalizeDeployment.server.ts +++ b/apps/webapp/app/v3/services/finalizeDeployment.server.ts @@ -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) diff --git a/apps/webapp/app/v3/services/resolveExternalDeployment.server.ts b/apps/webapp/app/v3/services/resolveExternalDeployment.server.ts new file mode 100644 index 0000000000..7b93740dd9 --- /dev/null +++ b/apps/webapp/app/v3/services/resolveExternalDeployment.server.ts @@ -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 { + 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 { + 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; +} diff --git a/apps/webapp/app/v3/utils/deploymentVersions.ts b/apps/webapp/app/v3/utils/deploymentVersions.ts index 4e792599bc..2cdcf10df4 100644 --- a/apps/webapp/app/v3/utils/deploymentVersions.ts +++ b/apps/webapp/app/v3/utils/deploymentVersions.ts @@ -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"; diff --git a/apps/webapp/test/engine/triggerTask.externalDeploymentId.test.ts b/apps/webapp/test/engine/triggerTask.externalDeploymentId.test.ts new file mode 100644 index 0000000000..d6dba9ccdd --- /dev/null +++ b/apps/webapp/test/engine/triggerTask.externalDeploymentId.test.ts @@ -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; + 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()) {} + + 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).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).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).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).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(); + } + ); +}); diff --git a/apps/webapp/test/externalDeploymentCache.test.ts b/apps/webapp/test/externalDeploymentCache.test.ts new file mode 100644 index 0000000000..3e49099575 --- /dev/null +++ b/apps/webapp/test/externalDeploymentCache.test.ts @@ -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(); + } + }); +}); diff --git a/apps/webapp/test/performTaskRunAlertsStoreRouting.test.ts b/apps/webapp/test/performTaskRunAlertsStoreRouting.test.ts index a7e0520df4..141b8274a8 100644 --- a/apps/webapp/test/performTaskRunAlertsStoreRouting.test.ts +++ b/apps/webapp/test/performTaskRunAlertsStoreRouting.test.ts @@ -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); } diff --git a/apps/webapp/test/runsReplicationServiceExternalDeploymentId.test.ts b/apps/webapp/test/runsReplicationServiceExternalDeploymentId.test.ts new file mode 100644 index 0000000000..9879fa0787 --- /dev/null +++ b/apps/webapp/test/runsReplicationServiceExternalDeploymentId.test.ts @@ -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( + (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(); + } + ); +}); diff --git a/apps/webapp/test/updateMetadataStoreRoutingHetero.test.ts b/apps/webapp/test/updateMetadataStoreRoutingHetero.test.ts index 764ff6f1ae..17d6ec9fc2 100644 --- a/apps/webapp/test/updateMetadataStoreRoutingHetero.test.ts +++ b/apps/webapp/test/updateMetadataStoreRoutingHetero.test.ts @@ -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); diff --git a/internal-packages/clickhouse/src/taskRuns.test.ts b/internal-packages/clickhouse/src/taskRuns.test.ts index 3595146a98..bc219cd2ba 100644 --- a/internal-packages/clickhouse/src/taskRuns.test.ts +++ b/internal-packages/clickhouse/src/taskRuns.test.ts @@ -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]; diff --git a/internal-packages/clickhouse/src/taskRuns.ts b/internal-packages/clickhouse/src/taskRuns.ts index f561be04d9..01818b4202 100644 --- a/internal-packages/clickhouse/src/taskRuns.ts +++ b/internal-packages/clickhouse/src/taskRuns.ts @@ -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, ]; /** diff --git a/internal-packages/run-engine/src/engine/index.ts b/internal-packages/run-engine/src/engine/index.ts index 2f7447af71..a9c6ea5b78 100644 --- a/internal-packages/run-engine/src/engine/index.ts +++ b/internal-packages/run-engine/src/engine/index.ts @@ -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; diff --git a/internal-packages/run-engine/src/engine/services/pendingVersionLookup.ts b/internal-packages/run-engine/src/engine/services/pendingVersionLookup.ts index 62ce67f4d9..bfd45643d3 100644 --- a/internal-packages/run-engine/src/engine/services/pendingVersionLookup.ts +++ b/internal-packages/run-engine/src/engine/services/pendingVersionLookup.ts @@ -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; }; diff --git a/internal-packages/run-engine/src/engine/systems/debounceSystem.ts b/internal-packages/run-engine/src/engine/systems/debounceSystem.ts index 5cf5ba35b5..3d8def2c24 100644 --- a/internal-packages/run-engine/src/engine/systems/debounceSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/debounceSystem.ts @@ -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, diff --git a/internal-packages/run-engine/src/engine/systems/pendingVersionSystem.test.ts b/internal-packages/run-engine/src/engine/systems/pendingVersionSystem.test.ts index 9604613bcb..16a0510487 100644 --- a/internal-packages/run-engine/src/engine/systems/pendingVersionSystem.test.ts +++ b/internal-packages/run-engine/src/engine/systems/pendingVersionSystem.test.ts @@ -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. diff --git a/internal-packages/run-engine/src/engine/systems/pendingVersionSystem.ts b/internal-packages/run-engine/src/engine/systems/pendingVersionSystem.ts index 1984c82ef2..d110874390 100644 --- a/internal-packages/run-engine/src/engine/systems/pendingVersionSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/pendingVersionSystem.ts @@ -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).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 { + 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 { + 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 { + 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 { + 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 { + 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 { const maxRetries = this.options.lagMaxRetries ?? DEFAULT_LAG_MAX_RETRIES; diff --git a/internal-packages/run-engine/src/engine/tests/externalDeploymentParking.test.ts b/internal-packages/run-engine/src/engine/tests/externalDeploymentParking.test.ts new file mode 100644 index 0000000000..f815b1b863 --- /dev/null +++ b/internal-packages/run-engine/src/engine/tests/externalDeploymentParking.test.ts @@ -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 +) { + 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).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(); + } + } + ); +}); diff --git a/internal-packages/run-engine/src/engine/tests/postgresPendingVersionLookup.ts b/internal-packages/run-engine/src/engine/tests/postgresPendingVersionLookup.ts index e15afdb00a..bfbb8884c9 100644 --- a/internal-packages/run-engine/src/engine/tests/postgresPendingVersionLookup.ts +++ b/internal-packages/run-engine/src/engine/tests/postgresPendingVersionLookup.ts @@ -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" }, diff --git a/internal-packages/run-engine/src/engine/types.ts b/internal-packages/run-engine/src/engine/types.ts index 9b7a3b1b8f..2516776373 100644 --- a/internal-packages/run-engine/src/engine/types.ts +++ b/internal-packages/run-engine/src/engine/types.ts @@ -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. diff --git a/internal-packages/run-engine/src/engine/workerCatalog.ts b/internal-packages/run-engine/src/engine/workerCatalog.ts index 8f49d7d606..5def3071b9 100644 --- a/internal-packages/run-engine/src/engine/workerCatalog.ts +++ b/internal-packages/run-engine/src/engine/workerCatalog.ts @@ -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(), diff --git a/internal-packages/run-engine/src/index.ts b/internal-packages/run-engine/src/index.ts index d8eaadf649..b7898e6433 100644 --- a/internal-packages/run-engine/src/index.ts +++ b/internal-packages/run-engine/src/index.ts @@ -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, diff --git a/internal-packages/run-store/src/PostgresRunStore.ts b/internal-packages/run-store/src/PostgresRunStore.ts index ec3943cb11..b3b5c565f0 100644 --- a/internal-packages/run-store/src/PostgresRunStore.ts +++ b/internal-packages/run-store/src/PostgresRunStore.ts @@ -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( runId: string, args: { include: I }, diff --git a/internal-packages/run-store/src/runOpsStore.ts b/internal-packages/run-store/src/runOpsStore.ts index 28df27a4f2..f8de40b7df 100644 --- a/internal-packages/run-store/src/runOpsStore.ts +++ b/internal-packages/run-store/src/runOpsStore.ts @@ -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( diff --git a/internal-packages/run-store/src/types.ts b/internal-packages/run-store/src/types.ts index ac7cc35e3c..c09522ccd0 100644 --- a/internal-packages/run-store/src/types.ts +++ b/internal-packages/run-store/src/types.ts @@ -55,6 +55,14 @@ export type CompletionSnapshotInput = { runnerId?: string; }; +export type PromotePendingVersionArgs = { + status?: Extract; + 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>; 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( diff --git a/packages/core/src/v3/apiClientManager/index.ts b/packages/core/src/v3/apiClientManager/index.ts index cf9f934891..e9ec9fd1e4 100644 --- a/packages/core/src/v3/apiClientManager/index.ts +++ b/packages/core/src/v3/apiClientManager/index.ts @@ -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, }; diff --git a/packages/core/src/v3/apiClientManager/types.ts b/packages/core/src/v3/apiClientManager/types.ts index 8cdb185146..fe73818227 100644 --- a/packages/core/src/v3/apiClientManager/types.ts +++ b/packages/core/src/v3/apiClientManager/types.ts @@ -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; }; diff --git a/packages/core/src/v3/isomorphic/deploymentVersions.ts b/packages/core/src/v3/isomorphic/deploymentVersions.ts new file mode 100644 index 0000000000..c3b5617991 --- /dev/null +++ b/packages/core/src/v3/isomorphic/deploymentVersions.ts @@ -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; +} diff --git a/packages/core/src/v3/isomorphic/index.ts b/packages/core/src/v3/isomorphic/index.ts index e51135f94c..3f37285473 100644 --- a/packages/core/src/v3/isomorphic/index.ts +++ b/packages/core/src/v3/isomorphic/index.ts @@ -6,3 +6,4 @@ export * from "./queueName.js"; export * from "./consts.js"; export * from "./traceContext.js"; export * from "./dates.js"; +export * from "./deploymentVersions.js"; diff --git a/packages/core/src/v3/types/tasks.ts b/packages/core/src/v3/types/tasks.ts index 253e09619e..224b7604db 100644 --- a/packages/core/src/v3/types/tasks.ts +++ b/packages/core/src/v3/types/tasks.ts @@ -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. * diff --git a/packages/trigger-sdk/src/v3/shared.ts b/packages/trigger-sdk/src/v3/shared.ts index 13125b305c..69b51f0ed3 100644 --- a/packages/trigger-sdk/src/v3/shared.ts +++ b/packages/trigger-sdk/src/v3/shared.ts @@ -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( 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( 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( 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( 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( 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( 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;