Daniel Sutton
65c545da4e
## Summary Adds read methods to `RunStore` (`findRun`, `findRunOrThrow`, `findRuns`) and routes every Postgres read of `TaskRun` through them, mirroring how writes already go through the store. Behavior-preserving: each relocated read keeps its exact query, field selection, and database client (writer, replica, or transaction). This lets `TaskRun` reads be retargeted to a different backing store later without touching call sites. Stacked on #3981 (the write adapter); that PR is the base of this one. ## Scope In scope: the run engine, webapp services, presenters, and route loaders. Three reads that pulled `TaskRun` in through a parent model's relation `include` (alert delivery, batch results, attempt-dependency cancellation) are decomposed to fetch the run(s) through the store and stitch them back, since a relation include would not follow `TaskRun` to a new table. Left reading the existing table (out of scope): the legacy MarQS paths, the legacy trigger idempotency read, and one raw-SQL recovery script (commented for revisiting at cutover). ## Notes Reads default to the read replica; callers pass the writer or a transaction client wherever the original read did, so writer-vs-replica behavior is unchanged.
102 lines
2.7 KiB
TypeScript
102 lines
2.7 KiB
TypeScript
import { Tracer } from "@opentelemetry/api";
|
|
import type { PrismaClientOrTransaction } from "@trigger.dev/database";
|
|
import { RunsReplicationService } from "~/services/runsReplicationService.server";
|
|
import { runStore } from "~/v3/runStore.server";
|
|
import { startSpan } from "~/v3/tracing.server";
|
|
import { FINAL_RUN_STATUSES } from "../v3/taskStatus";
|
|
import { Logger } from "@trigger.dev/core/logger";
|
|
|
|
export class RunsBackfillerService {
|
|
private readonly prisma: PrismaClientOrTransaction;
|
|
private readonly runsReplicationInstance: RunsReplicationService;
|
|
private readonly tracer: Tracer;
|
|
private readonly logger: Logger;
|
|
|
|
constructor(opts: {
|
|
prisma: PrismaClientOrTransaction;
|
|
runsReplicationInstance: RunsReplicationService;
|
|
tracer: Tracer;
|
|
logLevel?: "log" | "error" | "warn" | "info" | "debug";
|
|
}) {
|
|
this.prisma = opts.prisma;
|
|
this.runsReplicationInstance = opts.runsReplicationInstance;
|
|
this.tracer = opts.tracer;
|
|
this.logger = new Logger("RunsBackfillerService", opts.logLevel ?? "debug");
|
|
}
|
|
|
|
public async call({
|
|
from,
|
|
to,
|
|
cursor,
|
|
batchSize,
|
|
}: {
|
|
from: Date;
|
|
to: Date;
|
|
cursor?: string;
|
|
batchSize?: number;
|
|
}): Promise<string | undefined> {
|
|
return await startSpan(this.tracer, "RunsBackfillerService.call()", async (span) => {
|
|
span.setAttribute("from", from.toISOString());
|
|
span.setAttribute("to", to.toISOString());
|
|
span.setAttribute("cursor", cursor ?? "");
|
|
span.setAttribute("batchSize", batchSize ?? 0);
|
|
|
|
const runs = await runStore.findRuns(
|
|
{
|
|
where: {
|
|
createdAt: {
|
|
gte: from,
|
|
lte: to,
|
|
},
|
|
status: {
|
|
in: FINAL_RUN_STATUSES,
|
|
},
|
|
...(cursor ? { id: { gt: cursor } } : {}),
|
|
},
|
|
orderBy: {
|
|
id: "asc",
|
|
},
|
|
take: batchSize,
|
|
},
|
|
this.prisma
|
|
);
|
|
|
|
if (runs.length === 0) {
|
|
this.logger.info("No runs to backfill", { from, to, cursor });
|
|
|
|
return;
|
|
}
|
|
|
|
this.logger.info("Backfilling runs", {
|
|
from,
|
|
to,
|
|
cursor,
|
|
batchSize,
|
|
runCount: runs.length,
|
|
firstCreatedAt: runs[0].createdAt,
|
|
lastCreatedAt: runs[runs.length - 1].createdAt,
|
|
});
|
|
|
|
await this.runsReplicationInstance.backfill(
|
|
runs.map((run) => ({
|
|
...run,
|
|
masterQueue: run.workerQueue,
|
|
}))
|
|
);
|
|
|
|
const lastRun = runs[runs.length - 1];
|
|
|
|
this.logger.info("Backfilled runs", {
|
|
from,
|
|
to,
|
|
cursor,
|
|
batchSize,
|
|
lastRunId: lastRun.id,
|
|
});
|
|
|
|
// Return the last run ID to continue from
|
|
return lastRun.id;
|
|
});
|
|
}
|
|
}
|