From 424a68ed2e616e0611ee145986a355a7417877ee Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 2 Feb 2026 15:49:15 +0000 Subject: [PATCH] feat(run-engine): wire up TTL system callback and compute ttlExpiresAt on enqueue - Add TTL system options to RunEngineOptions.queue for configuration - Add #ttlExpiredCallback method on RunEngine that calls ttlSystem.expireRun() - Pass TTL system options to RunQueue in RunEngine constructor - Compute ttlExpiresAt from run.ttl when enqueuing runs in EnqueueSystem - Add env vars for TTL system configuration (disabled, shard count, poll interval, batch size) - Configure TTL system in webapp's runEngine.server.ts The TTL system enables automatic expiration of runs that have been in the queue past their TTL deadline. When runs expire, the callback updates their status to EXPIRED in the database and emits appropriate events. https://claude.ai/code/session_01AyzQp6tbj7th5QRTCYjJR5 --- apps/webapp/app/env.server.ts | 6 +++ apps/webapp/app/v3/runEngine.server.ts | 6 +++ .../run-engine/src/engine/index.ts | 37 +++++++++++++++++++ .../src/engine/systems/enqueueSystem.ts | 11 ++++++ .../run-engine/src/engine/types.ts | 11 ++++++ 5 files changed, 71 insertions(+) diff --git a/apps/webapp/app/env.server.ts b/apps/webapp/app/env.server.ts index dcbcac079..885232dd1 100644 --- a/apps/webapp/app/env.server.ts +++ b/apps/webapp/app/env.server.ts @@ -591,6 +591,12 @@ const EnvironmentSchema = z RUN_ENGINE_CONCURRENCY_SWEEPER_SCAN_JITTER_IN_MS: z.coerce.number().int().optional(), RUN_ENGINE_CONCURRENCY_SWEEPER_PROCESS_MARKED_JITTER_IN_MS: z.coerce.number().int().optional(), + // TTL System settings for automatic run expiration + RUN_ENGINE_TTL_SYSTEM_DISABLED: BoolEnv.default(false), + RUN_ENGINE_TTL_SYSTEM_SHARD_COUNT: z.coerce.number().int().optional(), + RUN_ENGINE_TTL_SYSTEM_POLL_INTERVAL_MS: z.coerce.number().int().default(1_000), + RUN_ENGINE_TTL_SYSTEM_BATCH_SIZE: z.coerce.number().int().default(100), + RUN_ENGINE_RUN_LOCK_DURATION: z.coerce.number().int().default(5000), RUN_ENGINE_RUN_LOCK_AUTOMATIC_EXTENSION_THRESHOLD: z.coerce.number().int().default(1000), RUN_ENGINE_RUN_LOCK_MAX_RETRIES: z.coerce.number().int().default(10), diff --git a/apps/webapp/app/v3/runEngine.server.ts b/apps/webapp/app/v3/runEngine.server.ts index efba5fbdb..b0dc1e8d0 100644 --- a/apps/webapp/app/v3/runEngine.server.ts +++ b/apps/webapp/app/v3/runEngine.server.ts @@ -80,6 +80,12 @@ function createRunEngine() { scanJitterInMs: env.RUN_ENGINE_CONCURRENCY_SWEEPER_SCAN_JITTER_IN_MS, processMarkedJitterInMs: env.RUN_ENGINE_CONCURRENCY_SWEEPER_PROCESS_MARKED_JITTER_IN_MS, }, + ttlSystem: { + disabled: env.RUN_ENGINE_TTL_SYSTEM_DISABLED, + shardCount: env.RUN_ENGINE_TTL_SYSTEM_SHARD_COUNT, + pollIntervalMs: env.RUN_ENGINE_TTL_SYSTEM_POLL_INTERVAL_MS, + batchSize: env.RUN_ENGINE_TTL_SYSTEM_BATCH_SIZE, + }, }, runLock: { redis: { diff --git a/internal-packages/run-engine/src/engine/index.ts b/internal-packages/run-engine/src/engine/index.ts index 8a4e04a18..08f0d7849 100644 --- a/internal-packages/run-engine/src/engine/index.ts +++ b/internal-packages/run-engine/src/engine/index.ts @@ -182,6 +182,14 @@ export class RunEngine { processWorkerQueueDebounceMs: options.queue?.processWorkerQueueDebounceMs, dequeueBlockingTimeoutSeconds: options.queue?.dequeueBlockingTimeoutSeconds, meter: options.meter, + ttlSystem: options.queue?.ttlSystem?.disabled + ? undefined + : { + shardCount: options.queue?.ttlSystem?.shardCount, + pollIntervalMs: options.queue?.ttlSystem?.pollIntervalMs, + batchSize: options.queue?.ttlSystem?.batchSize, + callback: this.#ttlExpiredCallback.bind(this), + }, // Run data provider for V3 optimized format - reads from PostgreSQL when no Redis message key exists runDataProvider: { getRunData: async (runId: string) => { @@ -2103,6 +2111,35 @@ export class RunEngine { })); } + /** + * Callback for the TTL system when runs expire. + * Calls ttlSystem.expireRun() for each expired run to update database and emit events. + */ + async #ttlExpiredCallback( + runs: Array<{ queueKey: string; runId: string; orgId: string }> + ): Promise { + // Process expired runs concurrently with limited parallelism + await pMap( + runs, + async (run) => { + try { + await this.ttlSystem.expireRun({ runId: run.runId }); + this.logger.debug("TTL system expired run", { + runId: run.runId, + orgId: run.orgId, + }); + } catch (error) { + this.logger.error("Failed to expire run via TTL system", { + runId: run.runId, + orgId: run.orgId, + error, + }); + } + }, + { concurrency: 10 } + ); + } + /** * Invalidates the billing cache for an organization when their plan changes * Runs in background and handles all errors internally diff --git a/internal-packages/run-engine/src/engine/systems/enqueueSystem.ts b/internal-packages/run-engine/src/engine/systems/enqueueSystem.ts index 395e44727..4726bdb73 100644 --- a/internal-packages/run-engine/src/engine/systems/enqueueSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/enqueueSystem.ts @@ -4,6 +4,7 @@ import { TaskRun, TaskRunExecutionStatus, } from "@trigger.dev/database"; +import { parseNaturalLanguageDuration } from "@trigger.dev/core/v3/isomorphic"; import { MinimalAuthenticatedEnvironment } from "../../shared/index.js"; import { ExecutionSnapshotSystem } from "./executionSnapshotSystem.js"; import { SystemResources } from "./systems.js"; @@ -81,6 +82,15 @@ export class EnqueueSystem { const timestamp = (run.queueTimestamp ?? run.createdAt).getTime() - run.priorityMs; + // Calculate TTL expiration timestamp if the run has a TTL + let ttlExpiresAt: number | undefined; + if (run.ttl) { + const expireAt = parseNaturalLanguageDuration(run.ttl); + if (expireAt) { + ttlExpiresAt = expireAt.getTime(); + } + } + await this.$.runQueue.enqueueMessage({ env, workerQueue, @@ -95,6 +105,7 @@ export class EnqueueSystem { concurrencyKey: run.concurrencyKey ?? undefined, timestamp, attempt: 0, + ttlExpiresAt, }, }); diff --git a/internal-packages/run-engine/src/engine/types.ts b/internal-packages/run-engine/src/engine/types.ts index ee5176c2f..a9e188f2a 100644 --- a/internal-packages/run-engine/src/engine/types.ts +++ b/internal-packages/run-engine/src/engine/types.ts @@ -63,6 +63,17 @@ export type RunEngineOptions = { scanJitterInMs?: number; processMarkedJitterInMs?: number; }; + /** TTL system options for automatic run expiration */ + ttlSystem?: { + /** Number of shards for TTL sorted sets (default: same as queue shards) */ + shardCount?: number; + /** How often to poll each shard for expired runs (ms, default: 1000) */ + pollIntervalMs?: number; + /** Max number of runs to expire per poll per shard (default: 100) */ + batchSize?: number; + /** Whether TTL consumers are disabled (default: false) */ + disabled?: boolean; + }; }; runLock: { redis: RedisOptions;