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

runs replication leader lock expiration fix (#2050)

Browse Source 
* runs replication leader lock expiration fix

* Allow configuring the container image --max-old-space-size using NODE_MAX_OLD_SPACE_SIZE

* Ability to configure the clickhouse keep alive settings

* Add some logging because we might not be able to do telemetry
This commit is contained in:
Eric Allam
2025-05-14 15:49:48 +01:00
committed by GitHub
parent 0661ee5fdd
commit eb3929880f
8 changed files with 320 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/webapp/app/env.server.ts
+3 -1
View File
@@ -766,9 +766,11 @@ const EnvironmentSchema = z.object({
RUN_REPLICATION_LEADER_LOCK_EXTEND_INTERVAL_MS: z.coerce.number().int().default(10_000),
RUN_REPLICATION_ACK_INTERVAL_SECONDS: z.coerce.number().int().default(10),
RUN_REPLICATION_LOG_LEVEL: z.enum(["log", "error", "warn", "info", "debug"]).default("info"),
RUN_REPLICATION_LEADER_LOCK_RETRY_COUNT: z.coerce.number().int().default(240),
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_IDLE_SOCKET_TTL_MS: z.coerce.number().int().default(9_000),
});
export type Environment = z.infer<typeof EnvironmentSchema>;
apps/webapp/app/services/runsReplicationInstance.server.ts
+6 -1
View File
@@ -23,6 +23,11 @@ function initializeRunsReplicationInstance() {
const clickhouse = new ClickHouse({
url: env.RUN_REPLICATION_CLICKHOUSE_URL,
name: "runs-replication",
keepAlive: {
enabled: env.RUN_REPLICATION_KEEP_ALIVE_ENABLED === "1",
idleSocketTtl: env.RUN_REPLICATION_KEEP_ALIVE_IDLE_SOCKET_TTL_MS,
},
logLevel: env.RUN_REPLICATION_LOG_LEVEL,
});
const service = new RunsReplicationService({
@@ -45,7 +50,7 @@ function initializeRunsReplicationInstance() {
flushBatchSize: env.RUN_REPLICATION_FLUSH_BATCH_SIZE,
leaderLockTimeoutMs: env.RUN_REPLICATION_LEADER_LOCK_TIMEOUT_MS,
leaderLockExtendIntervalMs: env.RUN_REPLICATION_LEADER_LOCK_EXTEND_INTERVAL_MS,
leaderLockRetryCount: env.RUN_REPLICATION_LEADER_LOCK_RETRY_COUNT,
leaderLockAcquireAdditionalTimeMs: env.RUN_REPLICATION_LEADER_LOCK_ADDITIONAL_TIME_MS,
leaderLockRetryIntervalMs: env.RUN_REPLICATION_LEADER_LOCK_RETRY_INTERVAL_MS,
ackIntervalSeconds: env.RUN_REPLICATION_ACK_INTERVAL_SECONDS,
logLevel: env.RUN_REPLICATION_LOG_LEVEL,
apps/webapp/app/services/runsReplicationService.server.ts
+60 -31
View File
@@ -43,7 +43,7 @@ export type RunsReplicationServiceOptions = {
flushBatchSize?: number;
leaderLockTimeoutMs?: number;
leaderLockExtendIntervalMs?: number;
leaderLockRetryCount?: number;
leaderLockAcquireAdditionalTimeMs?: number;
leaderLockRetryIntervalMs?: number;
ackIntervalSeconds?: number;
acknowledgeTimeoutMs?: number;
@@ -102,11 +102,11 @@ export class RunsReplicationService {
redisOptions: options.redisOptions,
autoAcknowledge: false,
publicationActions: ["insert", "update", "delete"],
logger: new Logger("LogicalReplicationClient", options.logLevel ?? "info"),
logger: options.logger ?? new Logger("LogicalReplicationClient", options.logLevel ?? "info"),
leaderLockTimeoutMs: options.leaderLockTimeoutMs ?? 30_000,
leaderLockExtendIntervalMs: options.leaderLockExtendIntervalMs ?? 10_000,
ackIntervalSeconds: options.ackIntervalSeconds ?? 10,
leaderLockRetryCount: options.leaderLockRetryCount ?? 240,
leaderLockAcquireAdditionalTimeMs: options.leaderLockAcquireAdditionalTimeMs ?? 10_000,
leaderLockRetryIntervalMs: options.leaderLockRetryIntervalMs ?? 500,
tracer: options.tracer,
});
@@ -330,10 +330,6 @@ export class RunsReplicationService {
return;
}
this.logger.debug("Handling transaction", {
transaction,
});
const lsnToUInt64Start = process.hrtime.bigint();
// If there are events, we need to handle them
@@ -349,20 +345,32 @@ export class RunsReplicationService {
}))
);
const currentSpan = this._tracer.startSpan("handle_transaction", {
attributes: {
"transaction.xid": transaction.xid,
"transaction.replication_lag_ms": transaction.replicationLagMs,
"transaction.events": transaction.events.length,
"transaction.commit_end_lsn": transaction.commitEndLsn,
"transaction.parse_duration_ms": this._currentParseDurationMs ?? undefined,
"transaction.lsn_to_uint64_ms": lsnToUInt64DurationMs,
"transaction.version": _version.toString(),
},
startTime: transaction.beginStartTimestamp,
});
this._tracer
.startSpan("handle_transaction", {
attributes: {
"transaction.xid": transaction.xid,
"transaction.replication_lag_ms": transaction.replicationLagMs,
"transaction.events": transaction.events.length,
"transaction.commit_end_lsn": transaction.commitEndLsn,
"transaction.parse_duration_ms": this._currentParseDurationMs ?? undefined,
"transaction.lsn_to_uint64_ms": lsnToUInt64DurationMs,
"transaction.version": _version.toString(),
},
startTime: transaction.beginStartTimestamp,
})
.end();
currentSpan.end();
this.logger.debug("handle_transaction", {
transaction: {
xid: transaction.xid,
commitLsn: transaction.commitLsn,
commitEndLsn: transaction.commitEndLsn,
events: transaction.events.length,
parseDurationMs: this._currentParseDurationMs,
lsnToUInt64DurationMs,
version: _version.toString(),
},
});
}
async #acknowledgeLatestTransaction() {
@@ -387,7 +395,7 @@ export class RunsReplicationService {
this._lastAcknowledgedAt = now;
this._lastAcknowledgedLsn = this._latestCommitEndLsn;
this.logger.debug("Acknowledging transaction", {
this.logger.debug("acknowledge_latest_transaction", {
commitEndLsn: this._latestCommitEndLsn,
lastAcknowledgedAt: this._lastAcknowledgedAt,
});
@@ -747,7 +755,7 @@ export class ConcurrentFlushScheduler<T> {
const callback = this.config.callback;
const promise = this.concurrencyLimiter(async () => {
await startSpan(this._tracer, "flushNextBatch", async (span) => {
return await startSpan(this._tracer, "flushNextBatch", async (span) => {
const batchId = nanoid();
span.setAttribute("batch_id", batchId);
@@ -756,26 +764,47 @@ export class ConcurrentFlushScheduler<T> {
span.setAttribute("concurrency_pending_count", this.concurrencyLimiter.pendingCount);
span.setAttribute("concurrency_concurrency", this.concurrencyLimiter.concurrency);
this.logger.debug("flush_next_batch", {
batchId,
batchSize: batch.length,
concurrencyActiveCount: this.concurrencyLimiter.activeCount,
concurrencyPendingCount: this.concurrencyLimiter.pendingCount,
concurrencyConcurrency: this.concurrencyLimiter.concurrency,
});
const start = performance.now();
await callback(batchId, batch);
const end = performance.now();
const duration = end - start;
return {
batchId,
duration,
};
});
});
const [error] = await tryCatch(promise);
const [error, result] = await tryCatch(promise);
if (error) {
this.logger.error("Error flushing batch", {
this.logger.error("flush_batch_error", {
error,
});
this.failedBatchCount++;
} else {
this.logger.debug("flush_batch_complete", {
totalBatches: 1,
successfulBatches: 1,
failedBatches: 0,
totalFailedBatches: this.failedBatchCount,
duration: result?.duration,
batchId: result?.batchId,
});
}
this.logger.debug("Batch flush complete", {
totalBatches: 1,
successfulBatches: 1,
failedBatches: 0,
totalFailedBatches: this.failedBatchCount,
});
}
}
apps/webapp/test/runsReplicationService.test.ts
+129
View File
@@ -1030,6 +1030,135 @@ describe("RunsReplicationService", () => {
}
);
containerTest(
"should handover leadership to a second service, and the second service should be able to extend the leader lock",
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-shutdown-handover",
});
// Service A
const runsReplicationServiceA = new RunsReplicationService({
clickhouse,
pgConnectionUrl: postgresContainer.getConnectionUri(),
serviceName: "runs-replication-shutdown-handover",
slotName: "task_runs_to_clickhouse_v1",
publicationName: "task_runs_to_clickhouse_v1_publication",
redisOptions,
maxFlushConcurrency: 1,
flushIntervalMs: 100,
flushBatchSize: 1,
leaderLockTimeoutMs: 5000,
leaderLockExtendIntervalMs: 1000,
leaderLockAcquireAdditionalTimeMs: 10_000,
ackIntervalSeconds: 5,
logger: new Logger("runs-replication-shutdown-handover-a", "debug"),
});
await runsReplicationServiceA.start();
// Service A
const runsReplicationServiceB = new RunsReplicationService({
clickhouse,
pgConnectionUrl: postgresContainer.getConnectionUri(),
serviceName: "runs-replication-shutdown-handover",
slotName: "task_runs_to_clickhouse_v1",
publicationName: "task_runs_to_clickhouse_v1_publication",
redisOptions,
maxFlushConcurrency: 1,
flushIntervalMs: 100,
flushBatchSize: 1,
leaderLockTimeoutMs: 5000,
leaderLockExtendIntervalMs: 1000,
leaderLockAcquireAdditionalTimeMs: 10_000,
ackIntervalSeconds: 5,
logger: new Logger("runs-replication-shutdown-handover-b", "debug"),
});
// Now we need to initiate starting the second service, and after 6 seconds, we need to shutdown the first service
await Promise.all([
setTimeout(6000).then(() => runsReplicationServiceA.stop()),
runsReplicationServiceB.start(),
]);
const organization = await prisma.organization.create({
data: {
title: "test",
slug: "test",
},
});
const project = await prisma.project.create({
data: {
name: "test",
slug: "test",
organizationId: organization.id,
externalRef: "test",
},
});
const runtimeEnvironment = await prisma.runtimeEnvironment.create({
data: {
slug: "test",
type: "DEVELOPMENT",
projectId: project.id,
organizationId: organization.id,
apiKey: "test",
pkApiKey: "test",
shortcode: "test",
},
});
// Now we insert a row into the table
const taskRun = await prisma.taskRun.create({
data: {
friendlyId: "run_1234",
taskIdentifier: "my-task",
payload: JSON.stringify({ foo: "bar" }),
traceId: "1234",
spanId: "1234",
queue: "test",
runtimeEnvironmentId: runtimeEnvironment.id,
projectId: project.id,
organizationId: organization.id,
environmentType: "DEVELOPMENT",
engine: "V2",
},
});
await setTimeout(10_000);
// Check that the row was replicated to clickhouse
const queryRuns = clickhouse.reader.query({
name: "runs-replication",
query: "SELECT * FROM trigger_dev.task_runs_v2",
schema: z.any(),
});
const [queryError, result] = await queryRuns({});
expect(queryError).toBeNull();
expect(result?.length).toBe(1);
expect(result?.[0]).toEqual(
expect.objectContaining({
run_id: taskRun.id,
friendly_id: taskRun.friendlyId,
task_identifier: taskRun.taskIdentifier,
environment_id: runtimeEnvironment.id,
project_id: project.id,
organization_id: organization.id,
environment_type: "DEVELOPMENT",
engine: "V2",
})
);
await runsReplicationServiceB.stop();
}
);
containerTest(
"should replicate all 1,000 TaskRuns inserted in bulk to ClickHouse",
async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
docker/scripts/entrypoint.sh
+9 -2
View File
@@ -27,6 +27,13 @@ cp internal-packages/database/prisma/schema.prisma apps/webapp/prisma/
cp node_modules/@prisma/engines/*.node apps/webapp/prisma/
cd /triggerdotdev/apps/webapp
# exec dumb-init pnpm run start:local
NODE_PATH='/triggerdotdev/node_modules/.pnpm/node_modules' exec dumb-init node --max-old-space-size=8192 ./build/server.js
# Decide how much old-space memory Node should get.
# Use $NODE_MAX_OLD_SPACE_SIZE if it’s set; otherwise fall back to 8192.
MAX_OLD_SPACE_SIZE="${NODE_MAX_OLD_SPACE_SIZE:-8192}"
echo "Setting max old space size to ${MAX_OLD_SPACE_SIZE}"
NODE_PATH='/triggerdotdev/node_modules/.pnpm/node_modules' exec dumb-init node --max-old-space-size=${MAX_OLD_SPACE_SIZE} ./build/server.js
internal-packages/clickhouse/src/client/client.ts
+12 -3
View File
@@ -15,14 +15,22 @@ import type {
ClickhouseWriter,
} from "./types.js";
import { generateErrorMessage } from "zod-error";
import { Logger } from "@trigger.dev/core/logger";
import { Logger, type LogLevel } from "@trigger.dev/core/logger";
import type { Agent as HttpAgent } from "http";
import type { Agent as HttpsAgent } from "https";
export type ClickhouseConfig = {
name: string;
url: string;
tracer?: Tracer;
keepAlive?: {
enabled?: boolean;
idleSocketTtl?: number;
};
httpAgent?: HttpAgent | HttpsAgent;
clickhouseSettings?: ClickHouseSettings;
logger?: Logger;
logLevel?: LogLevel;
};
export class ClickhouseClient implements ClickhouseReader, ClickhouseWriter {
@@ -33,11 +41,12 @@ export class ClickhouseClient implements ClickhouseReader, ClickhouseWriter {
constructor(config: ClickhouseConfig) {
this.name = config.name;
this.logger = config.logger ?? new Logger("ClickhouseClient", "debug");
this.logger = config.logger ?? new Logger("ClickhouseClient", config.logLevel ?? "info");
this.client = createClient({
url: config.url,
keep_alive: config.keepAlive,
http_agent: config.httpAgent,
clickhouse_settings: {
...config.clickhouseSettings,
output_format_json_quote_64bit_integers: 0,
internal-packages/clickhouse/src/index.ts
+27 -9
View File
@@ -3,29 +3,38 @@ import { ClickhouseClient } from "./client/client.js";
import { ClickhouseReader, ClickhouseWriter } from "./client/types.js";
import { NoopClient } from "./client/noop.js";
import { insertTaskRuns, insertRawTaskRunPayloads } from "./taskRuns.js";
import { Logger } from "@trigger.dev/core/logger";
import { Logger, type LogLevel } from "@trigger.dev/core/logger";
import type { Agent as HttpAgent } from "http";
import type { Agent as HttpsAgent } from "https";
export type * from "./taskRuns.js";
export type ClickhouseCommonConfig = {
keepAlive?: {
enabled?: boolean;
idleSocketTtl?: number;
};
httpAgent?: HttpAgent | HttpsAgent;
clickhouseSettings?: ClickHouseSettings;
logger?: Logger;
logLevel?: LogLevel;
};
export type ClickHouseConfig =
| {
| ({
name?: string;
url?: string;
writerUrl?: never;
readerUrl?: never;
clickhouseSettings?: ClickHouseSettings;
logger?: Logger;
}
| {
} & ClickhouseCommonConfig)
| ({
name?: never;
url?: never;
writerName?: string;
writerUrl: string;
readerName?: string;
readerUrl: string;
clickhouseSettings?: ClickHouseSettings;
logger?: Logger;
};
} & ClickhouseCommonConfig);
export class ClickHouse {
public readonly reader: ClickhouseReader;
@@ -47,6 +56,9 @@ export class ClickHouse {
url: config.url,
clickhouseSettings: config.clickhouseSettings,
logger: this.logger,
logLevel: config.logLevel,
keepAlive: config.keepAlive,
httpAgent: config.httpAgent,
});
this.reader = client;
this.writer = client;
@@ -58,12 +70,18 @@ export class ClickHouse {
url: config.readerUrl,
clickhouseSettings: config.clickhouseSettings,
logger: this.logger,
logLevel: config.logLevel,
keepAlive: config.keepAlive,
httpAgent: config.httpAgent,
});
this.writer = new ClickhouseClient({
name: config.writerName ?? "clickhouse-writer",
url: config.writerUrl,
clickhouseSettings: config.clickhouseSettings,
logger: this.logger,
logLevel: config.logLevel,
keepAlive: config.keepAlive,
httpAgent: config.httpAgent,
});
this._splitClients = true;
internal-packages/replication/src/client.ts
+74 -28
View File
@@ -52,16 +52,16 @@ export interface LogicalReplicationClientOptions {
*/
leaderLockExtendIntervalMs?: number;
/**
* The number of times to retry acquiring the leader lock (default: 120)
*/
leaderLockRetryCount?: number;
/**
* The interval in ms to retry acquiring the leader lock (default: 500)
*/
leaderLockRetryIntervalMs?: number;
/**
* The additional time in ms to retry acquiring the leader lock (default: 1000ms)
*/
leaderLockAcquireAdditionalTimeMs?: number;
/**
* The interval in seconds to automatically acknowledge the last LSN if no ack has been sent (default: 10)
*/
@@ -97,7 +97,7 @@ export class LogicalReplicationClient {
private lastAcknowledgedLsn: string | null = null;
private leaderLockTimeoutMs: number;
private leaderLockExtendIntervalMs: number;
private leaderLockRetryCount: number;
private leaderLockAcquireAdditionalTimeMs: number;
private leaderLockRetryIntervalMs: number;
private leaderLockHeartbeatTimer: NodeJS.Timeout | null = null;
private ackIntervalSeconds: number;
@@ -124,7 +124,7 @@ export class LogicalReplicationClient {
this.leaderLockTimeoutMs = options.leaderLockTimeoutMs ?? 30000;
this.leaderLockExtendIntervalMs = options.leaderLockExtendIntervalMs ?? 10000;
this.leaderLockRetryCount = options.leaderLockRetryCount ?? 120;
this.leaderLockAcquireAdditionalTimeMs = options.leaderLockAcquireAdditionalTimeMs ?? 1000;
this.leaderLockRetryIntervalMs = options.leaderLockRetryIntervalMs ?? 500;
this.ackIntervalSeconds = options.ackIntervalSeconds ?? 10;
@@ -578,34 +578,74 @@ export class LogicalReplicationClient {
}
async #acquireLeaderLock(): Promise<boolean> {
try {
this.leaderLock = await this.redlock.acquire(
[`logical-replication-client:${this.options.name}`],
this.leaderLockTimeoutMs,
{
retryCount: this.leaderLockRetryCount,
retryDelay: this.leaderLockRetryIntervalMs,
}
);
} catch (err) {
this.logger.error("Leader election failed", {
name: this.options.name,
table: this.options.table,
slotName: this.options.slotName,
publicationName: this.options.publicationName,
retryCount: this.leaderLockRetryCount,
retryIntervalMs: this.leaderLockRetryIntervalMs,
error: err,
});
const startTime = Date.now();
const maxWaitTime = this.leaderLockTimeoutMs + this.leaderLockAcquireAdditionalTimeMs;
return false;
this.logger.debug("Acquiring leader lock", {
name: this.options.name,
slotName: this.options.slotName,
publicationName: this.options.publicationName,
maxWaitTime,
});
let attempt = 0;
while (Date.now() - startTime < maxWaitTime) {
try {
this.leaderLock = await this.redlock.acquire(
[`logical-replication-client:${this.options.name}`],
this.leaderLockTimeoutMs
);
this.logger.debug("Acquired leader lock", {
name: this.options.name,
slotName: this.options.slotName,
publicationName: this.options.publicationName,
lockTimeoutMs: this.leaderLockTimeoutMs,
lockExtendIntervalMs: this.leaderLockExtendIntervalMs,
lock: this.leaderLock,
attempt,
});
return true;
} catch (err) {
attempt++;
this.logger.debug("Failed to acquire leader lock, retrying", {
name: this.options.name,
slotName: this.options.slotName,
publicationName: this.options.publicationName,
attempt,
retryIntervalMs: this.leaderLockRetryIntervalMs,
error: err,
});
await new Promise((resolve) => setTimeout(resolve, this.leaderLockRetryIntervalMs));
}
}
return true;
this.logger.error("Leader election failed after retries", {
name: this.options.name,
table: this.options.table,
slotName: this.options.slotName,
publicationName: this.options.publicationName,
totalAttempts: attempt,
totalWaitTimeMs: Date.now() - startTime,
});
return false;
}
async #releaseLeaderLock() {
if (!this.leaderLock) return;
this.logger.debug("Releasing leader lock", {
name: this.options.name,
slotName: this.options.slotName,
publicationName: this.options.publicationName,
lockTimeoutMs: this.leaderLockTimeoutMs,
lockExtendIntervalMs: this.leaderLockExtendIntervalMs,
lock: this.leaderLock,
});
const [releaseError] = await tryCatch(this.leaderLock.release());
this.leaderLock = null;
@@ -631,6 +671,9 @@ export class LogicalReplicationClient {
name: this.options.name,
slotName: this.options.slotName,
publicationName: this.options.publicationName,
lockTimeoutMs: this.leaderLockTimeoutMs,
lockExtendIntervalMs: this.leaderLockExtendIntervalMs,
lock: this.leaderLock,
});
} catch (err) {
this.logger.error("Failed to extend leader lock", {
@@ -638,6 +681,9 @@ export class LogicalReplicationClient {
slotName: this.options.slotName,
publicationName: this.options.publicationName,
error: err,
lockTimeoutMs: this.leaderLockTimeoutMs,
lockExtendIntervalMs: this.leaderLockExtendIntervalMs,
lock: this.leaderLock,
});
// Optionally emit an error or handle loss of leadership
this.events.emit("error", err instanceof Error ? err : new Error(String(err)));