Eric Allam
65da20c225
* WIP clickhouse package with test containers setup * More clickhouse client setup now with otel and real tests, and the v1 of raw run events * Add some additional columns to raw_run_events_v1 * WIP runs dashboard service * Create a new run engine event bus event for the runs dashboard to hook into * Track run events in the run engine * make sure engine v1 runs get synced to CH * Update the attemptNumber of v3 task runs * Restructure the run events to be more sparse * emit more stuff * Setup replication package * scaffold the replication package * replication wip * resolve conflicts * more replication stuff * Add ability to drop the replication slot completely on teardown * Use the new single replacingmergetree task events table for replication * get it working * insert payloads into their own table only on insert and then join * prepare for using clickhouse cloud and now running ch migrations during boot in the entrypoint.sh * Handover WIP and tests * Testing the replication service * Remove the runs dashboard stuff that we aren't using anymore * Added a test for large payloads * hacky typecheck fix * Fix new internal package typecheck issues and start adding telemetry to the replication service * tracing over spans, some other improvements * Improvements to the runs replication service, now ready for testing * Some fixes and cleanups * Don't need this code anymore * move transaction types into the runs replication service * only send spans where there are transaction events * A couple of suggested tweaks
110 lines
2.9 KiB
TypeScript
110 lines
2.9 KiB
TypeScript
import { parseNaturalLanguageDuration } from "@trigger.dev/core/v3/isomorphic";
|
|
import { logger } from "~/services/logger.server";
|
|
import { workerQueue } from "~/services/worker.server";
|
|
import { commonWorker } from "../commonWorker.server";
|
|
import { BaseService } from "./baseService.server";
|
|
import { enqueueRun } from "./enqueueRun.server";
|
|
import { ExpireEnqueuedRunService } from "./expireEnqueuedRun.server";
|
|
|
|
export class EnqueueDelayedRunService extends BaseService {
|
|
public static async enqueue(runId: string, runAt?: Date) {
|
|
await commonWorker.enqueue({
|
|
job: "v3.enqueueDelayedRun",
|
|
payload: { runId },
|
|
availableAt: runAt,
|
|
id: `v3.enqueueDelayed:${runId}`,
|
|
});
|
|
}
|
|
|
|
public static async reschedule(runId: string, runAt?: Date) {
|
|
// We have to do this for now because it's possible that the workerQueue
|
|
// was used when the run was first delayed, and EnqueueDelayedRunService.reschedule
|
|
// is called from RescheduleTaskRunService, which allows the runAt to be changed
|
|
// so if we don't dequeue the old job, we might end up with multiple jobs
|
|
await workerQueue.dequeue(`v3.enqueueDelayedRun.${runId}`);
|
|
|
|
await commonWorker.enqueue({
|
|
job: "v3.enqueueDelayedRun",
|
|
payload: { runId },
|
|
availableAt: runAt,
|
|
id: `v3.enqueueDelayed:${runId}`,
|
|
});
|
|
}
|
|
|
|
public async call(runId: string) {
|
|
const run = await this._prisma.taskRun.findFirst({
|
|
where: {
|
|
id: runId,
|
|
},
|
|
include: {
|
|
runtimeEnvironment: {
|
|
include: {
|
|
organization: true,
|
|
project: true,
|
|
},
|
|
},
|
|
dependency: {
|
|
include: {
|
|
dependentBatchRun: {
|
|
include: {
|
|
dependentTaskAttempt: {
|
|
include: {
|
|
taskRun: true,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
dependentAttempt: {
|
|
include: {
|
|
taskRun: true,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
});
|
|
|
|
if (!run) {
|
|
logger.debug("Could not find delayed run to enqueue", {
|
|
runId,
|
|
});
|
|
|
|
return;
|
|
}
|
|
|
|
if (run.status !== "DELAYED") {
|
|
logger.debug("Delayed run cannot be enqueued because it's not in DELAYED status", {
|
|
run,
|
|
});
|
|
|
|
return;
|
|
}
|
|
|
|
await this._prisma.taskRun.update({
|
|
where: {
|
|
id: run.id,
|
|
},
|
|
data: {
|
|
status: "PENDING",
|
|
queuedAt: new Date(),
|
|
},
|
|
});
|
|
|
|
if (run.ttl) {
|
|
const expireAt = parseNaturalLanguageDuration(run.ttl);
|
|
|
|
if (expireAt) {
|
|
await ExpireEnqueuedRunService.enqueue(run.id, expireAt);
|
|
}
|
|
}
|
|
|
|
await enqueueRun({
|
|
env: run.runtimeEnvironment,
|
|
run: run,
|
|
dependentRun:
|
|
run.dependency?.dependentAttempt?.taskRun ??
|
|
run.dependency?.dependentBatchRun?.dependentTaskAttempt?.taskRun,
|
|
});
|
|
}
|
|
}
|