github-featured/triggerdotdev--trigger.dev
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main' into v4/helm

Browse Source
This commit is contained in:
nicktrn
2025-06-25 14:03:38 +01:00
parent 6a1bab35d9 aaf40c3be3
commit 9c01c2089b
13 changed files with 136 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/webapp/app/env.server.ts
+7 -2
View File
@@ -618,6 +618,9 @@ const EnvironmentSchema = z.object({
LEGACY_RUN_ENGINE_WORKER_IMMEDIATE_POLL_INTERVAL: z.coerce.number().int().default(50),
LEGACY_RUN_ENGINE_WORKER_CONCURRENCY_LIMIT: z.coerce.number().int().default(100),
LEGACY_RUN_ENGINE_WORKER_SHUTDOWN_TIMEOUT_MS: z.coerce.number().int().default(60_000),
LEGACY_RUN_ENGINE_WORKER_LOG_LEVEL: z
.enum(["log", "error", "warn", "info", "debug"])
.default("info"),
LEGACY_RUN_ENGINE_WORKER_REDIS_HOST: z
.string()
@@ -661,6 +664,7 @@ const EnvironmentSchema = z.object({
COMMON_WORKER_IMMEDIATE_POLL_INTERVAL: z.coerce.number().int().default(50),
COMMON_WORKER_CONCURRENCY_LIMIT: z.coerce.number().int().default(100),
COMMON_WORKER_SHUTDOWN_TIMEOUT_MS: z.coerce.number().int().default(60_000),
COMMON_WORKER_LOG_LEVEL: z.enum(["log", "error", "warn", "info", "debug"]).default("info"),
COMMON_WORKER_REDIS_HOST: z
.string()
@@ -699,6 +703,7 @@ const EnvironmentSchema = z.object({
ALERTS_WORKER_IMMEDIATE_POLL_INTERVAL: z.coerce.number().int().default(100),
ALERTS_WORKER_CONCURRENCY_LIMIT: z.coerce.number().int().default(100),
ALERTS_WORKER_SHUTDOWN_TIMEOUT_MS: z.coerce.number().int().default(60_000),
ALERTS_WORKER_LOG_LEVEL: z.enum(["log", "error", "warn", "info", "debug"]).default("info"),
ALERTS_WORKER_REDIS_HOST: z
.string()
@@ -732,8 +737,8 @@ const EnvironmentSchema = z.object({
SCHEDULE_ENGINE_LOG_LEVEL: z.enum(["log", "error", "warn", "info", "debug"]).default("info"),
SCHEDULE_WORKER_ENABLED: z.string().default(process.env.WORKER_ENABLED ?? "true"),
SCHEDULE_WORKER_CONCURRENCY_WORKERS: z.coerce.number().int().default(1),
SCHEDULE_WORKER_CONCURRENCY_TASKS_PER_WORKER: z.coerce.number().int().default(1),
SCHEDULE_WORKER_CONCURRENCY_WORKERS: z.coerce.number().int().default(2),
SCHEDULE_WORKER_CONCURRENCY_TASKS_PER_WORKER: z.coerce.number().int().default(10),
SCHEDULE_WORKER_POLL_INTERVAL: z.coerce.number().int().default(1000),
SCHEDULE_WORKER_IMMEDIATE_POLL_INTERVAL: z.coerce.number().int().default(50),
SCHEDULE_WORKER_CONCURRENCY_LIMIT: z.coerce.number().int().default(50),
apps/webapp/app/v3/alertsWorker.server.ts
+1 -1
View File
@@ -61,7 +61,7 @@ function initializeWorker() {
pollIntervalMs: env.ALERTS_WORKER_POLL_INTERVAL,
immediatePollIntervalMs: env.ALERTS_WORKER_IMMEDIATE_POLL_INTERVAL,
shutdownTimeoutMs: env.ALERTS_WORKER_SHUTDOWN_TIMEOUT_MS,
logger: new Logger("AlertsWorker", "debug"),
logger: new Logger("AlertsWorker", env.ALERTS_WORKER_LOG_LEVEL),
jobs: {
"v3.deliverAlert": async ({ payload }) => {
const service = new DeliverAlertService();
apps/webapp/app/v3/commonWorker.server.ts
+1 -1
View File
@@ -196,7 +196,7 @@ function initializeWorker() {
pollIntervalMs: env.COMMON_WORKER_POLL_INTERVAL,
immediatePollIntervalMs: env.COMMON_WORKER_IMMEDIATE_POLL_INTERVAL,
shutdownTimeoutMs: env.COMMON_WORKER_SHUTDOWN_TIMEOUT_MS,
logger: new Logger("CommonWorker", "debug"),
logger: new Logger("CommonWorker", env.COMMON_WORKER_LOG_LEVEL),
jobs: {
scheduleEmail: async ({ payload }) => {
await sendEmail(payload);
apps/webapp/app/v3/legacyRunEngineWorker.server.ts
+1 -1
View File
@@ -68,7 +68,7 @@ function initializeWorker() {
pollIntervalMs: env.LEGACY_RUN_ENGINE_WORKER_POLL_INTERVAL,
immediatePollIntervalMs: env.LEGACY_RUN_ENGINE_WORKER_IMMEDIATE_POLL_INTERVAL,
shutdownTimeoutMs: env.LEGACY_RUN_ENGINE_WORKER_SHUTDOWN_TIMEOUT_MS,
logger: new Logger("LegacyRunEngineWorker", "debug"),
logger: new Logger("LegacyRunEngineWorker", env.LEGACY_RUN_ENGINE_WORKER_LOG_LEVEL),
jobs: {
runHeartbeat: async ({ payload }) => {
const service = new TaskRunHeartbeatFailedService();
apps/webapp/app/v3/scheduleEngine.server.ts
+2
View File
@@ -61,6 +61,8 @@ function createScheduleEngine() {
},
worker: {
concurrency: env.SCHEDULE_WORKER_CONCURRENCY_LIMIT,
workers: env.SCHEDULE_WORKER_CONCURRENCY_WORKERS,
tasksPerWorker: env.SCHEDULE_WORKER_CONCURRENCY_TASKS_PER_WORKER,
pollIntervalMs: env.SCHEDULE_WORKER_POLL_INTERVAL,
shutdownTimeoutMs: env.SCHEDULE_WORKER_SHUTDOWN_TIMEOUT_MS,
disabled: env.SCHEDULE_WORKER_ENABLED === "0",
docs/tags.mdx
+1 -1
View File
@@ -5,7 +5,7 @@ description: "Tags allow you to easily filter runs in the dashboard and when usi
## What are tags?
We support up to 5 tags per run. Each one must be a string between 1 and 64 characters long.
We support up to 10 tags per run. Each one must be a string between 1 and 64 characters long.
We recommend prefixing your tags with their type and then an underscore or colon. For example, `user_123456` or `video:123`.
internal-packages/clickhouse/src/taskRuns.test.ts
+50
View File
@@ -344,4 +344,54 @@ describe("Task Runs V2", () => {
expect(result2).toEqual([]);
}
);
clickhouseTest(
"should be able to insert payloads with a duplicate path",
async ({ clickhouseContainer }) => {
const client = new ClickhouseClient({
name: "test",
url: clickhouseContainer.getConnectionUrl(),
});
const insertPayloads = insertRawTaskRunPayloads(client, {
async_insert: 0, // turn off async insert for this test
});
const [insertPayloadsError, insertPayloadsResult] = await insertPayloads([
{
run_id: "run_1234",
created_at: Date.now(),
payload: {
data: {
title: {
id: "123",
},
"title.id": 123,
},
},
},
]);
expect(insertPayloadsError).toBeNull();
expect(insertPayloadsResult).toEqual(expect.objectContaining({ executed: true }));
expect(insertPayloadsResult?.summary?.written_rows).toEqual("1");
const queryPayloads = client.query({
name: "query-raw-task-run-payloads",
query: "SELECT * FROM trigger_dev.raw_task_runs_payload_v1",
schema: z.object({
run_id: z.string(),
created_at: z.coerce.date(),
payload: z.unknown(),
}),
});
const [queryPayloadsError, resultPayloads] = await queryPayloads({ run_id: "run_1234" });
expect(queryPayloadsError).toBeNull();
expect(resultPayloads).toEqual(
expect.arrayContaining([expect.objectContaining({ run_id: "run_1234" })])
);
}
);
});
internal-packages/clickhouse/src/taskRuns.ts
+2
View File
@@ -59,6 +59,7 @@ export function insertTaskRuns(ch: ClickhouseWriter, settings?: ClickHouseSettin
async_insert_max_data_size: "1000000",
async_insert_busy_timeout_ms: 1000,
enable_json_type: 1,
type_json_skip_duplicated_paths: 1,
...settings,
},
});
@@ -83,6 +84,7 @@ export function insertRawTaskRunPayloads(ch: ClickhouseWriter, settings?: ClickH
async_insert_max_data_size: "1000000",
async_insert_busy_timeout_ms: 1000,
enable_json_type: 1,
type_json_skip_duplicated_paths: 1,
...settings,
},
});
internal-packages/run-engine/src/engine/index.ts
+1 -1
View File
@@ -152,7 +152,7 @@ export class RunEngine {
pollIntervalMs: options.worker.pollIntervalMs,
immediatePollIntervalMs: options.worker.immediatePollIntervalMs,
shutdownTimeoutMs: options.worker.shutdownTimeoutMs,
logger: new Logger("RunEngineWorker", "debug"),
logger: new Logger("RunEngineWorker", options.logLevel ?? "info"),
jobs: {
finishWaitpoint: async ({ payload }) => {
await this.waitpointSystem.completeWaitpoint({
internal-packages/schedule-engine/src/engine/distributedScheduling.ts
+16 -10
View File
@@ -5,21 +5,27 @@
*/
export function calculateDistributedExecutionTime(
exactScheduleTime: Date,
distributionWindowSeconds: number = 30
distributionWindowSeconds: number = 30,
instanceId?: string
): Date {
// Use the ISO string of the exact schedule time as the seed for consistency
const seed = exactScheduleTime.toISOString();
// Create seed by combining ISO timestamp with optional instanceId
// This ensures different instances get different distributions even with same schedule time
const timeSeed = exactScheduleTime.toISOString();
const seed = instanceId ? `${timeSeed}:${instanceId}` : timeSeed;
// Use a better hash function (FNV-1a variant) for more uniform distribution
let hash = 2166136261; // FNV offset basis (32-bit)
// Create a simple hash from the seed string
let hash = 0;
for (let i = 0; i < seed.length; i++) {
const char = seed.charCodeAt(i);
hash = (hash << 5) - hash + char;
hash = hash & hash; // Convert to 32-bit integer
hash ^= seed.charCodeAt(i);
hash *= 16777619; // FNV prime (32-bit)
// Keep it as 32-bit unsigned integer
hash = hash >>> 0;
}
// Convert hash to a value between 0 and 1
const normalized = Math.abs(hash) / Math.pow(2, 31);
// Convert hash to a value between 0 and 1 using better normalization
// Use the full 32-bit range for better distribution
const normalized = hash / 0xffffffff;
// Calculate offset in milliseconds (0 to distributionWindowSeconds * 1000)
const offsetMs = Math.floor(normalized * distributionWindowSeconds * 1000);
internal-packages/schedule-engine/src/engine/index.ts
+4 -1
View File
@@ -92,6 +92,8 @@ export class ScheduleEngine {
catalog: scheduleWorkerCatalog,
concurrency: {
limit: options.worker.concurrency,
workers: options.worker.workers,
tasksPerWorker: options.worker.tasksPerWorker,
},
pollIntervalMs: options.worker.pollIntervalMs,
shutdownTimeoutMs: options.worker.shutdownTimeoutMs,
@@ -590,7 +592,8 @@ export class ScheduleEngine {
const distributedExecutionTime = calculateDistributedExecutionTime(
exactScheduleTime,
this.distributionWindowSeconds
this.distributionWindowSeconds,
instanceId
);
const distributionOffsetMs = exactScheduleTime.getTime() - distributedExecutionTime.getTime();
internal-packages/schedule-engine/src/engine/types.ts
+2
View File
@@ -35,6 +35,8 @@ export interface ScheduleEngineOptions {
redis: RedisOptions;
worker: {
concurrency: number;
workers?: number;
tasksPerWorker?: number;
pollIntervalMs?: number;
shutdownTimeoutMs?: number;
disabled?: boolean;
packages/redis-worker/src/worker.ts
+48 -2
View File
@@ -206,7 +206,7 @@ class Worker<TCatalog extends WorkerCatalog> {
// Launch a number of "worker loops" on the main thread.
for (let i = 0; i < workers; i++) {
this.workerLoops.push(this.runWorkerLoop(`worker-${nanoid(12)}`, tasksPerWorker));
this.workerLoops.push(this.runWorkerLoop(`worker-${nanoid(12)}`, tasksPerWorker, i, workers));
}
this.setupShutdownHandlers();
@@ -390,14 +390,43 @@ class Worker<TCatalog extends WorkerCatalog> {
* The main loop that each worker runs. It repeatedly polls for items,
* processes them, and then waits before the next iteration.
*/
private async runWorkerLoop(workerId: string, taskCount: number): Promise<void> {
private async runWorkerLoop(
workerId: string,
taskCount: number,
workerIndex: number,
totalWorkers: number
): Promise<void> {
const pollIntervalMs = this.options.pollIntervalMs ?? 1000;
const immediatePollIntervalMs = this.options.immediatePollIntervalMs ?? 100;
// Calculate the delay between starting each worker loop so that they don't all start at the same time.
const delayBetweenWorkers = this.options.pollIntervalMs ?? 1000;
const delay = delayBetweenWorkers * (totalWorkers - workerIndex);
await Worker.delay(delay);
this.logger.info("Starting worker loop", {
workerIndex,
totalWorkers,
delay,
workerId,
taskCount,
pollIntervalMs,
immediatePollIntervalMs,
concurrencyOptions: this.concurrency,
});
while (!this.isShuttingDown) {
// Check overall load. If at capacity, wait a bit before trying to dequeue more.
if (this.limiter.activeCount + this.limiter.pendingCount >= this.concurrency.limit) {
this.logger.debug("Worker at capacity, waiting", {
workerId,
concurrencyOptions: this.concurrency,
activeCount: this.limiter.activeCount,
pendingCount: this.limiter.pendingCount,
});
await Worker.delay(pollIntervalMs);
continue;
}
@@ -412,10 +441,25 @@ class Worker<TCatalog extends WorkerCatalog> {
);
if (items.length === 0) {
this.logger.debug("No items to dequeue", {
workerId,
concurrencyOptions: this.concurrency,
activeCount: this.limiter.activeCount,
pendingCount: this.limiter.pendingCount,
});
await Worker.delay(pollIntervalMs);
continue;
}
this.logger.debug("Dequeued items", {
workerId,
itemCount: items.length,
concurrencyOptions: this.concurrency,
activeCount: this.limiter.activeCount,
pendingCount: this.limiter.pendingCount,
});
// Schedule each item using the limiter.
for (const item of items) {
this.limiter(() => this.processItem(item as AnyQueueItem, items.length, workerId)).catch(
@@ -433,6 +477,8 @@ class Worker<TCatalog extends WorkerCatalog> {
// Wait briefly before immediately polling again since we processed items
await Worker.delay(immediatePollIntervalMs);
}
this.logger.info("Worker loop finished", { workerId });
}
/**