Merge remote-tracking branch 'origin/main' into feat/arm64-registry
This commit is contained in:
nicktrn
@@ -430,6 +430,10 @@ const EnvironmentSchema = z.object({
|
||||
RUN_ENGINE_PROCESS_WORKER_QUEUE_DEBOUNCE_MS: z.coerce.number().int().default(200),
|
||||
RUN_ENGINE_DEQUEUE_BLOCKING_TIMEOUT_SECONDS: z.coerce.number().int().default(10),
|
||||
RUN_ENGINE_MASTER_QUEUE_CONSUMERS_INTERVAL_MS: z.coerce.number().int().default(500),
|
||||
RUN_ENGINE_CONCURRENCY_SWEEPER_SCAN_SCHEDULE: z.string().optional(),
|
||||
RUN_ENGINE_CONCURRENCY_SWEEPER_PROCESS_MARKED_SCHEDULE: z.string().optional(),
|
||||
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(),
|
||||
|
||||
RUN_ENGINE_RUN_LOCK_DURATION: z.coerce.number().int().default(5000),
|
||||
RUN_ENGINE_RUN_LOCK_AUTOMATIC_EXTENSION_THRESHOLD: z.coerce.number().int().default(1000),
|
||||
@@ -595,6 +599,7 @@ const EnvironmentSchema = z.object({
|
||||
|
||||
RUN_ENGINE_WORKER_ENABLED: z.string().default("1"),
|
||||
RUN_ENGINE_WORKER_LOG_LEVEL: z.enum(["log", "error", "warn", "info", "debug"]).default("info"),
|
||||
RUN_ENGINE_RUN_QUEUE_LOG_LEVEL: z.enum(["log", "error", "warn", "info", "debug"]).default("info"),
|
||||
|
||||
/** How long should the presence ttl last */
|
||||
DEV_PRESENCE_SSE_TIMEOUT: z.coerce.number().int().default(30_000),
|
||||
@@ -839,9 +844,13 @@ const EnvironmentSchema = z.object({
|
||||
RUN_REPLICATION_LEADER_LOCK_ADDITIONAL_TIME_MS: z.coerce.number().int().default(10_000),
|
||||
RUN_REPLICATION_LEADER_LOCK_RETRY_INTERVAL_MS: z.coerce.number().int().default(500),
|
||||
RUN_REPLICATION_WAIT_FOR_ASYNC_INSERT: z.string().default("0"),
|
||||
RUN_REPLICATION_KEEP_ALIVE_ENABLED: z.string().default("1"),
|
||||
RUN_REPLICATION_KEEP_ALIVE_ENABLED: z.string().default("0"),
|
||||
RUN_REPLICATION_KEEP_ALIVE_IDLE_SOCKET_TTL_MS: z.coerce.number().int().optional(),
|
||||
RUN_REPLICATION_MAX_OPEN_CONNECTIONS: z.coerce.number().int().default(10),
|
||||
// Retry configuration for insert operations
|
||||
RUN_REPLICATION_INSERT_MAX_RETRIES: z.coerce.number().int().default(3),
|
||||
RUN_REPLICATION_INSERT_BASE_DELAY_MS: z.coerce.number().int().default(100),
|
||||
RUN_REPLICATION_INSERT_MAX_DELAY_MS: z.coerce.number().int().default(2000),
|
||||
|
||||
// Clickhouse
|
||||
CLICKHOUSE_URL: z.string().optional(),
|
||||
|
||||
@@ -62,6 +62,9 @@ function initializeRunsReplicationInstance() {
|
||||
logLevel: env.RUN_REPLICATION_LOG_LEVEL,
|
||||
waitForAsyncInsert: env.RUN_REPLICATION_WAIT_FOR_ASYNC_INSERT === "1",
|
||||
tracer: provider.getTracer("runs-replication-service"),
|
||||
insertMaxRetries: env.RUN_REPLICATION_INSERT_MAX_RETRIES,
|
||||
insertBaseDelayMs: env.RUN_REPLICATION_INSERT_BASE_DELAY_MS,
|
||||
insertMaxDelayMs: env.RUN_REPLICATION_INSERT_MAX_DELAY_MS,
|
||||
});
|
||||
|
||||
if (env.RUN_REPLICATION_ENABLED === "1") {
|
||||
|
||||
@@ -15,6 +15,8 @@ import { TaskRun } from "@trigger.dev/database";
|
||||
import { nanoid } from "nanoid";
|
||||
import EventEmitter from "node:events";
|
||||
import pLimit from "p-limit";
|
||||
import { logger } from "./logger.server";
|
||||
import { detectBadJsonStrings } from "~/utils/detectBadJsonStrings";
|
||||
|
||||
interface TransactionEvent<T = any> {
|
||||
tag: "insert" | "update" | "delete";
|
||||
@@ -51,6 +53,10 @@ export type RunsReplicationServiceOptions = {
|
||||
logLevel?: LogLevel;
|
||||
tracer?: Tracer;
|
||||
waitForAsyncInsert?: boolean;
|
||||
// Retry configuration for insert operations
|
||||
insertMaxRetries?: number;
|
||||
insertBaseDelayMs?: number;
|
||||
insertMaxDelayMs?: number;
|
||||
};
|
||||
|
||||
type TaskRunInsert = { _version: bigint; run: TaskRun; event: "insert" | "update" | "delete" };
|
||||
@@ -80,6 +86,10 @@ export class RunsReplicationService {
|
||||
private _latestCommitEndLsn: string | null = null;
|
||||
private _lastAcknowledgedLsn: string | null = null;
|
||||
private _acknowledgeInterval: NodeJS.Timeout | null = null;
|
||||
// Retry configuration
|
||||
private _insertMaxRetries: number;
|
||||
private _insertBaseDelayMs: number;
|
||||
private _insertMaxDelayMs: number;
|
||||
|
||||
public readonly events: EventEmitter<RunsReplicationServiceEvents>;
|
||||
|
||||
@@ -151,6 +161,11 @@ export class RunsReplicationService {
|
||||
this._replicationClient.events.on("leaderElection", (isLeader) => {
|
||||
this.logger.info("Leader election", { isLeader });
|
||||
});
|
||||
|
||||
// Initialize retry configuration
|
||||
this._insertMaxRetries = options.insertMaxRetries ?? 3;
|
||||
this._insertBaseDelayMs = options.insertBaseDelayMs ?? 100;
|
||||
this._insertMaxDelayMs = options.insertMaxDelayMs ?? 2000;
|
||||
}
|
||||
|
||||
public async shutdown() {
|
||||
@@ -445,8 +460,37 @@ export class RunsReplicationService {
|
||||
payloadInserts: payloadInserts.length,
|
||||
});
|
||||
|
||||
await this.#insertTaskRunInserts(taskRunInserts);
|
||||
await this.#insertPayloadInserts(payloadInserts);
|
||||
// Insert task runs and payloads with retry logic for connection errors
|
||||
const [taskRunError, taskRunResult] = await this.#insertWithRetry(
|
||||
() => this.#insertTaskRunInserts(taskRunInserts),
|
||||
"task run inserts",
|
||||
flushId
|
||||
);
|
||||
|
||||
const [payloadError, payloadResult] = await this.#insertWithRetry(
|
||||
() => this.#insertPayloadInserts(payloadInserts),
|
||||
"payload inserts",
|
||||
flushId
|
||||
);
|
||||
|
||||
// Log any errors that occurred
|
||||
if (taskRunError) {
|
||||
this.logger.error("Error inserting task run inserts", {
|
||||
error: taskRunError,
|
||||
flushId,
|
||||
runIds: taskRunInserts.map((r) => r.run_id),
|
||||
});
|
||||
recordSpanError(span, taskRunError);
|
||||
}
|
||||
|
||||
if (payloadError) {
|
||||
this.logger.error("Error inserting payload inserts", {
|
||||
error: payloadError,
|
||||
flushId,
|
||||
runIds: payloadInserts.map((r) => r.run_id),
|
||||
});
|
||||
recordSpanError(span, payloadError);
|
||||
}
|
||||
|
||||
this.logger.debug("Flushed inserts", {
|
||||
flushId,
|
||||
@@ -456,6 +500,73 @@ export class RunsReplicationService {
|
||||
});
|
||||
}
|
||||
|
||||
// New method to handle inserts with retry logic for connection errors
|
||||
async #insertWithRetry<T>(
|
||||
insertFn: () => Promise<T>,
|
||||
operationName: string,
|
||||
flushId: string
|
||||
): Promise<[Error | null, T | null]> {
|
||||
let lastError: Error | null = null;
|
||||
|
||||
for (let attempt = 1; attempt <= this._insertMaxRetries; attempt++) {
|
||||
try {
|
||||
const result = await insertFn();
|
||||
return [null, result];
|
||||
} catch (error) {
|
||||
lastError = error instanceof Error ? error : new Error(String(error));
|
||||
|
||||
// Check if this is a retryable connection error
|
||||
if (this.#isRetryableConnectionError(lastError) && attempt < this._insertMaxRetries) {
|
||||
const delay = this.#calculateConnectionRetryDelay(attempt);
|
||||
|
||||
this.logger.warn(`Retrying ${operationName} due to connection error`, {
|
||||
flushId,
|
||||
attempt,
|
||||
maxRetries: this._insertMaxRetries,
|
||||
error: lastError.message,
|
||||
delay,
|
||||
});
|
||||
|
||||
await new Promise((resolve) => setTimeout(resolve, delay));
|
||||
continue;
|
||||
}
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
return [lastError, null];
|
||||
}
|
||||
|
||||
// New method to check if an error is a retryable connection error
|
||||
#isRetryableConnectionError(error: Error): boolean {
|
||||
const errorMessage = error.message.toLowerCase();
|
||||
const retryableConnectionPatterns = [
|
||||
"socket hang up",
|
||||
"econnreset",
|
||||
"connection reset",
|
||||
"connection refused",
|
||||
"connection timeout",
|
||||
"network error",
|
||||
"read econnreset",
|
||||
"write econnreset",
|
||||
];
|
||||
|
||||
return retryableConnectionPatterns.some((pattern) => errorMessage.includes(pattern));
|
||||
}
|
||||
|
||||
// New method to calculate retry delay for connection errors
|
||||
#calculateConnectionRetryDelay(attempt: number): number {
|
||||
// Exponential backoff: baseDelay, baseDelay*2, baseDelay*4, etc.
|
||||
const delay = Math.min(
|
||||
this._insertBaseDelayMs * Math.pow(2, attempt - 1),
|
||||
this._insertMaxDelayMs
|
||||
);
|
||||
|
||||
// Add some jitter to prevent thundering herd
|
||||
const jitter = Math.random() * 100;
|
||||
return delay + jitter;
|
||||
}
|
||||
|
||||
async #insertTaskRunInserts(taskRunInserts: TaskRunV2[]) {
|
||||
return await startSpan(this._tracer, "insertTaskRunsInserts", async (span) => {
|
||||
const [insertError, insertResult] = await this.options.clickhouse.taskRuns.insert(
|
||||
@@ -604,6 +715,7 @@ export class RunsReplicationService {
|
||||
idempotency_key: run.idempotencyKey ?? "",
|
||||
expiration_ttl: run.ttl ?? "",
|
||||
output,
|
||||
concurrency_key: run.concurrencyKey ?? "",
|
||||
_version: _version.toString(),
|
||||
_is_deleted: event === "delete" ? 1 : 0,
|
||||
};
|
||||
@@ -631,6 +743,14 @@ export class RunsReplicationService {
|
||||
return { data: undefined };
|
||||
}
|
||||
|
||||
if (detectBadJsonStrings(data)) {
|
||||
this.logger.warn("Detected bad JSON strings", {
|
||||
data,
|
||||
dataType,
|
||||
});
|
||||
return { data: undefined };
|
||||
}
|
||||
|
||||
const packet = {
|
||||
data,
|
||||
dataType,
|
||||
|
||||
@@ -0,0 +1,68 @@
|
||||
export function detectBadJsonStrings(jsonString: string): boolean {
|
||||
// Fast path: skip everything if no \u
|
||||
let idx = jsonString.indexOf("\\u");
|
||||
if (idx === -1) return false;
|
||||
|
||||
// Use a more efficient scanning strategy
|
||||
const length = jsonString.length;
|
||||
|
||||
while (idx !== -1 && idx < length - 5) {
|
||||
// Only check if we have enough characters left
|
||||
if (idx + 6 > length) break;
|
||||
|
||||
if (jsonString[idx + 1] === "u" && jsonString[idx + 2] === "d") {
|
||||
const third = jsonString[idx + 3];
|
||||
|
||||
// High surrogate check
|
||||
if (
|
||||
/[89ab]/.test(third) &&
|
||||
/[0-9a-f]/.test(jsonString[idx + 4]) &&
|
||||
/[0-9a-f]/.test(jsonString[idx + 5])
|
||||
) {
|
||||
// Check for low surrogate after (need at least 6 more chars)
|
||||
if (idx + 12 > length) {
|
||||
return true; // Incomplete high surrogate (not enough chars left)
|
||||
}
|
||||
|
||||
if (
|
||||
jsonString[idx + 6] !== "\\" ||
|
||||
jsonString[idx + 7] !== "u" ||
|
||||
jsonString[idx + 8] !== "d" ||
|
||||
!/[cd]/.test(jsonString[idx + 9]) ||
|
||||
!/[0-9a-f]/.test(jsonString[idx + 10]) ||
|
||||
!/[0-9a-f]/.test(jsonString[idx + 11])
|
||||
) {
|
||||
return true; // Incomplete high surrogate
|
||||
}
|
||||
}
|
||||
|
||||
// Low surrogate check
|
||||
if (
|
||||
(third === "c" || third === "d") &&
|
||||
/[0-9a-f]/.test(jsonString[idx + 4]) &&
|
||||
/[0-9a-f]/.test(jsonString[idx + 5])
|
||||
) {
|
||||
// Check for high surrogate before (need at least 6 chars before)
|
||||
if (idx < 6) {
|
||||
return true; // Incomplete low surrogate (not enough chars before)
|
||||
}
|
||||
|
||||
if (
|
||||
jsonString[idx - 6] !== "\\" ||
|
||||
jsonString[idx - 5] !== "u" ||
|
||||
jsonString[idx - 4] !== "d" ||
|
||||
!/[89ab]/.test(jsonString[idx - 3]) ||
|
||||
!/[0-9a-f]/.test(jsonString[idx - 2]) ||
|
||||
!/[0-9a-f]/.test(jsonString[idx - 1])
|
||||
) {
|
||||
return true; // Incomplete low surrogate
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// More efficient next search - skip ahead by 2 to avoid overlapping matches
|
||||
idx = jsonString.indexOf("\\u", idx + 2);
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
@@ -1,10 +1,10 @@
|
||||
import { RunEngine } from "@internal/run-engine";
|
||||
import { defaultMachine } from "~/services/platform.v3.server";
|
||||
import { prisma } from "~/db.server";
|
||||
import { $replica, prisma } from "~/db.server";
|
||||
import { env } from "~/env.server";
|
||||
import { defaultMachine } from "~/services/platform.v3.server";
|
||||
import { singleton } from "~/utils/singleton";
|
||||
import { allMachines } from "./machinePresets.server";
|
||||
import { tracer, meter } from "./tracer.server";
|
||||
import { meter, tracer } from "./tracer.server";
|
||||
|
||||
export const engine = singleton("RunEngine", createRunEngine);
|
||||
|
||||
@@ -13,6 +13,7 @@ export type { RunEngine };
|
||||
function createRunEngine() {
|
||||
const engine = new RunEngine({
|
||||
prisma,
|
||||
readOnlyPrisma: $replica,
|
||||
logLevel: env.RUN_ENGINE_WORKER_LOG_LEVEL,
|
||||
worker: {
|
||||
disabled: env.RUN_ENGINE_WORKER_ENABLED === "0",
|
||||
@@ -39,6 +40,7 @@ function createRunEngine() {
|
||||
},
|
||||
queue: {
|
||||
defaultEnvConcurrency: env.DEFAULT_ENV_EXECUTION_CONCURRENCY_LIMIT,
|
||||
logLevel: env.RUN_ENGINE_RUN_QUEUE_LOG_LEVEL,
|
||||
redis: {
|
||||
keyPrefix: "engine:",
|
||||
port: env.RUN_ENGINE_RUN_QUEUE_REDIS_PORT ?? undefined,
|
||||
@@ -64,6 +66,12 @@ function createRunEngine() {
|
||||
dequeueBlockingTimeoutSeconds: env.RUN_ENGINE_DEQUEUE_BLOCKING_TIMEOUT_SECONDS,
|
||||
masterQueueConsumersIntervalMs: env.RUN_ENGINE_MASTER_QUEUE_CONSUMERS_INTERVAL_MS,
|
||||
masterQueueConsumersDisabled: env.RUN_ENGINE_WORKER_ENABLED === "0",
|
||||
concurrencySweeper: {
|
||||
scanSchedule: env.RUN_ENGINE_CONCURRENCY_SWEEPER_SCAN_SCHEDULE,
|
||||
processMarkedSchedule: env.RUN_ENGINE_CONCURRENCY_SWEEPER_PROCESS_MARKED_SCHEDULE,
|
||||
scanJitterInMs: env.RUN_ENGINE_CONCURRENCY_SWEEPER_SCAN_JITTER_IN_MS,
|
||||
processMarkedJitterInMs: env.RUN_ENGINE_CONCURRENCY_SWEEPER_PROCESS_MARKED_JITTER_IN_MS,
|
||||
},
|
||||
},
|
||||
runLock: {
|
||||
redis: {
|
||||
|
||||
@@ -0,0 +1,3 @@
|
||||
{
|
||||
"title": "❜ 𝐒 𝐏𝗈𝗌𝗍 . . . 𝐍𝖾𝗐 𝐂𝗈𝗇𝗍𝖾𝗇𝗍 ꒰ ⚔️ ꒱ 𝐒𝐋 ❜ 𝐔𝐋\n\n꒰ ❤️ ꒱ 𓃊 𝐋𝗲𝗮𝘃𝗲 𝖺 𝗹𝗶𝗸𝗲 𝖺𝗇\ud835"
|
||||
}
|
||||
@@ -0,0 +1,190 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { detectBadJsonStrings } from "~/utils/detectBadJsonStrings";
|
||||
|
||||
describe("detectBadJsonStrings", () => {
|
||||
it("should not detect valid JSON string", () => {
|
||||
const goodJson = `{"title": "hello"}`;
|
||||
const result = detectBadJsonStrings(goodJson);
|
||||
expect(result).toBe(false);
|
||||
});
|
||||
|
||||
it("should detect incomplete Unicode escape sequences", () => {
|
||||
const badJson = `{"title": "hello\\ud835"}`;
|
||||
const result = detectBadJsonStrings(badJson);
|
||||
expect(result).toBe(true);
|
||||
});
|
||||
|
||||
it("should not detect complete Unicode escape sequences", () => {
|
||||
const goodJson = `{"title": "hello\\ud835\\udc00"}`;
|
||||
const result = detectBadJsonStrings(goodJson);
|
||||
expect(result).toBe(false);
|
||||
});
|
||||
|
||||
it("should detect incomplete low surrogate", () => {
|
||||
const badJson = `{"title": "hello\\udc00"}`;
|
||||
const result = detectBadJsonStrings(badJson);
|
||||
expect(result).toBe(true);
|
||||
});
|
||||
|
||||
it("should handle multiple Unicode sequences correctly", () => {
|
||||
const goodJson = `{"title": "hello\\ud835\\udc00\\ud835\\udc01"}`;
|
||||
const result = detectBadJsonStrings(goodJson);
|
||||
expect(result).toBe(false);
|
||||
});
|
||||
|
||||
it("should detect mixed complete and incomplete sequences", () => {
|
||||
const badJson = `{"title": "hello\\ud835\\udc00\\ud835"}`;
|
||||
const result = detectBadJsonStrings(badJson);
|
||||
expect(result).toBe(true);
|
||||
});
|
||||
|
||||
it("should have acceptable performance overhead", () => {
|
||||
const longText = `hello world `.repeat(1_000);
|
||||
const goodJson = `{"title": "hello", "text": "${longText}"}`;
|
||||
const badJson = `{"title": "hello\\ud835", "text": "${longText}"}`;
|
||||
|
||||
const iterations = 100_000;
|
||||
|
||||
// Warm up
|
||||
for (let i = 0; i < 1000; i++) {
|
||||
detectBadJsonStrings(goodJson);
|
||||
detectBadJsonStrings(badJson);
|
||||
}
|
||||
|
||||
// Measure good JSON (most common case)
|
||||
const goodStart = performance.now();
|
||||
for (let i = 0; i < iterations; i++) {
|
||||
detectBadJsonStrings(goodJson);
|
||||
}
|
||||
const goodTime = performance.now() - goodStart;
|
||||
|
||||
// Measure bad JSON (edge case)
|
||||
const badStart = performance.now();
|
||||
for (let i = 0; i < iterations; i++) {
|
||||
detectBadJsonStrings(badJson);
|
||||
}
|
||||
const badTime = performance.now() - badStart;
|
||||
|
||||
// Measure baseline (just function call overhead)
|
||||
const baselineStart = performance.now();
|
||||
for (let i = 0; i < iterations; i++) {
|
||||
// Empty function call to measure baseline
|
||||
}
|
||||
const baselineTime = performance.now() - baselineStart;
|
||||
|
||||
const goodOverhead = goodTime - baselineTime;
|
||||
const badOverhead = badTime - baselineTime;
|
||||
|
||||
console.log(`Baseline (${iterations} iterations): ${baselineTime.toFixed(2)}ms`);
|
||||
console.log(
|
||||
`Good JSON (${iterations} iterations): ${goodTime.toFixed(
|
||||
2
|
||||
)}ms (overhead: ${goodOverhead.toFixed(2)}ms)`
|
||||
);
|
||||
console.log(
|
||||
`Bad JSON (${iterations} iterations): ${badTime.toFixed(
|
||||
2
|
||||
)}ms (overhead: ${badOverhead.toFixed(2)}ms)`
|
||||
);
|
||||
console.log(
|
||||
`Average per call - Good: ${(goodOverhead / iterations).toFixed(4)}ms, Bad: ${(
|
||||
badOverhead / iterations
|
||||
).toFixed(4)}ms`
|
||||
);
|
||||
|
||||
// Assertions for performance expectations
|
||||
// Good JSON should be reasonably fast (most common case)
|
||||
expect(goodOverhead / iterations).toBeLessThan(0.01); // Less than 10 microseconds per call
|
||||
|
||||
// Bad JSON can be slower due to regex matching, but still reasonable
|
||||
expect(badOverhead / iterations).toBeLessThan(0.01); // Less than 20 microseconds per call
|
||||
|
||||
// Total overhead for 100k calls should be reasonable
|
||||
expect(goodOverhead).toBeLessThan(1000); // Less than 1 second for 100k calls
|
||||
});
|
||||
|
||||
it("should handle various JSON sizes efficiently", () => {
|
||||
const sizes = [100, 1000, 10000, 100000];
|
||||
const iterations = 10_000;
|
||||
|
||||
for (const size of sizes) {
|
||||
const text = `hello world `.repeat(size / 11); // Approximate size
|
||||
const goodJson = `{"title": "hello", "text": "${text}"}`;
|
||||
|
||||
const start = performance.now();
|
||||
for (let i = 0; i < iterations; i++) {
|
||||
detectBadJsonStrings(goodJson);
|
||||
}
|
||||
const time = performance.now() - start;
|
||||
|
||||
console.log(
|
||||
`Size ${size} chars (${iterations} iterations): ${time.toFixed(2)}ms (${(
|
||||
time / iterations
|
||||
).toFixed(4)}ms per call)`
|
||||
);
|
||||
|
||||
// Performance should scale reasonably with size
|
||||
expect(time / iterations).toBeLessThan(size / 1000); // Roughly linear scaling
|
||||
}
|
||||
});
|
||||
|
||||
it("should show significant performance improvement with quick rejection", () => {
|
||||
const longText = `hello world `.repeat(1_000);
|
||||
const goodJson = `{"title": "hello", "text": "${longText}"}`;
|
||||
const badJson = `{"title": "hello\\ud835", "text": "${longText}"}`;
|
||||
const noUnicodeJson = `{"title": "hello", "text": "${longText}"}`;
|
||||
|
||||
const iterations = 100_000;
|
||||
|
||||
// Warm up
|
||||
for (let i = 0; i < 1000; i++) {
|
||||
detectBadJsonStrings(goodJson);
|
||||
detectBadJsonStrings(badJson);
|
||||
detectBadJsonStrings(noUnicodeJson);
|
||||
}
|
||||
|
||||
// Test strings with no Unicode escapes (99.9% case)
|
||||
const noUnicodeStart = performance.now();
|
||||
for (let i = 0; i < iterations; i++) {
|
||||
detectBadJsonStrings(noUnicodeJson);
|
||||
}
|
||||
const noUnicodeTime = performance.now() - noUnicodeStart;
|
||||
|
||||
// Test strings with Unicode escapes (0.1% case)
|
||||
const withUnicodeStart = performance.now();
|
||||
for (let i = 0; i < iterations; i++) {
|
||||
detectBadJsonStrings(badJson);
|
||||
}
|
||||
const withUnicodeTime = performance.now() - withUnicodeStart;
|
||||
|
||||
console.log(
|
||||
`No Unicode escapes (${iterations} iterations): ${noUnicodeTime.toFixed(2)}ms (${(
|
||||
noUnicodeTime / iterations
|
||||
).toFixed(4)}ms per call)`
|
||||
);
|
||||
console.log(
|
||||
`With Unicode escapes (${iterations} iterations): ${withUnicodeTime.toFixed(2)}ms (${(
|
||||
withUnicodeTime / iterations
|
||||
).toFixed(4)}ms per call)`
|
||||
);
|
||||
console.log(
|
||||
`Performance ratio: ${(withUnicodeTime / noUnicodeTime).toFixed(
|
||||
2
|
||||
)}x slower for Unicode strings`
|
||||
);
|
||||
|
||||
// Both cases should be extremely fast (under 1 microsecond per call)
|
||||
expect(noUnicodeTime / iterations).toBeLessThan(0.001); // Less than 1 microsecond
|
||||
expect(withUnicodeTime / iterations).toBeLessThan(0.001); // Less than 1 microsecond
|
||||
|
||||
// The difference should be reasonable (not more than 5x)
|
||||
expect(noUnicodeTime / withUnicodeTime).toBeLessThan(5);
|
||||
});
|
||||
});
|
||||
|
||||
function processPacket(data: string): { data?: string; dataType?: string } {
|
||||
if (detectBadJsonStrings(data)) {
|
||||
return { data: undefined };
|
||||
}
|
||||
return { data, dataType: "application/json" };
|
||||
}
|
||||
@@ -3,6 +3,7 @@ import { describe, expect, vi } from "vitest";
|
||||
// Mock the db prisma client
|
||||
vi.mock("~/db.server", () => ({
|
||||
prisma: {},
|
||||
$replica: {},
|
||||
}));
|
||||
|
||||
vi.mock("~/services/platform.v3.server", async (importOriginal) => {
|
||||
|
||||
@@ -1,12 +1,11 @@
|
||||
import { ClickHouse } from "@internal/clickhouse";
|
||||
import { containerTest } from "@internal/testcontainers";
|
||||
import { Logger } from "@trigger.dev/core/logger";
|
||||
import { readFile } from "node:fs/promises";
|
||||
import { setTimeout } from "node:timers/promises";
|
||||
import { z } from "zod";
|
||||
import { TaskRunStatus } from "~/database-types";
|
||||
import { RunsReplicationService } from "~/services/runsReplicationService.server";
|
||||
import { createInMemoryTracing } from "./utils/tracing";
|
||||
import superjson from "superjson";
|
||||
import { detectBadJsonStrings } from "~/utils/detectBadJsonStrings";
|
||||
|
||||
vi.setConfig({ testTimeout: 60_000 });
|
||||
|
||||
@@ -611,4 +610,177 @@ describe("RunsReplicationService (part 2/2)", () => {
|
||||
},
|
||||
{ timeout: 60_000 * 5 }
|
||||
);
|
||||
|
||||
containerTest(
|
||||
"should insert TaskRuns even if there are incomplete Unicode escape sequences in the JSON",
|
||||
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-stress-bulk-insert",
|
||||
});
|
||||
|
||||
const runsReplicationService = new RunsReplicationService({
|
||||
clickhouse,
|
||||
pgConnectionUrl: postgresContainer.getConnectionUri(),
|
||||
serviceName: "runs-replication-stress-bulk-insert",
|
||||
slotName: "task_runs_to_clickhouse_v1",
|
||||
publicationName: "task_runs_to_clickhouse_v1_publication",
|
||||
redisOptions,
|
||||
maxFlushConcurrency: 10,
|
||||
flushIntervalMs: 100,
|
||||
flushBatchSize: 50,
|
||||
leaderLockTimeoutMs: 5000,
|
||||
leaderLockExtendIntervalMs: 1000,
|
||||
ackIntervalSeconds: 5,
|
||||
logger: new Logger("runs-replication-stress-bulk-insert", "info"),
|
||||
});
|
||||
|
||||
await runsReplicationService.start();
|
||||
|
||||
const organization = await prisma.organization.create({
|
||||
data: {
|
||||
title: "test-stress-bulk-insert",
|
||||
slug: "test-stress-bulk-insert",
|
||||
},
|
||||
});
|
||||
|
||||
const project = await prisma.project.create({
|
||||
data: {
|
||||
name: "test-stress-bulk-insert",
|
||||
slug: "test-stress-bulk-insert",
|
||||
organizationId: organization.id,
|
||||
externalRef: "test-stress-bulk-insert",
|
||||
},
|
||||
});
|
||||
|
||||
const runtimeEnvironment = await prisma.runtimeEnvironment.create({
|
||||
data: {
|
||||
slug: "test-stress-bulk-insert",
|
||||
type: "DEVELOPMENT",
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
apiKey: "test-stress-bulk-insert",
|
||||
pkApiKey: "test-stress-bulk-insert",
|
||||
shortcode: "test-stress-bulk-insert",
|
||||
},
|
||||
});
|
||||
|
||||
// Prepare 9 unique TaskRuns
|
||||
const now = Date.now();
|
||||
const runsData = Array.from({ length: 9 }, (_, i) => ({
|
||||
friendlyId: `run_bulk_${now}_${i}`,
|
||||
taskIdentifier: `my-task-bulk`,
|
||||
payload: `{"title": "hello"}`,
|
||||
payloadType: "application/json",
|
||||
traceId: `bulk-${i}`,
|
||||
spanId: `bulk-${i}`,
|
||||
queue: "test-stress-bulk-insert",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT" as const,
|
||||
engine: "V2" as const,
|
||||
status: "PENDING" as const,
|
||||
attemptNumber: 1,
|
||||
createdAt: new Date(now + i),
|
||||
updatedAt: new Date(now + i),
|
||||
}));
|
||||
|
||||
//add a run with incomplete Unicode escape sequences
|
||||
const badPayload = await readFile(`${__dirname}/bad-clickhouse-output.json`, "utf-8");
|
||||
const hasProblems = detectBadJsonStrings(badPayload);
|
||||
expect(hasProblems).toBe(true);
|
||||
|
||||
runsData.push({
|
||||
friendlyId: `run_bulk_${now}_10`,
|
||||
taskIdentifier: `my-task-bulk`,
|
||||
payload: badPayload,
|
||||
payloadType: "application/json",
|
||||
traceId: `bulk-10`,
|
||||
spanId: `bulk-10`,
|
||||
queue: "test-stress-bulk-insert",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT" as const,
|
||||
engine: "V2" as const,
|
||||
status: "PENDING" as const,
|
||||
attemptNumber: 1,
|
||||
createdAt: new Date(now + 10),
|
||||
updatedAt: new Date(now + 10),
|
||||
});
|
||||
|
||||
// Bulk insert
|
||||
const created = await prisma.taskRun.createMany({ data: runsData });
|
||||
expect(created.count).toBe(10);
|
||||
|
||||
// Update the runs (not the 10th one)
|
||||
await prisma.taskRun.updateMany({
|
||||
where: {
|
||||
spanId: { not: "bulk-10" },
|
||||
},
|
||||
data: {
|
||||
status: "COMPLETED_SUCCESSFULLY",
|
||||
output: `{"foo":"bar"}`,
|
||||
outputType: "application/json",
|
||||
},
|
||||
});
|
||||
|
||||
// Give the 10th one a bad payload
|
||||
await prisma.taskRun.updateMany({
|
||||
where: {
|
||||
spanId: "bulk-10",
|
||||
},
|
||||
data: {
|
||||
status: "COMPLETED_SUCCESSFULLY",
|
||||
output: badPayload,
|
||||
outputType: "application/json",
|
||||
},
|
||||
});
|
||||
|
||||
// Wait for replication
|
||||
await setTimeout(5000);
|
||||
|
||||
// Query ClickHouse for all runs using FINAL
|
||||
const queryRuns = clickhouse.reader.query({
|
||||
name: "runs-replication-stress-bulk-insert",
|
||||
query: `SELECT * FROM trigger_dev.task_runs_v2 FINAL`,
|
||||
schema: z.any(),
|
||||
});
|
||||
|
||||
const [queryError, result] = await queryRuns({});
|
||||
expect(queryError).toBeNull();
|
||||
expect(result?.length).toBe(10);
|
||||
|
||||
console.log("Data", {
|
||||
runsData,
|
||||
result,
|
||||
});
|
||||
|
||||
// Check a few random runs for correctness
|
||||
for (let i = 0; i < 9; i++) {
|
||||
const expected = runsData[i];
|
||||
const found = result?.find((r: any) => r.friendly_id === expected.friendlyId);
|
||||
expect(found).toBeDefined();
|
||||
expect(found).toEqual(
|
||||
expect.objectContaining({
|
||||
friendly_id: expected.friendlyId,
|
||||
trace_id: expected.traceId,
|
||||
task_identifier: expected.taskIdentifier,
|
||||
status: "COMPLETED_SUCCESSFULLY",
|
||||
})
|
||||
);
|
||||
expect(found?.output).toBeDefined();
|
||||
}
|
||||
|
||||
// Check the run with the bad JSON
|
||||
const foundBad = result?.find((r: any) => r.span_id === "bulk-10");
|
||||
expect(foundBad).toBeDefined();
|
||||
expect(foundBad?.output).toStrictEqual({});
|
||||
|
||||
await runsReplicationService.stop();
|
||||
}
|
||||
);
|
||||
});
|
||||
|
||||
@@ -111,7 +111,7 @@ services:
|
||||
ports:
|
||||
- "5521:5521"
|
||||
environment:
|
||||
VITE_CLICKHOUSE_URL: "http://clickhouse:8123"
|
||||
VITE_CLICKHOUSE_URL: "http://localhost:8123"
|
||||
VITE_CLICKHOUSE_USER: "default"
|
||||
VITE_CLICKHOUSE_PASS: "password"
|
||||
networks:
|
||||
|
||||
@@ -25,5 +25,15 @@
|
||||
"db:migrate:down": "GOOSE_COMMAND=down pnpm run db:migrate",
|
||||
"test": "vitest --sequence.concurrent=false --no-file-parallelism",
|
||||
"test:coverage": "vitest --sequence.concurrent=false --no-file-parallelism --coverage.enabled"
|
||||
},
|
||||
"exports": {
|
||||
"./package.json": "./package.json",
|
||||
".": {
|
||||
"import": {
|
||||
"@triggerdotdev/source": "./src/index.ts",
|
||||
"types": "./dist/src/index.d.ts",
|
||||
"default": "./dist/src/index.js"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,12 @@
|
||||
-- +goose Up
|
||||
/*
|
||||
Add concurrency_key and bulk_action_group_ids columns with defaults.
|
||||
*/
|
||||
ALTER TABLE trigger_dev.task_runs_v2
|
||||
ADD COLUMN concurrency_key String DEFAULT '',
|
||||
ADD COLUMN bulk_action_group_ids Array(String) DEFAULT [];
|
||||
|
||||
-- +goose Down
|
||||
ALTER TABLE trigger_dev.task_runs_v2
|
||||
DROP COLUMN concurrency_key,
|
||||
DROP COLUMN bulk_action_group_ids;
|
||||
@@ -61,6 +61,8 @@ describe("Task Runs V2", () => {
|
||||
root_run_id: "root_run_1234",
|
||||
parent_run_id: "parent_run_1234",
|
||||
depth: 1,
|
||||
concurrency_key: "concurrency_key_1234",
|
||||
bulk_action_group_ids: ["bulk_action_group_id_1234", "bulk_action_group_id_1235"],
|
||||
_version: "1",
|
||||
},
|
||||
]);
|
||||
@@ -75,6 +77,8 @@ describe("Task Runs V2", () => {
|
||||
schema: z.object({
|
||||
environment_id: z.string(),
|
||||
run_id: z.string(),
|
||||
concurrency_key: z.string(),
|
||||
bulk_action_group_ids: z.array(z.string()),
|
||||
}),
|
||||
params: z.object({
|
||||
run_id: z.string(),
|
||||
@@ -89,6 +93,8 @@ describe("Task Runs V2", () => {
|
||||
expect.objectContaining({
|
||||
environment_id: "env_1234",
|
||||
run_id: "run_1234",
|
||||
concurrency_key: "concurrency_key_1234",
|
||||
bulk_action_group_ids: ["bulk_action_group_id_1234", "bulk_action_group_id_1235"],
|
||||
}),
|
||||
])
|
||||
);
|
||||
|
||||
@@ -42,6 +42,8 @@ export const TaskRunV2 = z.object({
|
||||
idempotency_key: z.string(),
|
||||
expiration_ttl: z.string(),
|
||||
is_test: z.boolean().default(false),
|
||||
concurrency_key: z.string().default(""),
|
||||
bulk_action_group_ids: z.array(z.string()).default([]),
|
||||
_version: z.string(),
|
||||
_is_deleted: z.number().int().default(0),
|
||||
});
|
||||
|
||||
@@ -16,6 +16,7 @@ import {
|
||||
Prisma,
|
||||
PrismaClient,
|
||||
PrismaClientOrTransaction,
|
||||
PrismaReplicaClient,
|
||||
TaskRun,
|
||||
TaskRunExecutionSnapshot,
|
||||
Waitpoint,
|
||||
@@ -50,6 +51,7 @@ import { TtlSystem } from "./systems/ttlSystem.js";
|
||||
import { WaitpointSystem } from "./systems/waitpointSystem.js";
|
||||
import { EngineWorker, HeartbeatTimeouts, RunEngineOptions, TriggerParams } from "./types.js";
|
||||
import { workerCatalog } from "./workerCatalog.js";
|
||||
import { getFinalRunStatuses, isFinalRunStatus } from "./statuses.js";
|
||||
|
||||
export class RunEngine {
|
||||
private runLockRedis: Redis;
|
||||
@@ -61,6 +63,7 @@ export class RunEngine {
|
||||
private heartbeatTimeouts: HeartbeatTimeouts;
|
||||
|
||||
prisma: PrismaClient;
|
||||
readOnlyPrisma: PrismaReplicaClient;
|
||||
runQueue: RunQueue;
|
||||
eventBus: EventBus = new EventEmitter<EventBusEvents>();
|
||||
executionSnapshotSystem: ExecutionSnapshotSystem;
|
||||
@@ -79,6 +82,7 @@ export class RunEngine {
|
||||
constructor(private readonly options: RunEngineOptions) {
|
||||
this.logger = options.logger ?? new Logger("RunEngine", this.options.logLevel ?? "info");
|
||||
this.prisma = options.prisma;
|
||||
this.readOnlyPrisma = options.readOnlyPrisma ?? this.prisma;
|
||||
this.runLockRedis = createRedisClient(
|
||||
{
|
||||
...options.runLock.redis,
|
||||
@@ -123,7 +127,7 @@ export class RunEngine {
|
||||
defaultEnvConcurrencyLimit: options.queue?.defaultEnvConcurrency ?? 10,
|
||||
}),
|
||||
defaultEnvConcurrency: options.queue?.defaultEnvConcurrency ?? 10,
|
||||
logger: new Logger("RunQueue", this.options.logLevel ?? "info"),
|
||||
logger: new Logger("RunQueue", options.queue?.logLevel ?? "info"),
|
||||
redis: { ...options.queue.redis, keyPrefix: `${options.queue.redis.keyPrefix}runqueue:` },
|
||||
retryOptions: options.queue?.retryOptions,
|
||||
workerOptions: {
|
||||
@@ -133,6 +137,13 @@ export class RunEngine {
|
||||
immediatePollIntervalMs: options.worker.immediatePollIntervalMs,
|
||||
shutdownTimeoutMs: options.worker.shutdownTimeoutMs,
|
||||
},
|
||||
concurrencySweeper: {
|
||||
scanSchedule: options.queue?.concurrencySweeper?.scanSchedule,
|
||||
processMarkedSchedule: options.queue?.concurrencySweeper?.processMarkedSchedule,
|
||||
scanJitterInMs: options.queue?.concurrencySweeper?.scanJitterInMs,
|
||||
processMarkedJitterInMs: options.queue?.concurrencySweeper?.processMarkedJitterInMs,
|
||||
callback: this.#concurrencySweeperCallback.bind(this),
|
||||
},
|
||||
shardCount: options.queue?.shardCount,
|
||||
masterQueueConsumersDisabled: options.queue?.masterQueueConsumersDisabled,
|
||||
masterQueueConsumersIntervalMs: options.queue?.masterQueueConsumersIntervalMs,
|
||||
@@ -1329,4 +1340,44 @@ export class RunEngine {
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
async #concurrencySweeperCallback(
|
||||
runIds: string[]
|
||||
): Promise<Array<{ id: string; orgId: string }>> {
|
||||
const runs = await this.readOnlyPrisma.taskRun.findMany({
|
||||
where: {
|
||||
id: { in: runIds },
|
||||
completedAt: {
|
||||
lte: new Date(Date.now() - 1000 * 60 * 10), // This only finds runs that were completed more than 10 minutes ago
|
||||
},
|
||||
organizationId: {
|
||||
not: null,
|
||||
},
|
||||
status: {
|
||||
in: getFinalRunStatuses(),
|
||||
},
|
||||
},
|
||||
select: {
|
||||
id: true,
|
||||
status: true,
|
||||
organizationId: true,
|
||||
},
|
||||
});
|
||||
|
||||
// Log the finished runs
|
||||
for (const run of runs) {
|
||||
this.logger.info("Concurrency sweeper callback found finished run", {
|
||||
runId: run.id,
|
||||
orgId: run.organizationId,
|
||||
status: run.status,
|
||||
});
|
||||
}
|
||||
|
||||
return runs
|
||||
.filter((run) => !!run.organizationId)
|
||||
.map((run) => ({
|
||||
id: run.id,
|
||||
orgId: run.organizationId!,
|
||||
}));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -41,21 +41,25 @@ export function isInitialState(status: TaskRunExecutionStatus): boolean {
|
||||
return startedStatuses.includes(status);
|
||||
}
|
||||
|
||||
export function isFinalRunStatus(status: TaskRunStatus): boolean {
|
||||
const finalStatuses: TaskRunStatus[] = [
|
||||
"CANCELED",
|
||||
"INTERRUPTED",
|
||||
"COMPLETED_SUCCESSFULLY",
|
||||
"COMPLETED_WITH_ERRORS",
|
||||
"SYSTEM_FAILURE",
|
||||
"CRASHED",
|
||||
"EXPIRED",
|
||||
"TIMED_OUT",
|
||||
];
|
||||
const finalStatuses: TaskRunStatus[] = [
|
||||
"CANCELED",
|
||||
"INTERRUPTED",
|
||||
"COMPLETED_SUCCESSFULLY",
|
||||
"COMPLETED_WITH_ERRORS",
|
||||
"SYSTEM_FAILURE",
|
||||
"CRASHED",
|
||||
"EXPIRED",
|
||||
"TIMED_OUT",
|
||||
];
|
||||
|
||||
export function isFinalRunStatus(status: TaskRunStatus): boolean {
|
||||
return finalStatuses.includes(status);
|
||||
}
|
||||
|
||||
export function getFinalRunStatuses(): TaskRunStatus[] {
|
||||
return finalStatuses;
|
||||
}
|
||||
|
||||
export function canReleaseConcurrency(status: TaskRunExecutionStatus): boolean {
|
||||
const releaseableStatuses: TaskRunExecutionStatus[] = ["SUSPENDED", "EXECUTING_WITH_WAITPOINTS"];
|
||||
return releaseableStatuses.includes(status);
|
||||
|
||||
@@ -8,7 +8,7 @@ import {
|
||||
RetryOptions,
|
||||
RunChainState,
|
||||
} from "@trigger.dev/core/v3";
|
||||
import { PrismaClient } from "@trigger.dev/database";
|
||||
import { PrismaClient, PrismaReplicaClient } from "@trigger.dev/database";
|
||||
import { FairQueueSelectionStrategyOptions } from "../run-queue/fairQueueSelectionStrategy.js";
|
||||
import { MinimalAuthenticatedEnvironment } from "../shared/index.js";
|
||||
import { workerCatalog } from "./workerCatalog.js";
|
||||
@@ -17,6 +17,7 @@ import { LockRetryConfig } from "./locking.js";
|
||||
|
||||
export type RunEngineOptions = {
|
||||
prisma: PrismaClient;
|
||||
readOnlyPrisma?: PrismaReplicaClient;
|
||||
worker: {
|
||||
disabled?: boolean;
|
||||
redis: RedisOptions;
|
||||
@@ -38,11 +39,18 @@ export type RunEngineOptions = {
|
||||
workerOptions?: WorkerConcurrencyOptions;
|
||||
retryOptions?: RetryOptions;
|
||||
defaultEnvConcurrency?: number;
|
||||
logLevel?: LogLevel;
|
||||
queueSelectionStrategyOptions?: Pick<
|
||||
FairQueueSelectionStrategyOptions,
|
||||
"parentQueueLimit" | "tracer" | "biases" | "reuseSnapshotCount" | "maximumEnvCount"
|
||||
>;
|
||||
dequeueBlockingTimeoutSeconds?: number;
|
||||
concurrencySweeper?: {
|
||||
scanSchedule?: string;
|
||||
processMarkedSchedule?: string;
|
||||
scanJitterInMs?: number;
|
||||
processMarkedJitterInMs?: number;
|
||||
};
|
||||
};
|
||||
runLock: {
|
||||
redis: RedisOptions;
|
||||
|
||||
@@ -36,11 +36,13 @@ import {
|
||||
type Result,
|
||||
} from "@internal/redis";
|
||||
import { MessageNotFoundError } from "./errors.js";
|
||||
import { tryCatch } from "@trigger.dev/core";
|
||||
import { promiseWithResolvers, tryCatch } from "@trigger.dev/core";
|
||||
import { setInterval } from "node:timers/promises";
|
||||
import { nanoid } from "nanoid";
|
||||
import { Worker, type WorkerConcurrencyOptions } from "@trigger.dev/redis-worker";
|
||||
import { CronSchema, Worker, type WorkerConcurrencyOptions } from "@trigger.dev/redis-worker";
|
||||
import { z } from "zod";
|
||||
import { Readable } from "node:stream";
|
||||
import { setTimeout } from "node:timers/promises";
|
||||
|
||||
const SemanticAttributes = {
|
||||
QUEUE: "runqueue.queue",
|
||||
@@ -78,14 +80,31 @@ export type RunQueueOptions = {
|
||||
};
|
||||
meter?: Meter;
|
||||
dequeueBlockingTimeoutSeconds?: number;
|
||||
concurrencySweeper?: {
|
||||
scanSchedule?: string;
|
||||
scanJitterInMs?: number;
|
||||
processMarkedSchedule?: string;
|
||||
processMarkedJitterInMs?: number;
|
||||
callback: ConcurrencySweeperCallback;
|
||||
};
|
||||
};
|
||||
|
||||
export interface ConcurrencySweeperCallback {
|
||||
(runIds: string[]): Promise<Array<{ id: string; orgId: string }>>;
|
||||
}
|
||||
|
||||
type DequeuedMessage = {
|
||||
messageId: string;
|
||||
messageScore: string;
|
||||
message: OutputPayload;
|
||||
};
|
||||
|
||||
type MarkedRun = {
|
||||
orgId: string;
|
||||
messageId: string;
|
||||
score: number;
|
||||
};
|
||||
|
||||
const defaultRetrySettings = {
|
||||
maxAttempts: 12,
|
||||
factor: 2,
|
||||
@@ -102,6 +121,24 @@ const workerCatalog = {
|
||||
}),
|
||||
visibilityTimeoutMs: 30_000,
|
||||
},
|
||||
scanConcurrencySets: {
|
||||
schema: CronSchema,
|
||||
visibilityTimeoutMs: 60_000 * 5,
|
||||
cron: "*/10 * * * *",
|
||||
jitterInMs: 60_000,
|
||||
retry: {
|
||||
maxAttempts: 1,
|
||||
},
|
||||
},
|
||||
processMarkedRuns: {
|
||||
schema: CronSchema,
|
||||
visibilityTimeoutMs: 60_000 * 5,
|
||||
cron: "*/5 * * * *",
|
||||
jitterInMs: 30_000,
|
||||
retry: {
|
||||
maxAttempts: 1,
|
||||
},
|
||||
},
|
||||
};
|
||||
|
||||
/**
|
||||
@@ -112,7 +149,7 @@ export class RunQueue {
|
||||
private subscriber: Redis;
|
||||
private luaDebugSubscriber: Redis;
|
||||
private logger: Logger;
|
||||
private redis: Redis;
|
||||
public redis: Redis;
|
||||
public keys: RunQueueKeyProducer;
|
||||
private queueSelectionStrategy: RunQueueSelectionStrategy;
|
||||
private shardCount: number;
|
||||
@@ -121,7 +158,7 @@ export class RunQueue {
|
||||
private _observableWorkerQueues: Set<string> = new Set();
|
||||
private _meter: Meter;
|
||||
|
||||
constructor(private readonly options: RunQueueOptions) {
|
||||
constructor(public readonly options: RunQueueOptions) {
|
||||
this.shardCount = options.shardCount ?? 2;
|
||||
this.retryOptions = options.retryOptions ?? defaultRetrySettings;
|
||||
this.redis = createRedisClient(options.redis, {
|
||||
@@ -170,16 +207,40 @@ export class RunQueue {
|
||||
...options.redis,
|
||||
keyPrefix: `${options.redis.keyPrefix}:worker`,
|
||||
},
|
||||
catalog: workerCatalog,
|
||||
catalog: {
|
||||
...workerCatalog,
|
||||
scanConcurrencySets: {
|
||||
...workerCatalog.scanConcurrencySets,
|
||||
cron: options.concurrencySweeper?.scanSchedule ?? workerCatalog.scanConcurrencySets.cron,
|
||||
jitter:
|
||||
options.concurrencySweeper?.scanJitterInMs ??
|
||||
workerCatalog.scanConcurrencySets.jitterInMs,
|
||||
},
|
||||
processMarkedRuns: {
|
||||
...workerCatalog.processMarkedRuns,
|
||||
cron:
|
||||
options.concurrencySweeper?.processMarkedSchedule ??
|
||||
workerCatalog.processMarkedRuns.cron,
|
||||
jitterInMs:
|
||||
options.concurrencySweeper?.processMarkedJitterInMs ??
|
||||
workerCatalog.processMarkedRuns.jitterInMs,
|
||||
},
|
||||
},
|
||||
concurrency: options.workerOptions?.concurrency,
|
||||
pollIntervalMs: options.workerOptions?.pollIntervalMs ?? 1000,
|
||||
immediatePollIntervalMs: options.workerOptions?.immediatePollIntervalMs ?? 100,
|
||||
shutdownTimeoutMs: options.workerOptions?.shutdownTimeoutMs ?? 10_000,
|
||||
logger: new Logger("RunQueueWorker", options.logLevel ?? "log"),
|
||||
logger: new Logger("RunQueueWorker", options.logLevel ?? "info"),
|
||||
jobs: {
|
||||
processQueueForWorkerQueue: async (job) => {
|
||||
await this.#processQueueForWorkerQueue(job.payload.queueKey, job.payload.environmentId);
|
||||
},
|
||||
scanConcurrencySets: async (job) => {
|
||||
await this.scanConcurrencySets();
|
||||
},
|
||||
processMarkedRuns: async (job) => {
|
||||
await this.processMarkedRuns();
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
@@ -404,38 +465,7 @@ export class RunQueue {
|
||||
}
|
||||
|
||||
public async readMessage(orgId: string, messageId: string) {
|
||||
return this.#trace(
|
||||
"readMessage",
|
||||
async (span) => {
|
||||
const rawMessage = await this.redis.get(this.keys.messageKey(orgId, messageId));
|
||||
|
||||
if (!rawMessage) {
|
||||
return;
|
||||
}
|
||||
|
||||
const message = OutputPayload.safeParse(JSON.parse(rawMessage));
|
||||
|
||||
if (!message.success) {
|
||||
this.logger.error(`[${this.name}] Failed to parse message`, {
|
||||
messageId,
|
||||
error: message.error,
|
||||
service: this.name,
|
||||
});
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
return message.data;
|
||||
},
|
||||
{
|
||||
attributes: {
|
||||
[SEMATTRS_MESSAGING_OPERATION]: "receive",
|
||||
[SEMATTRS_MESSAGE_ID]: messageId,
|
||||
[SEMATTRS_MESSAGING_SYSTEM]: "marqs",
|
||||
[SemanticAttributes.RUN_ID]: messageId,
|
||||
},
|
||||
}
|
||||
);
|
||||
return this.readMessageFromKey(this.keys.messageKey(orgId, messageId));
|
||||
}
|
||||
|
||||
public async readMessageFromKey(messageKey: string) {
|
||||
@@ -448,24 +478,34 @@ export class RunQueue {
|
||||
return;
|
||||
}
|
||||
|
||||
const message = OutputPayload.safeParse(JSON.parse(rawMessage));
|
||||
const deserializedMessage = safeJsonParse(rawMessage);
|
||||
|
||||
const message = OutputPayload.safeParse(deserializedMessage);
|
||||
|
||||
if (!message.success) {
|
||||
this.logger.error(`[${this.name}] Failed to parse message`, {
|
||||
messageKey,
|
||||
error: message.error,
|
||||
service: this.name,
|
||||
deserializedMessage,
|
||||
});
|
||||
|
||||
return;
|
||||
return deserializedMessage as OutputPayload;
|
||||
}
|
||||
|
||||
span.setAttributes({
|
||||
[SemanticAttributes.QUEUE]: message.data.queue,
|
||||
[SemanticAttributes.RUN_ID]: message.data.runId,
|
||||
[SemanticAttributes.CONCURRENCY_KEY]: message.data.concurrencyKey,
|
||||
[SemanticAttributes.WORKER_QUEUE]: this.#getWorkerQueueFromMessage(message.data),
|
||||
});
|
||||
|
||||
return message.data;
|
||||
},
|
||||
{
|
||||
attributes: {
|
||||
[SEMATTRS_MESSAGING_OPERATION]: "receive",
|
||||
[SEMATTRS_MESSAGING_SYSTEM]: "marqs",
|
||||
[SEMATTRS_MESSAGING_SYSTEM]: "runqueue",
|
||||
},
|
||||
}
|
||||
);
|
||||
@@ -896,6 +936,35 @@ export class RunQueue {
|
||||
return await this.redis.lrange(workerQueueKey, 0, -1);
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a scan stream for queue current concurrency keys
|
||||
*/
|
||||
public currentConcurrencyScanStream(
|
||||
count: number = 10,
|
||||
onEnd?: () => void,
|
||||
onError?: (error: Error) => void
|
||||
): { stream: Readable; redis: Redis } {
|
||||
const pattern = this.keys.currentConcurrencySetKeyScanPattern();
|
||||
const stream = this.redis.scanStream({
|
||||
match: pattern,
|
||||
count,
|
||||
type: "set",
|
||||
});
|
||||
|
||||
if (onEnd) {
|
||||
stream.on("end", onEnd);
|
||||
}
|
||||
|
||||
if (onError) {
|
||||
stream.on("error", onError);
|
||||
}
|
||||
|
||||
return {
|
||||
stream,
|
||||
redis: this.redis,
|
||||
};
|
||||
}
|
||||
|
||||
private async handleRedriveMessage(channel: string, message: string) {
|
||||
try {
|
||||
const { runId, envId, projectId, orgId } = JSON.parse(message) as any;
|
||||
@@ -1651,6 +1720,249 @@ export class RunQueue {
|
||||
return blockingClient;
|
||||
}
|
||||
|
||||
// Call this every 10 minutes
|
||||
private async scanConcurrencySets() {
|
||||
if (this.abortController.signal.aborted) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.logger.debug("Scanning concurrency sets for completed runs");
|
||||
|
||||
const stats = {
|
||||
streamCallbacks: 0,
|
||||
processedKeys: 0,
|
||||
};
|
||||
|
||||
const { promise, resolve, reject } = promiseWithResolvers<typeof stats>();
|
||||
|
||||
const { stream, redis } = this.currentConcurrencyScanStream(
|
||||
10,
|
||||
() => {
|
||||
this.logger.debug("Concurrency scan stream closed", { stats });
|
||||
|
||||
resolve(stats);
|
||||
},
|
||||
(error) => {
|
||||
this.logger.error("Concurrency scan stream error", {
|
||||
stats,
|
||||
error: {
|
||||
name: error.name,
|
||||
message: error.message,
|
||||
stack: error.stack,
|
||||
},
|
||||
});
|
||||
|
||||
reject(error);
|
||||
}
|
||||
);
|
||||
|
||||
stream.on("data", async (keys: string[]) => {
|
||||
if (!keys || keys.length === 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
stream.pause();
|
||||
|
||||
if (this.abortController.signal.aborted) {
|
||||
stream.destroy();
|
||||
return;
|
||||
}
|
||||
|
||||
stats.streamCallbacks++;
|
||||
|
||||
const uniqueKeys = Array.from(new Set<string>(keys)).map((key) =>
|
||||
key.replace(redis.options.keyPrefix ?? "", "")
|
||||
);
|
||||
|
||||
if (uniqueKeys.length === 0) {
|
||||
stream.resume();
|
||||
return;
|
||||
}
|
||||
|
||||
this.logger.debug("Processing concurrency keys from stream", {
|
||||
keys: uniqueKeys,
|
||||
});
|
||||
|
||||
stats.processedKeys += uniqueKeys.length;
|
||||
|
||||
await Promise.allSettled(uniqueKeys.map((key) => this.processConcurrencySet(key))).finally(
|
||||
() => {
|
||||
stream.resume();
|
||||
}
|
||||
);
|
||||
});
|
||||
|
||||
return promise;
|
||||
}
|
||||
|
||||
private async processConcurrencySet(concurrencyKey: string) {
|
||||
const stream = this.redis.sscanStream(concurrencyKey, {
|
||||
count: 100,
|
||||
});
|
||||
|
||||
const { promise, resolve, reject } = promiseWithResolvers<void>();
|
||||
|
||||
stream.on("end", () => {
|
||||
resolve();
|
||||
});
|
||||
|
||||
stream.on("error", (error) => {
|
||||
this.logger.error("Error in sscanStream for concurrency set", {
|
||||
concurrencyKey,
|
||||
error,
|
||||
});
|
||||
|
||||
reject(error);
|
||||
});
|
||||
|
||||
stream.on("data", async (runIds: string[]) => {
|
||||
stream.pause();
|
||||
|
||||
if (this.abortController.signal.aborted) {
|
||||
stream.destroy();
|
||||
return;
|
||||
}
|
||||
|
||||
if (!runIds || runIds.length === 0) {
|
||||
stream.resume();
|
||||
return;
|
||||
}
|
||||
|
||||
const deduplicatedRunIds = Array.from(new Set(runIds));
|
||||
|
||||
const [processError] = await tryCatch(
|
||||
this.processCurrentConcurrencyRunIds(concurrencyKey, deduplicatedRunIds)
|
||||
);
|
||||
|
||||
if (processError) {
|
||||
this.logger.error("Error processing concurrency set", {
|
||||
concurrencyKey,
|
||||
runIds,
|
||||
error: processError,
|
||||
});
|
||||
}
|
||||
|
||||
stream.resume();
|
||||
});
|
||||
|
||||
return promise;
|
||||
}
|
||||
|
||||
private async processCurrentConcurrencyRunIds(concurrencyKey: string, runIds: string[]) {
|
||||
this.logger.debug(`Processing concurrency set with ${runIds.length} runs`, {
|
||||
concurrencyKey,
|
||||
runIds: runIds.slice(0, 5), // Log first 5 for debugging
|
||||
});
|
||||
|
||||
// Call the callback to determine which runs are completed
|
||||
const completedRuns = await this.options.concurrencySweeper?.callback(runIds);
|
||||
|
||||
if (!completedRuns) {
|
||||
this.logger.debug("No completed runs found in concurrency set", { concurrencyKey });
|
||||
return;
|
||||
}
|
||||
|
||||
if (completedRuns.length === 0) {
|
||||
this.logger.debug("No completed runs found in concurrency set", { concurrencyKey });
|
||||
return;
|
||||
}
|
||||
|
||||
this.logger.debug(`Found ${completedRuns.length} completed runs to mark for ack`, {
|
||||
concurrencyKey,
|
||||
completedRunIds: completedRuns.map((r) => r.id).slice(0, 5),
|
||||
});
|
||||
|
||||
// Mark the completed runs for acknowledgment
|
||||
await this.markRunsForAck(completedRuns);
|
||||
}
|
||||
|
||||
private async markRunsForAck(completedRuns: Array<{ id: string; orgId: string }>) {
|
||||
const markedForAckKey = this.keys.markedForAckKey();
|
||||
|
||||
// Prepare arguments: alternating orgId, messageId pairs
|
||||
const args: Array<number | string> = [];
|
||||
for (const run of completedRuns) {
|
||||
this.logger.info("Marking run for acknowledgment", {
|
||||
orgId: run.orgId,
|
||||
runId: run.id,
|
||||
});
|
||||
|
||||
args.push(Date.now());
|
||||
args.push(`${run.orgId}:${run.id}`);
|
||||
}
|
||||
|
||||
const count = await this.redis.zadd(markedForAckKey, ...args);
|
||||
|
||||
this.logger.debug(`Marked ${count} runs for acknowledgment`, {
|
||||
markedForAckKey,
|
||||
count,
|
||||
});
|
||||
}
|
||||
|
||||
// Call this every 5 minutes
|
||||
private async processMarkedRuns() {
|
||||
if (this.abortController.signal.aborted) {
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
const markedForAckKey = this.keys.markedForAckKey();
|
||||
const results = await this.redis.getMarkedRunsForAck(markedForAckKey, "100");
|
||||
|
||||
if (results.length === 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
const markedRuns: MarkedRun[] = [];
|
||||
|
||||
// Parse results: [orgId1, messageId1, score1, orgId2, messageId2, score2, ...]
|
||||
for (let i = 0; i < results.length; i += 3) {
|
||||
markedRuns.push({
|
||||
orgId: results[i],
|
||||
messageId: results[i + 1],
|
||||
score: Number(results[i + 2]),
|
||||
});
|
||||
}
|
||||
|
||||
this.logger.debug(`Processing ${markedRuns.length} marked runs for acknowledgment`, {
|
||||
markedRuns: markedRuns, // Log first 3 for debugging
|
||||
});
|
||||
|
||||
for (const run of markedRuns) {
|
||||
const [processError] = await tryCatch(this.processMarkedRun(run));
|
||||
|
||||
if (processError) {
|
||||
this.logger.error("Error processing marked run", {
|
||||
error: processError,
|
||||
orgId: run.orgId,
|
||||
messageId: run.messageId,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
const shouldProcessMoreRuns = (await this.redis.zcard(markedForAckKey)) > 0;
|
||||
|
||||
if (shouldProcessMoreRuns) {
|
||||
await setTimeout(1000);
|
||||
await this.processMarkedRuns();
|
||||
}
|
||||
} catch (error) {
|
||||
this.logger.error("Error processing marked runs", { error });
|
||||
}
|
||||
}
|
||||
|
||||
async processMarkedRun(run: MarkedRun) {
|
||||
this.logger.info("Acknowledging marked run", {
|
||||
orgId: run.orgId,
|
||||
messageId: run.messageId,
|
||||
});
|
||||
|
||||
await this.acknowledgeMessage(run.orgId, run.messageId, {
|
||||
skipDequeueProcessing: true,
|
||||
removeFromWorkerQueue: false,
|
||||
});
|
||||
}
|
||||
|
||||
#registerCommands() {
|
||||
this.redis.defineCommand("migrateLegacyMasterQueues", {
|
||||
numberOfKeys: 1,
|
||||
@@ -2020,6 +2332,77 @@ local envConcurrencyLimit = ARGV[1]
|
||||
redis.call('SET', envConcurrencyLimitKey, envConcurrencyLimit)
|
||||
`,
|
||||
});
|
||||
|
||||
this.redis.defineCommand("markCompletedRunsForAck", {
|
||||
numberOfKeys: 1,
|
||||
lua: `
|
||||
-- Keys:
|
||||
local markedForAckKey = KEYS[1]
|
||||
|
||||
-- Args: alternating orgId, messageId pairs
|
||||
local currentTime = tonumber(redis.call('TIME')[1]) * 1000
|
||||
|
||||
for i = 1, #ARGV, 2 do
|
||||
local orgId = ARGV[i]
|
||||
local messageId = ARGV[i + 1]
|
||||
local markedValue = orgId .. ':' .. messageId
|
||||
|
||||
redis.call('ZADD', markedForAckKey, currentTime, markedValue)
|
||||
end
|
||||
|
||||
return #ARGV / 2
|
||||
`,
|
||||
});
|
||||
|
||||
this.redis.defineCommand("getMarkedRunsForAck", {
|
||||
numberOfKeys: 1,
|
||||
lua: `
|
||||
-- Keys:
|
||||
local markedForAckKey = KEYS[1]
|
||||
|
||||
-- Args:
|
||||
local maxCount = tonumber(ARGV[1] or '10')
|
||||
|
||||
-- Get the oldest marked runs
|
||||
local markedRuns = redis.call('ZRANGE', markedForAckKey, 0, maxCount - 1, 'WITHSCORES')
|
||||
|
||||
local results = {}
|
||||
for i = 1, #markedRuns, 2 do
|
||||
local markedValue = markedRuns[i]
|
||||
local score = markedRuns[i + 1]
|
||||
|
||||
-- Parse orgId:messageId
|
||||
local colonIndex = string.find(markedValue, ':')
|
||||
if colonIndex then
|
||||
local orgId = string.sub(markedValue, 1, colonIndex - 1)
|
||||
local messageId = string.sub(markedValue, colonIndex + 1)
|
||||
|
||||
table.insert(results, orgId)
|
||||
table.insert(results, messageId)
|
||||
table.insert(results, score)
|
||||
end
|
||||
end
|
||||
|
||||
-- Remove the processed items
|
||||
if #results > 0 then
|
||||
local itemsToRemove = {}
|
||||
for i = 1, #markedRuns, 2 do
|
||||
table.insert(itemsToRemove, markedRuns[i])
|
||||
end
|
||||
redis.call('ZREM', markedForAckKey, unpack(itemsToRemove))
|
||||
end
|
||||
|
||||
return results
|
||||
`,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
function safeJsonParse(rawMessage: string): unknown {
|
||||
try {
|
||||
return JSON.parse(rawMessage);
|
||||
} catch (e) {
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2145,5 +2528,11 @@ declare module "@internal/redis" {
|
||||
keyPrefix: string,
|
||||
...queueNames: string[]
|
||||
): Result<void, Context>;
|
||||
|
||||
getMarkedRunsForAck(
|
||||
markedForAckKey: string,
|
||||
maxCount: string,
|
||||
callback?: Callback<string[]>
|
||||
): Result<string[], Context>;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -219,9 +219,16 @@ export class RunQueueFullKeyProducer implements RunQueueKeyProducer {
|
||||
}
|
||||
}
|
||||
deadLetterQueueKeyFromQueue(queue: string): string {
|
||||
const descriptor = this.descriptorFromQueue(queue);
|
||||
const { orgId, projectId, envId } = this.descriptorFromQueue(queue);
|
||||
return this.deadLetterQueueKey({ orgId, projectId, envId });
|
||||
}
|
||||
|
||||
return this.deadLetterQueueKey(descriptor);
|
||||
markedForAckKey(): string {
|
||||
return "markedForAck";
|
||||
}
|
||||
|
||||
currentConcurrencySetKeyScanPattern(): string {
|
||||
return `*:${constants.ENV_PART}:*:queue:*:${constants.CURRENT_CONCURRENCY_PART}`;
|
||||
}
|
||||
|
||||
descriptorFromQueue(queue: string): QueueDescriptor {
|
||||
|
||||
@@ -0,0 +1,171 @@
|
||||
import { redisTest } from "@internal/testcontainers";
|
||||
import { trace } from "@internal/tracing";
|
||||
import { Logger } from "@trigger.dev/core/logger";
|
||||
import { describe } from "node:test";
|
||||
import { setTimeout } from "node:timers/promises";
|
||||
import { FairQueueSelectionStrategy } from "../fairQueueSelectionStrategy.js";
|
||||
import { RunQueue } from "../index.js";
|
||||
import { RunQueueFullKeyProducer } from "../keyProducer.js";
|
||||
import { InputPayload } from "../types.js";
|
||||
|
||||
const testOptions = {
|
||||
name: "rq",
|
||||
tracer: trace.getTracer("rq"),
|
||||
workers: 1,
|
||||
defaultEnvConcurrency: 25,
|
||||
logger: new Logger("RunQueue", "warn"),
|
||||
retryOptions: {
|
||||
maxAttempts: 5,
|
||||
factor: 1.1,
|
||||
minTimeoutInMs: 100,
|
||||
maxTimeoutInMs: 1_000,
|
||||
randomize: true,
|
||||
},
|
||||
keys: new RunQueueFullKeyProducer(),
|
||||
};
|
||||
|
||||
const authenticatedEnvDev = {
|
||||
id: "e1234",
|
||||
type: "DEVELOPMENT" as const,
|
||||
maximumConcurrencyLimit: 10,
|
||||
project: { id: "p1234" },
|
||||
organization: { id: "o1234" },
|
||||
};
|
||||
|
||||
const messageDev: InputPayload = {
|
||||
runId: "r4321",
|
||||
taskIdentifier: "task/my-task",
|
||||
orgId: "o1234",
|
||||
projectId: "p1234",
|
||||
environmentId: "e4321",
|
||||
environmentType: "DEVELOPMENT",
|
||||
queue: "task/my-task",
|
||||
timestamp: Date.now(),
|
||||
attempt: 0,
|
||||
};
|
||||
|
||||
const messageDev2: InputPayload = {
|
||||
...messageDev,
|
||||
runId: "r4322",
|
||||
};
|
||||
|
||||
vi.setConfig({ testTimeout: 60_000 });
|
||||
|
||||
describe("RunQueue Concurrency Sweeper", () => {
|
||||
redisTest(
|
||||
"should process queue current concurrency sets and mark runs for ack if they are completed",
|
||||
async ({ redisContainer }) => {
|
||||
let enableConcurrencySweeper = false;
|
||||
|
||||
const queue = new RunQueue({
|
||||
...testOptions,
|
||||
queueSelectionStrategy: new FairQueueSelectionStrategy({
|
||||
redis: {
|
||||
keyPrefix: "runqueue:test:",
|
||||
host: redisContainer.getHost(),
|
||||
port: redisContainer.getPort(),
|
||||
},
|
||||
keys: testOptions.keys,
|
||||
}),
|
||||
redis: {
|
||||
keyPrefix: "runqueue:test:",
|
||||
host: redisContainer.getHost(),
|
||||
port: redisContainer.getPort(),
|
||||
},
|
||||
concurrencySweeper: {
|
||||
scanSchedule: "* * * * * *", // Every second
|
||||
scanJitterInMs: 5,
|
||||
processMarkedSchedule: "* * * * * *", // Every second
|
||||
processMarkedJitterInMs: 5,
|
||||
callback: async (runIds) => {
|
||||
if (!enableConcurrencySweeper) {
|
||||
return [];
|
||||
}
|
||||
|
||||
return [{ id: messageDev.runId, orgId: "o1234" }];
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
try {
|
||||
//enqueue message
|
||||
const enqueueResult = await queue.enqueueMessage({
|
||||
env: authenticatedEnvDev,
|
||||
message: messageDev,
|
||||
workerQueue: authenticatedEnvDev.id,
|
||||
});
|
||||
|
||||
expect(enqueueResult).toBe(undefined);
|
||||
|
||||
const enqueueResult2 = await queue.enqueueMessage({
|
||||
env: authenticatedEnvDev,
|
||||
message: messageDev2,
|
||||
workerQueue: authenticatedEnvDev.id,
|
||||
});
|
||||
|
||||
expect(enqueueResult2).toBe(undefined);
|
||||
|
||||
//queue length
|
||||
const result2 = await queue.lengthOfQueue(authenticatedEnvDev, messageDev.queue);
|
||||
expect(result2).toBe(2);
|
||||
|
||||
const envQueueLength2 = await queue.lengthOfEnvQueue(authenticatedEnvDev);
|
||||
expect(envQueueLength2).toBe(2);
|
||||
|
||||
//concurrencies
|
||||
const queueConcurrency = await queue.currentConcurrencyOfQueue(
|
||||
authenticatedEnvDev,
|
||||
messageDev.queue
|
||||
);
|
||||
expect(queueConcurrency).toBe(0);
|
||||
|
||||
const envConcurrency = await queue.currentConcurrencyOfEnvironment(authenticatedEnvDev);
|
||||
expect(envConcurrency).toBe(0);
|
||||
|
||||
//dequeue message
|
||||
const dequeued = await queue.dequeueMessageFromWorkerQueue(
|
||||
"test_12345",
|
||||
authenticatedEnvDev.id
|
||||
);
|
||||
|
||||
expect(dequeued).toBeDefined();
|
||||
expect(dequeued?.messageId).toEqual(messageDev.runId);
|
||||
|
||||
const dequeued2 = await queue.dequeueMessageFromWorkerQueue(
|
||||
"test_12345",
|
||||
authenticatedEnvDev.id
|
||||
);
|
||||
expect(dequeued2).toBeDefined();
|
||||
expect(dequeued2?.messageId).toEqual(messageDev2.runId);
|
||||
|
||||
// queue concurrency should be 2
|
||||
const queueConcurrency2 = await queue.currentConcurrencyOfQueue(
|
||||
authenticatedEnvDev,
|
||||
messageDev.queue
|
||||
);
|
||||
expect(queueConcurrency2).toBe(2);
|
||||
|
||||
// env concurrency should be 2
|
||||
const envConcurrency2 = await queue.currentConcurrencyOfEnvironment(authenticatedEnvDev);
|
||||
expect(envConcurrency2).toBe(2);
|
||||
|
||||
enableConcurrencySweeper = true;
|
||||
|
||||
await setTimeout(5_000); // Now a run is "completed" and should be removed from the concurrency set
|
||||
|
||||
// queue concurrency should be 1
|
||||
const queueConcurrency3 = await queue.currentConcurrencyOfQueue(
|
||||
authenticatedEnvDev,
|
||||
messageDev.queue
|
||||
);
|
||||
expect(queueConcurrency3).toBe(1);
|
||||
|
||||
// env concurrency should be 1
|
||||
const envConcurrency3 = await queue.currentConcurrencyOfEnvironment(authenticatedEnvDev);
|
||||
expect(envConcurrency3).toBe(1);
|
||||
} finally {
|
||||
await queue.quit();
|
||||
}
|
||||
}
|
||||
);
|
||||
});
|
||||
@@ -96,6 +96,10 @@ export interface RunQueueKeyProducer {
|
||||
deadLetterQueueKey(env: MinimalAuthenticatedEnvironment): string;
|
||||
deadLetterQueueKey(env: EnvDescriptor): string;
|
||||
deadLetterQueueKeyFromQueue(queue: string): string;
|
||||
|
||||
// Concurrency sweeper methods
|
||||
markedForAckKey(): string;
|
||||
currentConcurrencySetKeyScanPattern(): string;
|
||||
}
|
||||
|
||||
export type EnvQueues = {
|
||||
|
||||
@@ -27,7 +27,8 @@
|
||||
"lodash.omit": "^4.5.0",
|
||||
"nanoid": "^5.0.7",
|
||||
"p-limit": "^6.2.0",
|
||||
"zod": "3.23.8"
|
||||
"zod": "3.23.8",
|
||||
"cron-parser": "^4.9.0"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@internal/redis": "workspace:*",
|
||||
|
||||
@@ -0,0 +1,130 @@
|
||||
import { redisTest } from "@internal/testcontainers";
|
||||
import { Logger } from "@trigger.dev/core/logger";
|
||||
import { describe } from "node:test";
|
||||
import { expect } from "vitest";
|
||||
import { Worker, CronSchema } from "./worker.js";
|
||||
import { setTimeout } from "node:timers/promises";
|
||||
|
||||
describe("Worker with cron", () => {
|
||||
redisTest(
|
||||
"process items on the cron schedule",
|
||||
{ timeout: 180_000 },
|
||||
async ({ redisContainer }) => {
|
||||
const processedItems: CronSchema[] = [];
|
||||
const worker = new Worker({
|
||||
name: "test-worker",
|
||||
redisOptions: {
|
||||
host: redisContainer.getHost(),
|
||||
port: redisContainer.getPort(),
|
||||
password: redisContainer.getPassword(),
|
||||
},
|
||||
catalog: {
|
||||
cronJob: {
|
||||
cron: "*/5 * * * * *", // Every 5 seconds
|
||||
schema: CronSchema,
|
||||
visibilityTimeoutMs: 5000,
|
||||
retry: { maxAttempts: 3 },
|
||||
jitterInMs: 100,
|
||||
},
|
||||
},
|
||||
jobs: {
|
||||
cronJob: async ({ payload }) => {
|
||||
await setTimeout(30); // Simulate work
|
||||
processedItems.push(payload);
|
||||
},
|
||||
},
|
||||
concurrency: {
|
||||
workers: 2,
|
||||
tasksPerWorker: 3,
|
||||
},
|
||||
logger: new Logger("test", "debug"),
|
||||
}).start();
|
||||
|
||||
await setTimeout(6_000);
|
||||
|
||||
expect(processedItems.length).toBe(1);
|
||||
|
||||
const firstItem = processedItems[0];
|
||||
|
||||
expect(firstItem?.timestamp).toBeGreaterThan(0);
|
||||
expect(firstItem?.lastTimestamp).toBeUndefined();
|
||||
expect(firstItem?.cron).toBe("*/5 * * * * *");
|
||||
|
||||
await setTimeout(6_000);
|
||||
|
||||
expect(processedItems.length).toBeGreaterThanOrEqual(2);
|
||||
|
||||
const secondItem = processedItems[1];
|
||||
expect(secondItem?.timestamp).toBeGreaterThan(firstItem!.timestamp);
|
||||
expect(secondItem?.lastTimestamp).toBe(firstItem?.timestamp);
|
||||
expect(secondItem?.cron).toBe("*/5 * * * * *");
|
||||
|
||||
await worker.stop();
|
||||
}
|
||||
);
|
||||
|
||||
redisTest(
|
||||
"continues processing cron items even when job handler throws errors",
|
||||
{ timeout: 180_000 },
|
||||
async ({ redisContainer }) => {
|
||||
const processedItems: CronSchema[] = [];
|
||||
let executionCount = 0;
|
||||
|
||||
const worker = new Worker({
|
||||
name: "test-worker-error",
|
||||
redisOptions: {
|
||||
host: redisContainer.getHost(),
|
||||
port: redisContainer.getPort(),
|
||||
password: redisContainer.getPassword(),
|
||||
},
|
||||
catalog: {
|
||||
cronJob: {
|
||||
cron: "*/3 * * * * *", // Every 3 seconds
|
||||
schema: CronSchema,
|
||||
visibilityTimeoutMs: 5000,
|
||||
retry: { maxAttempts: 1 }, // Only try once to fail faster
|
||||
jitterInMs: 100,
|
||||
},
|
||||
},
|
||||
jobs: {
|
||||
cronJob: async ({ payload }) => {
|
||||
executionCount++;
|
||||
await setTimeout(30); // Simulate work
|
||||
|
||||
// Throw error on first and third execution
|
||||
if (executionCount === 1 || executionCount === 3) {
|
||||
throw new Error(`Simulated error on execution ${executionCount}`);
|
||||
}
|
||||
|
||||
processedItems.push(payload);
|
||||
},
|
||||
},
|
||||
concurrency: {
|
||||
workers: 2,
|
||||
tasksPerWorker: 3,
|
||||
},
|
||||
logger: new Logger("test", "debug"),
|
||||
}).start();
|
||||
|
||||
// Wait long enough for 4 executions (12 seconds + buffer)
|
||||
await setTimeout(14_000);
|
||||
|
||||
// Should have at least 4 executions total
|
||||
expect(executionCount).toBeGreaterThanOrEqual(4);
|
||||
|
||||
// Should have 2 successful items (executions 2 and 4)
|
||||
expect(processedItems.length).toBeGreaterThanOrEqual(2);
|
||||
|
||||
// Verify that some executions failed (execution count > successful count)
|
||||
// This proves that errors occurred but cron scheduling continued
|
||||
expect(executionCount).toBeGreaterThan(processedItems.length);
|
||||
|
||||
// Verify that successful executions still have correct structure
|
||||
const firstSuccessful = processedItems[0];
|
||||
expect(firstSuccessful?.timestamp).toBeGreaterThan(0);
|
||||
expect(firstSuccessful?.cron).toBe("*/3 * * * * *");
|
||||
|
||||
await worker.stop();
|
||||
}
|
||||
);
|
||||
});
|
||||
@@ -19,15 +19,28 @@ import { nanoid } from "nanoid";
|
||||
import pLimit from "p-limit";
|
||||
import { z } from "zod";
|
||||
import { AnyQueueItem, SimpleQueue } from "./queue.js";
|
||||
import { parseExpression } from "cron-parser";
|
||||
|
||||
export const CronSchema = z.object({
|
||||
cron: z.string(),
|
||||
lastTimestamp: z.number().optional(),
|
||||
timestamp: z.number(),
|
||||
});
|
||||
|
||||
export type CronSchema = z.infer<typeof CronSchema>;
|
||||
|
||||
export type WorkerCatalog = {
|
||||
[key: string]: {
|
||||
schema: z.ZodFirstPartySchemaTypes | z.ZodDiscriminatedUnion<any, any>;
|
||||
visibilityTimeoutMs: number;
|
||||
retry?: RetryOptions;
|
||||
cron?: string;
|
||||
jitterInMs?: number;
|
||||
};
|
||||
};
|
||||
|
||||
type WorkerCatalogItem = WorkerCatalog[keyof WorkerCatalog];
|
||||
|
||||
type QueueCatalogFromWorkerCatalog<Catalog extends WorkerCatalog> = {
|
||||
[K in keyof Catalog]: Catalog[K]["schema"];
|
||||
};
|
||||
@@ -204,6 +217,12 @@ class Worker<TCatalog extends WorkerCatalog> {
|
||||
public start() {
|
||||
const { workers, tasksPerWorker } = this.concurrency;
|
||||
|
||||
this.logger.info("Starting worker", {
|
||||
workers,
|
||||
tasksPerWorker,
|
||||
concurrency: this.concurrency,
|
||||
});
|
||||
|
||||
// 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, i, workers));
|
||||
@@ -219,7 +238,9 @@ class Worker<TCatalog extends WorkerCatalog> {
|
||||
});
|
||||
},
|
||||
});
|
||||
|
||||
this.setupSubscriber();
|
||||
this.setupCron();
|
||||
|
||||
return this;
|
||||
}
|
||||
@@ -496,6 +517,11 @@ class Worker<TCatalog extends WorkerCatalog> {
|
||||
return;
|
||||
}
|
||||
|
||||
if (!catalogItem) {
|
||||
this.logger.error(`No catalog item found for job type: ${job}`);
|
||||
return;
|
||||
}
|
||||
|
||||
await startSpan(
|
||||
this.tracer,
|
||||
"processItem",
|
||||
@@ -513,6 +539,10 @@ class Worker<TCatalog extends WorkerCatalog> {
|
||||
|
||||
// On success, acknowledge the item.
|
||||
await this.queue.ack(id, deduplicationKey);
|
||||
|
||||
if (catalogItem.cron) {
|
||||
await this.rescheduleCronJob(job, catalogItem, item);
|
||||
}
|
||||
},
|
||||
{
|
||||
kind: SpanKind.CONSUMER,
|
||||
@@ -560,7 +590,13 @@ class Worker<TCatalog extends WorkerCatalog> {
|
||||
attempt: newAttempt,
|
||||
errorMessage,
|
||||
});
|
||||
|
||||
await this.queue.moveToDeadLetterQueue(id, errorMessage);
|
||||
|
||||
if (catalogItem.cron) {
|
||||
await this.rescheduleCronJob(job, catalogItem, item);
|
||||
}
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -622,6 +658,113 @@ class Worker<TCatalog extends WorkerCatalog> {
|
||||
return new Promise((resolve) => setTimeout(resolve, ms));
|
||||
}
|
||||
|
||||
private setupCron() {
|
||||
const cronJobs = Object.entries(this.options.catalog).filter(([_, value]) => value.cron);
|
||||
|
||||
if (cronJobs.length === 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.logger.info("Setting up cron jobs", {
|
||||
cronJobs: cronJobs.map(([job, value]) => ({
|
||||
job,
|
||||
cron: value.cron,
|
||||
jitterInMs: value.jitterInMs,
|
||||
})),
|
||||
});
|
||||
|
||||
// For each cron job, we need to try and enqueue a job with the next timestamp of the cron job.
|
||||
const enqueuePromises = cronJobs.map(([job, value]) =>
|
||||
this.enqueueCronJob(value.cron!, job, value.jitterInMs)
|
||||
);
|
||||
|
||||
Promise.allSettled(enqueuePromises).then((results) => {
|
||||
results.forEach((result) => {
|
||||
if (result.status === "fulfilled") {
|
||||
this.logger.info("Enqueued cron job", { result: result.value });
|
||||
} else {
|
||||
this.logger.error("Failed to enqueue cron job", { reason: result.reason });
|
||||
}
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
private async enqueueCronJob(cron: string, job: string, jitter?: number, lastTimestamp?: Date) {
|
||||
const scheduledAt = this.calculateNextScheduledAt(cron, lastTimestamp);
|
||||
const identifier = [job, this.timestampIdentifier(scheduledAt)].join(":");
|
||||
// Calculate the availableAt date by calculating a random number between -jitter/2 and jitter/2 and adding it to the scheduledAt
|
||||
const availableAt = jitter
|
||||
? new Date(scheduledAt.getTime() + Math.random() * jitter - jitter / 2)
|
||||
: scheduledAt;
|
||||
|
||||
const enqueued = await this.enqueueOnce({
|
||||
id: identifier,
|
||||
job,
|
||||
payload: {
|
||||
timestamp: scheduledAt.getTime(),
|
||||
lastTimestamp: lastTimestamp?.getTime(),
|
||||
cron,
|
||||
},
|
||||
availableAt,
|
||||
});
|
||||
|
||||
this.logger.info("Enqueued cron job", {
|
||||
identifier,
|
||||
cron,
|
||||
job,
|
||||
scheduledAt,
|
||||
enqueued,
|
||||
availableAt,
|
||||
});
|
||||
|
||||
return {
|
||||
identifier,
|
||||
cron,
|
||||
job,
|
||||
scheduledAt,
|
||||
enqueued,
|
||||
};
|
||||
}
|
||||
|
||||
private async rescheduleCronJob(job: string, catalogItem: WorkerCatalogItem, item: CronSchema) {
|
||||
if (!catalogItem.cron) {
|
||||
return;
|
||||
}
|
||||
|
||||
return this.enqueueCronJob(
|
||||
catalogItem.cron,
|
||||
job,
|
||||
catalogItem.jitterInMs,
|
||||
new Date(item.timestamp)
|
||||
);
|
||||
}
|
||||
|
||||
private calculateNextScheduledAt(cron: string, lastTimestamp?: Date): Date {
|
||||
const scheduledAt = parseExpression(cron, {
|
||||
currentDate: lastTimestamp,
|
||||
})
|
||||
.next()
|
||||
.toDate();
|
||||
|
||||
// If scheduledAt is in the past, we should just calculate the next one based on the current time
|
||||
if (scheduledAt < new Date()) {
|
||||
return this.calculateNextScheduledAt(cron);
|
||||
}
|
||||
|
||||
return scheduledAt;
|
||||
}
|
||||
|
||||
private timestampIdentifier(timestamp: Date) {
|
||||
const year = timestamp.getUTCFullYear();
|
||||
const month = timestamp.getUTCMonth();
|
||||
const day = timestamp.getUTCDate();
|
||||
const hour = timestamp.getUTCHours();
|
||||
const minute = timestamp.getUTCMinutes();
|
||||
const second = timestamp.getUTCSeconds();
|
||||
|
||||
return `${year}-${month}-${day}-${hour}-${minute}-${second}`;
|
||||
}
|
||||
|
||||
private setupSubscriber() {
|
||||
const channel = `${this.options.name}:redrive`;
|
||||
this.subscriber?.subscribe(channel, (err) => {
|
||||
|
||||
Generated
+3
@@ -1663,6 +1663,9 @@ importers:
|
||||
'@trigger.dev/core':
|
||||
specifier: workspace:4.0.0-v4-beta.21
|
||||
version: link:../core
|
||||
cron-parser:
|
||||
specifier: ^4.9.0
|
||||
version: 4.9.0
|
||||
lodash.omit:
|
||||
specifier: ^4.5.0
|
||||
version: 4.5.0
|
||||
|
||||
@@ -5,7 +5,7 @@ export default defineConfig({
|
||||
compatibilityFlags: ["run_engine_v2"],
|
||||
project: "proj_rrkpdguyagvsoktglnod",
|
||||
logLevel: "log",
|
||||
maxDuration: 60,
|
||||
maxDuration: 3600,
|
||||
retries: {
|
||||
enabledInDev: true,
|
||||
default: {
|
||||
|
||||
Reference in New Issue
Block a user