feat(run-ops): activation — drop cross-DB FKs, provision run-ops DB, enable split (#4124)
This commit is contained in:
Daniel Sutton
@@ -0,0 +1,6 @@
|
||||
---
|
||||
area: webapp
|
||||
type: feature
|
||||
---
|
||||
|
||||
Enable the dedicated run-ops database split: run records and their related rows are served from a separate database, with cross-database references resolved in application code instead of database foreign keys.
|
||||
@@ -10,7 +10,6 @@ export class PerformBulkActionService extends BaseService {
|
||||
const item = await this._prisma.bulkActionItem.findFirst({
|
||||
where: { id: bulkActionItemId },
|
||||
include: {
|
||||
group: true,
|
||||
sourceRun: true,
|
||||
destinationRun: true,
|
||||
},
|
||||
@@ -24,7 +23,7 @@ export class PerformBulkActionService extends BaseService {
|
||||
return;
|
||||
}
|
||||
|
||||
switch (item.group.type) {
|
||||
switch (item.type) {
|
||||
case "REPLAY": {
|
||||
const service = new ReplayTaskRunService(this._prisma);
|
||||
const result = await service.call(item.sourceRun, { triggerSource: "dashboard" });
|
||||
@@ -57,7 +56,7 @@ export class PerformBulkActionService extends BaseService {
|
||||
break;
|
||||
}
|
||||
default: {
|
||||
assertNever(item.group.type);
|
||||
assertNever(item.type);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -94,17 +93,20 @@ export class PerformBulkActionService extends BaseService {
|
||||
|
||||
public async call(bulkActionGroupId: string) {
|
||||
const actionGroup = await this._prisma.bulkActionGroup.findFirst({
|
||||
include: {
|
||||
items: true,
|
||||
},
|
||||
where: { id: bulkActionGroupId },
|
||||
select: { id: true },
|
||||
});
|
||||
|
||||
if (!actionGroup) {
|
||||
return;
|
||||
}
|
||||
|
||||
for (const item of actionGroup.items) {
|
||||
const items = await this._prisma.bulkActionItem.findMany({
|
||||
where: { groupId: bulkActionGroupId },
|
||||
select: { id: true },
|
||||
});
|
||||
|
||||
for (const item of items) {
|
||||
await this.enqueueBulkActionItem(item.id, bulkActionGroupId);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,169 @@
|
||||
// Proof for dropping the canonical BatchTaskRun -> RuntimeEnvironment FK
|
||||
// (constraint "BatchTaskRun_runtimeEnvironmentId_fkey", onDelete: Cascade) while keeping the
|
||||
// runtimeEnvironmentId scalar and its compound @@unique/@@index. BatchTaskRun is run-ops and
|
||||
// RuntimeEnvironment is control-plane, so the two may live on different servers; create-time
|
||||
// integrity is preserved app-side via the ControlPlaneResolver's assertEnvExists. Env-delete
|
||||
// orphan cleanup is handled separately — here the batch row is tolerated.
|
||||
|
||||
import { heteroPostgresTest, postgresTest } from "@internal/testcontainers";
|
||||
import type { PrismaClient } from "@trigger.dev/database";
|
||||
import { describe, expect, vi } from "vitest";
|
||||
import { ControlPlaneCache } from "~/v3/runOpsMigration/controlPlaneCache.server";
|
||||
import {
|
||||
ControlPlaneReferenceError,
|
||||
ControlPlaneResolver,
|
||||
} from "~/v3/runOpsMigration/controlPlaneResolver.server";
|
||||
|
||||
// Cross-DB testcontainer spin-up + queries can exceed the 5s default on the first test.
|
||||
vi.setConfig({ testTimeout: 60_000 });
|
||||
|
||||
let seedCounter = 0;
|
||||
|
||||
async function seedEnvironment(prisma: PrismaClient) {
|
||||
const n = seedCounter++;
|
||||
const organization = await prisma.organization.create({
|
||||
data: { title: `Org ${n}`, slug: `org-${n}` },
|
||||
});
|
||||
const project = await prisma.project.create({
|
||||
data: {
|
||||
name: `Project ${n}`,
|
||||
slug: `project-${n}`,
|
||||
externalRef: `proj_${n}`,
|
||||
organizationId: organization.id,
|
||||
},
|
||||
});
|
||||
const environment = await prisma.runtimeEnvironment.create({
|
||||
data: {
|
||||
type: "PRODUCTION",
|
||||
slug: `env-${n}`,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
apiKey: `tr_prod_${n}`,
|
||||
pkApiKey: `pk_prod_${n}`,
|
||||
shortcode: `short_${n}`,
|
||||
},
|
||||
});
|
||||
return { organization, project, environment };
|
||||
}
|
||||
|
||||
let batchCounter = 0;
|
||||
|
||||
async function createBatch(prisma: PrismaClient, runtimeEnvironmentId: string) {
|
||||
const n = batchCounter++;
|
||||
return prisma.batchTaskRun.create({
|
||||
data: {
|
||||
friendlyId: `batch_${n}`,
|
||||
runtimeEnvironmentId,
|
||||
runCount: 1,
|
||||
runIds: [],
|
||||
batchVersion: "runengine:v2",
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
// Asserts the post-migration state of BatchTaskRun on a given client: the FK is gone, but the
|
||||
// scalar and both compound constraints are retained. Shared by the single-version and the
|
||||
// cross-version suites.
|
||||
async function assertSchemaState(prisma: PrismaClient) {
|
||||
const foreignKeys = await prisma.$queryRaw<{ constraint_name: string }[]>`
|
||||
SELECT constraint_name
|
||||
FROM information_schema.table_constraints
|
||||
WHERE table_name = 'BatchTaskRun'
|
||||
AND constraint_type = 'FOREIGN KEY'
|
||||
`;
|
||||
expect(foreignKeys.map((c) => c.constraint_name)).not.toContain(
|
||||
"BatchTaskRun_runtimeEnvironmentId_fkey"
|
||||
);
|
||||
|
||||
const columns = await prisma.$queryRaw<{ column_name: string }[]>`
|
||||
SELECT column_name
|
||||
FROM information_schema.columns
|
||||
WHERE table_name = 'BatchTaskRun'
|
||||
AND column_name = 'runtimeEnvironmentId'
|
||||
`;
|
||||
expect(columns).toHaveLength(1);
|
||||
|
||||
// The @@unique([runtimeEnvironmentId, idempotencyKey]) and
|
||||
// @@index([runtimeEnvironmentId, id(sort: Desc)]) both survive the FK drop.
|
||||
const indexes = await prisma.$queryRaw<{ indexdef: string }[]>`
|
||||
SELECT indexdef FROM pg_indexes WHERE tablename = 'BatchTaskRun'
|
||||
`;
|
||||
const defs = indexes.map((i) => i.indexdef);
|
||||
const hasUnique = defs.some(
|
||||
(d) => /UNIQUE/i.test(d) && d.includes("runtimeEnvironmentId") && d.includes("idempotencyKey")
|
||||
);
|
||||
const hasIndex = defs.some(
|
||||
(d) => !/UNIQUE/i.test(d) && d.includes("runtimeEnvironmentId") && /\bid\b/.test(d)
|
||||
);
|
||||
expect(hasUnique).toBe(true);
|
||||
expect(hasIndex).toBe(true);
|
||||
}
|
||||
|
||||
// Inserts an env + batch, deletes the env, and asserts the batch survives (cascade gone).
|
||||
async function assertOrphanTolerated(prisma: PrismaClient) {
|
||||
const { environment } = await seedEnvironment(prisma);
|
||||
const batch = await createBatch(prisma, environment.id);
|
||||
|
||||
await prisma.runtimeEnvironment.delete({ where: { id: environment.id } });
|
||||
|
||||
const survivor = await prisma.batchTaskRun.findFirst({ where: { id: batch.id } });
|
||||
expect(survivor).not.toBeNull();
|
||||
expect(survivor?.runtimeEnvironmentId).toBe(environment.id);
|
||||
}
|
||||
|
||||
describe("drop BatchTaskRun -> RuntimeEnvironment FK", () => {
|
||||
postgresTest("FK constraint absent; scalar + unique + index retained", async ({ prisma }) => {
|
||||
await assertSchemaState(prisma);
|
||||
});
|
||||
|
||||
postgresTest(
|
||||
"deleting the env leaves the BatchTaskRun row alive (no cascade; orphan cleanup handled separately)",
|
||||
async ({ prisma }) => {
|
||||
await assertOrphanTolerated(prisma);
|
||||
}
|
||||
);
|
||||
|
||||
postgresTest(
|
||||
"app-side env validation: assertEnvExists rejects an invalid env and a valid-env create succeeds by scalar",
|
||||
async ({ prisma }) => {
|
||||
const { environment } = await seedEnvironment(prisma);
|
||||
|
||||
const resolver = new ControlPlaneResolver({
|
||||
controlPlanePrimary: prisma,
|
||||
controlPlaneReplica: prisma,
|
||||
cache: new ControlPlaneCache(),
|
||||
splitEnabled: () => true,
|
||||
});
|
||||
|
||||
// The exact guard call the create services place before batchTaskRun.create.
|
||||
await expect(resolver.assertEnvExists("env_does_not_exist")).rejects.toBeInstanceOf(
|
||||
ControlPlaneReferenceError
|
||||
);
|
||||
|
||||
await expect(resolver.assertEnvExists(environment.id)).resolves.toBeUndefined();
|
||||
|
||||
// Once the guard passes, the batch is linked by the runtimeEnvironmentId scalar (no FK).
|
||||
const batch = await createBatch(prisma, environment.id);
|
||||
expect(batch.runtimeEnvironmentId).toBe(environment.id);
|
||||
}
|
||||
);
|
||||
});
|
||||
|
||||
// Cross-version gate: the migration applies and the post-state is identical across major versions.
|
||||
describe("drop BatchTaskRun -> RuntimeEnvironment FK — cross-version (legacy + new Postgres)", () => {
|
||||
heteroPostgresTest(
|
||||
"migration applies and FK is absent on both the legacy and new databases",
|
||||
async ({ prisma14, prisma17 }) => {
|
||||
await assertSchemaState(prisma14);
|
||||
await assertSchemaState(prisma17);
|
||||
}
|
||||
);
|
||||
|
||||
heteroPostgresTest(
|
||||
"env delete leaves the batch orphaned on both the legacy and new databases",
|
||||
async ({ prisma14, prisma17 }) => {
|
||||
await assertOrphanTolerated(prisma14);
|
||||
await assertOrphanTolerated(prisma17);
|
||||
}
|
||||
);
|
||||
});
|
||||
@@ -0,0 +1,64 @@
|
||||
// Single-version proof for dropping the dead `_TaskRunToTaskRunTag` implicit join.
|
||||
|
||||
import { describe, expect } from "vitest";
|
||||
import { postgresTest } from "@internal/testcontainers";
|
||||
|
||||
describe("drop _TaskRunToTaskRunTag implicit join", () => {
|
||||
postgresTest("runTags scalar round-trips and the join table is gone", async ({ prisma }) => {
|
||||
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",
|
||||
},
|
||||
});
|
||||
|
||||
const taskRun = await prisma.taskRun.create({
|
||||
data: {
|
||||
friendlyId: "run_1234",
|
||||
taskIdentifier: "my-task",
|
||||
payload: JSON.stringify({ foo: "bar" }),
|
||||
payloadType: "application/json",
|
||||
traceId: "1234",
|
||||
spanId: "1234",
|
||||
queue: "test",
|
||||
runtimeEnvironmentId: runtimeEnvironment.id,
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
environmentType: "DEVELOPMENT",
|
||||
engine: "V2",
|
||||
runTags: ["alpha", "beta"],
|
||||
},
|
||||
});
|
||||
|
||||
const readBack = await prisma.taskRun.findFirstOrThrow({
|
||||
where: { id: taskRun.id },
|
||||
});
|
||||
expect(readBack.runTags).toEqual(["alpha", "beta"]);
|
||||
|
||||
const result = await prisma.$queryRaw<{ t: string | null }[]>`
|
||||
SELECT to_regclass('public._TaskRunToTaskRunTag')::text as t
|
||||
`;
|
||||
expect(result[0].t).toBeNull();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,660 @@
|
||||
import { heteroPostgresTest, heteroRunOpsPostgresTest } from "@internal/testcontainers";
|
||||
import type { PrismaClient } from "@trigger.dev/database";
|
||||
import type { RunOpsPrismaClient } from "@internal/run-ops-database";
|
||||
import { describe, expect, vi } from "vitest";
|
||||
import { RunOpsCascadeCleanupService } from "~/v3/runOpsMigration/runOpsCascadeCleanup.server";
|
||||
|
||||
// Cross-DB testcontainer spin-up + the multi-table seed can exceed the 5s default.
|
||||
vi.setConfig({ testTimeout: 120_000 });
|
||||
|
||||
// Run-subgraph tables that live in BOTH the control-plane schema AND the dedicated run-ops SUBSET
|
||||
// schema, so they are deleted on every run-ops writer.
|
||||
const SUBGRAPH_TABLES = [
|
||||
"taskRun",
|
||||
"taskRunAttempt",
|
||||
"waitpoint",
|
||||
"taskRunWaitpoint",
|
||||
"taskRunCheckpoint",
|
||||
"checkpoint",
|
||||
"checkpointRestoreEvent",
|
||||
"batchTaskRun",
|
||||
] as const;
|
||||
|
||||
type SubgraphTable = (typeof SUBGRAPH_TABLES)[number];
|
||||
|
||||
let seedCounter = 0;
|
||||
|
||||
/**
|
||||
* The cross-seam (run-ops -> control-plane) Cascade FKs that the cloud DB physically drops. Applied
|
||||
* to the FK-dropped fixture to model cloud; the other side keeps them to model self-host. Only the
|
||||
* run-subgraph constraints exist on the dedicated run-ops schema; BulkActionItem's are control-plane
|
||||
* only and are dropped separately on a full-schema client.
|
||||
*/
|
||||
const SUBGRAPH_CROSS_SEAM_FKS: Array<{ table: string; constraint: string }> = [
|
||||
{ table: "TaskRun", constraint: "TaskRun_runtimeEnvironmentId_fkey" },
|
||||
{ table: "TaskRun", constraint: "TaskRun_projectId_fkey" },
|
||||
{ table: "TaskRunAttempt", constraint: "TaskRunAttempt_runtimeEnvironmentId_fkey" },
|
||||
{ table: "Waitpoint", constraint: "Waitpoint_environmentId_fkey" },
|
||||
{ table: "Waitpoint", constraint: "Waitpoint_projectId_fkey" },
|
||||
{ table: "TaskRunWaitpoint", constraint: "TaskRunWaitpoint_projectId_fkey" },
|
||||
{ table: "TaskRunCheckpoint", constraint: "TaskRunCheckpoint_runtimeEnvironmentId_fkey" },
|
||||
{ table: "TaskRunCheckpoint", constraint: "TaskRunCheckpoint_projectId_fkey" },
|
||||
{ table: "Checkpoint", constraint: "Checkpoint_runtimeEnvironmentId_fkey" },
|
||||
{ table: "Checkpoint", constraint: "Checkpoint_projectId_fkey" },
|
||||
{
|
||||
table: "CheckpointRestoreEvent",
|
||||
constraint: "CheckpointRestoreEvent_runtimeEnvironmentId_fkey",
|
||||
},
|
||||
{ table: "CheckpointRestoreEvent", constraint: "CheckpointRestoreEvent_projectId_fkey" },
|
||||
{ table: "BatchTaskRun", constraint: "BatchTaskRun_runtimeEnvironmentId_fkey" },
|
||||
];
|
||||
|
||||
const BULK_ACTION_CROSS_SEAM_FKS: Array<{ table: string; constraint: string }> = [
|
||||
{ table: "BulkActionItem", constraint: "BulkActionItem_sourceRunId_fkey" },
|
||||
{ table: "BulkActionItem", constraint: "BulkActionItem_destinationRunId_fkey" },
|
||||
];
|
||||
|
||||
async function dropCrossSeamFks(
|
||||
prisma: { $executeRawUnsafe: (q: string) => Promise<unknown> },
|
||||
fks: Array<{ table: string; constraint: string }>
|
||||
) {
|
||||
for (const { table, constraint } of fks) {
|
||||
await prisma.$executeRawUnsafe(
|
||||
`ALTER TABLE "${table}" DROP CONSTRAINT IF EXISTS "${constraint}"`
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
type Scope = { projectId: string; environmentId: string; organizationId: string };
|
||||
type FullScope = Scope & { workerTaskId: string; queueId: string; backgroundWorkerId: string };
|
||||
|
||||
// Minimal structural client covering the control-plane prerequisites + run-subgraph models the
|
||||
// seed/count helpers touch. Both PrismaClient and RunOpsPrismaClient are assignable.
|
||||
type SeedClient = {
|
||||
organization: any;
|
||||
project: any;
|
||||
runtimeEnvironment: any;
|
||||
backgroundWorker: any;
|
||||
backgroundWorkerTask: any;
|
||||
taskQueue: any;
|
||||
taskRun: any;
|
||||
taskRunAttempt: any;
|
||||
waitpoint: any;
|
||||
taskRunWaitpoint: any;
|
||||
taskRunCheckpoint: any;
|
||||
checkpoint: any;
|
||||
checkpointRestoreEvent: any;
|
||||
batchTaskRun: any;
|
||||
};
|
||||
|
||||
// Synthetic scope for the dedicated run-ops subset client, whose schema scalarizes every
|
||||
// control-plane FK so no org/project/env rows are required.
|
||||
function makeSyntheticScope(): FullScope {
|
||||
const n = seedCounter++;
|
||||
return {
|
||||
projectId: `proj_synthetic_${n}`,
|
||||
environmentId: `env_synthetic_${n}`,
|
||||
organizationId: `org_synthetic_${n}`,
|
||||
workerTaskId: `task_synthetic_${n}`,
|
||||
queueId: `queue_synthetic_${n}`,
|
||||
backgroundWorkerId: `worker_synthetic_${n}`,
|
||||
};
|
||||
}
|
||||
|
||||
/** Create the control-plane prerequisites (org, project, env, worker, task, queue). */
|
||||
async function seedScope(prisma: SeedClient): Promise<FullScope> {
|
||||
const n = seedCounter++;
|
||||
const org = await prisma.organization.create({
|
||||
data: { title: `Org ${n}`, slug: `org-${n}` },
|
||||
});
|
||||
const project = await prisma.project.create({
|
||||
data: {
|
||||
name: `Project ${n}`,
|
||||
slug: `project-${n}`,
|
||||
externalRef: `proj_${n}`,
|
||||
organizationId: org.id,
|
||||
},
|
||||
});
|
||||
const environment = await prisma.runtimeEnvironment.create({
|
||||
data: {
|
||||
type: "PRODUCTION",
|
||||
slug: `env-${n}`,
|
||||
projectId: project.id,
|
||||
organizationId: org.id,
|
||||
apiKey: `tr_prod_${n}`,
|
||||
pkApiKey: `pk_prod_${n}`,
|
||||
shortcode: `short_${n}`,
|
||||
},
|
||||
});
|
||||
const worker = await prisma.backgroundWorker.create({
|
||||
data: {
|
||||
friendlyId: `worker_${n}`,
|
||||
contentHash: `hash_${n}`,
|
||||
projectId: project.id,
|
||||
runtimeEnvironmentId: environment.id,
|
||||
version: `2024.1.${n}`,
|
||||
metadata: {},
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
const task = await prisma.backgroundWorkerTask.create({
|
||||
data: {
|
||||
friendlyId: `task_${n}`,
|
||||
slug: `my-task-${n}`,
|
||||
filePath: "index.ts",
|
||||
exportName: "myTask",
|
||||
workerId: worker.id,
|
||||
runtimeEnvironmentId: environment.id,
|
||||
projectId: project.id,
|
||||
},
|
||||
});
|
||||
const queue = await prisma.taskQueue.create({
|
||||
data: {
|
||||
friendlyId: `queue_${n}`,
|
||||
name: `task/my-task-${n}`,
|
||||
runtimeEnvironmentId: environment.id,
|
||||
projectId: project.id,
|
||||
},
|
||||
});
|
||||
return {
|
||||
projectId: project.id,
|
||||
environmentId: environment.id,
|
||||
organizationId: org.id,
|
||||
workerTaskId: task.id,
|
||||
queueId: queue.id,
|
||||
backgroundWorkerId: worker.id,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Seed one full run-ops subgraph for a scope: a TaskRun tree (root + child), an attempt, a
|
||||
* Waitpoint with a blocking edge (TaskRunWaitpoint), a TaskRunCheckpoint, a Checkpoint + a
|
||||
* CheckpointRestoreEvent, and a BatchTaskRun with a member run. Returns the source + destination
|
||||
* runs so a caller with a control-plane client can attach a BulkActionItem.
|
||||
*/
|
||||
async function seedRunOpsSubgraph(
|
||||
prisma: SeedClient,
|
||||
scope: Scope & { backgroundWorkerId: string; workerTaskId: string; queueId: string }
|
||||
): Promise<{ sourceRunId: string; destinationRunId: string }> {
|
||||
const n = seedCounter++;
|
||||
const { projectId, environmentId } = scope;
|
||||
|
||||
const baseRun = (suffix: string) => ({
|
||||
friendlyId: `run_${n}_${suffix}`,
|
||||
taskIdentifier: `my-task-${n}`,
|
||||
payload: "{}",
|
||||
payloadType: "application/json",
|
||||
traceId: `trace_${n}_${suffix}`,
|
||||
spanId: `span_${n}_${suffix}`,
|
||||
queue: `task/my-task-${n}`,
|
||||
runtimeEnvironmentId: environmentId,
|
||||
projectId,
|
||||
});
|
||||
|
||||
const rootRun = await prisma.taskRun.create({ data: baseRun("root") });
|
||||
const childRun = await prisma.taskRun.create({
|
||||
data: { ...baseRun("child"), parentTaskRunId: rootRun.id, rootTaskRunId: rootRun.id },
|
||||
});
|
||||
|
||||
const attempt = await prisma.taskRunAttempt.create({
|
||||
data: {
|
||||
friendlyId: `attempt_${n}`,
|
||||
taskRunId: rootRun.id,
|
||||
backgroundWorkerId: scope.backgroundWorkerId,
|
||||
backgroundWorkerTaskId: scope.workerTaskId,
|
||||
runtimeEnvironmentId: environmentId,
|
||||
queueId: scope.queueId,
|
||||
},
|
||||
});
|
||||
|
||||
const waitpoint = await prisma.waitpoint.create({
|
||||
data: {
|
||||
friendlyId: `wp_${n}`,
|
||||
type: "MANUAL",
|
||||
idempotencyKey: `wp_idem_${n}`,
|
||||
userProvidedIdempotencyKey: false,
|
||||
environmentId,
|
||||
projectId,
|
||||
},
|
||||
});
|
||||
await prisma.taskRunWaitpoint.create({
|
||||
data: { taskRunId: rootRun.id, waitpointId: waitpoint.id, projectId },
|
||||
});
|
||||
|
||||
await prisma.taskRunCheckpoint.create({
|
||||
data: {
|
||||
friendlyId: `trcp_${n}`,
|
||||
type: "DOCKER",
|
||||
location: "loc",
|
||||
runtimeEnvironmentId: environmentId,
|
||||
projectId,
|
||||
},
|
||||
});
|
||||
|
||||
const checkpoint = await prisma.checkpoint.create({
|
||||
data: {
|
||||
friendlyId: `cp_${n}`,
|
||||
type: "DOCKER",
|
||||
location: "loc",
|
||||
imageRef: "ref",
|
||||
runId: rootRun.id,
|
||||
attemptId: attempt.id,
|
||||
runtimeEnvironmentId: environmentId,
|
||||
projectId,
|
||||
},
|
||||
});
|
||||
await prisma.checkpointRestoreEvent.create({
|
||||
data: {
|
||||
type: "CHECKPOINT",
|
||||
checkpointId: checkpoint.id,
|
||||
runId: rootRun.id,
|
||||
attemptId: attempt.id,
|
||||
runtimeEnvironmentId: environmentId,
|
||||
projectId,
|
||||
},
|
||||
});
|
||||
|
||||
const batch = await prisma.batchTaskRun.create({
|
||||
data: { friendlyId: `batch_${n}`, runtimeEnvironmentId: environmentId },
|
||||
});
|
||||
await prisma.taskRun.update({ where: { id: childRun.id }, data: { batchId: batch.id } });
|
||||
|
||||
const sourceRun = await prisma.taskRun.create({ data: baseRun("src") });
|
||||
const destRun = await prisma.taskRun.create({ data: baseRun("dst") });
|
||||
return { sourceRunId: sourceRun.id, destinationRunId: destRun.id };
|
||||
}
|
||||
|
||||
/** Attach a BulkActionItem (control-plane-resident) over the given source/destination runs. */
|
||||
async function seedBulkActionItem(
|
||||
prisma: PrismaClient,
|
||||
runs: { sourceRunId: string; destinationRunId: string }
|
||||
): Promise<void> {
|
||||
await prisma.bulkActionItem.create({
|
||||
data: {
|
||||
groupId: `grp_${seedCounter++}`,
|
||||
type: "REPLAY",
|
||||
sourceRunId: runs.sourceRunId,
|
||||
destinationRunId: runs.destinationRunId,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
async function subgraphCountsForScope(
|
||||
prisma: SeedClient,
|
||||
scope: { projectId: string }
|
||||
): Promise<Record<SubgraphTable, number>> {
|
||||
const { projectId } = scope;
|
||||
return {
|
||||
taskRun: await prisma.taskRun.count({ where: { projectId } }),
|
||||
taskRunAttempt: await prisma.taskRunAttempt.count({ where: { taskRun: { projectId } } }),
|
||||
waitpoint: await prisma.waitpoint.count({ where: { projectId } }),
|
||||
taskRunWaitpoint: await prisma.taskRunWaitpoint.count({ where: { projectId } }),
|
||||
taskRunCheckpoint: await prisma.taskRunCheckpoint.count({ where: { projectId } }),
|
||||
checkpoint: await prisma.checkpoint.count({ where: { projectId } }),
|
||||
checkpointRestoreEvent: await prisma.checkpointRestoreEvent.count({ where: { projectId } }),
|
||||
batchTaskRun: await prisma.batchTaskRun.count({
|
||||
where: { runs: { some: { projectId } } },
|
||||
}),
|
||||
};
|
||||
}
|
||||
|
||||
async function bulkActionItemCountForScope(
|
||||
prisma: PrismaClient,
|
||||
scope: { projectId: string }
|
||||
): Promise<number> {
|
||||
return prisma.bulkActionItem.count({ where: { sourceRun: { projectId: scope.projectId } } });
|
||||
}
|
||||
|
||||
function expectSubgraphAllZero(counts: Record<SubgraphTable, number>) {
|
||||
for (const table of SUBGRAPH_TABLES) {
|
||||
expect(counts[table], `${table} should be empty`).toBe(0);
|
||||
}
|
||||
}
|
||||
|
||||
function expectSubgraphAllNonZero(counts: Record<SubgraphTable, number>) {
|
||||
for (const table of SUBGRAPH_TABLES) {
|
||||
expect(counts[table], `${table} should be seeded`).toBeGreaterThan(0);
|
||||
}
|
||||
}
|
||||
|
||||
describe("RunOpsCascadeCleanupService", () => {
|
||||
// REGRESSION: the NEW run-ops writer is a real RunOpsPrismaClient over the dedicated
|
||||
// SUBSET schema — it has NO `bulkActionItem` delegate. Before the fix, the per-writer pass called
|
||||
// `writer.bulkActionItem.deleteMany` on this client => TypeError (Cannot read properties of
|
||||
// undefined). After the fix, BulkActionItem is cleaned ONLY on the control-plane writer (prisma14),
|
||||
// and the run-subgraph is deleted on the NEW DB without throwing.
|
||||
heteroRunOpsPostgresTest(
|
||||
"cleanupProject does not throw on the dedicated RunOpsPrismaClient and clears the new DB subgraph",
|
||||
async ({ prisma14, prisma17 }) => {
|
||||
await dropCrossSeamFks(prisma14, BULK_ACTION_CROSS_SEAM_FKS);
|
||||
|
||||
// The dedicated run-ops subset schema scalarizes every control-plane FK, so the NEW DB needs
|
||||
// NO org/project/env prereqs — seed the subgraph directly with synthetic scope ids.
|
||||
const newScope = makeSyntheticScope();
|
||||
await seedRunOpsSubgraph(prisma17 as unknown as SeedClient, newScope);
|
||||
|
||||
// BulkActionItem (control-plane-resident) lives only on the control-plane DB.
|
||||
const cp = await seedScope(prisma14);
|
||||
const cpRuns = await seedRunOpsSubgraph(prisma14, cp);
|
||||
await seedBulkActionItem(prisma14, cpRuns);
|
||||
|
||||
// prisma17 is a real RunOpsPrismaClient (subset, no bulkActionItem delegate); prisma14 is the
|
||||
// control-plane writer. Before the fix this threw a TypeError on writer.bulkActionItem.
|
||||
const result = await new RunOpsCascadeCleanupService({
|
||||
runOpsWriters: [prisma17, prisma14 as unknown as RunOpsPrismaClient],
|
||||
controlPlaneWriter: prisma14,
|
||||
}).cleanupProject(newScope.projectId);
|
||||
|
||||
expectSubgraphAllZero(
|
||||
await subgraphCountsForScope(prisma17 as unknown as SeedClient, newScope)
|
||||
);
|
||||
// BulkActionItem cleanup ran against the control-plane writer and deleted the control-plane
|
||||
// project's item; the subset client was never asked for the missing delegate.
|
||||
expect(result.bulkActionItem).toBeGreaterThanOrEqual(0);
|
||||
}
|
||||
);
|
||||
|
||||
// REGRESSION (env variant): same guarantee for cleanupEnvironment.
|
||||
heteroRunOpsPostgresTest(
|
||||
"cleanupEnvironment does not throw on the dedicated RunOpsPrismaClient and clears the new DB subgraph",
|
||||
async ({ prisma14, prisma17 }) => {
|
||||
await dropCrossSeamFks(prisma14, BULK_ACTION_CROSS_SEAM_FKS);
|
||||
|
||||
const newScope = makeSyntheticScope();
|
||||
await seedRunOpsSubgraph(prisma17 as unknown as SeedClient, newScope);
|
||||
|
||||
const cp = await seedScope(prisma14);
|
||||
const cpRuns = await seedRunOpsSubgraph(prisma14, cp);
|
||||
await seedBulkActionItem(prisma14, cpRuns);
|
||||
|
||||
const result = await new RunOpsCascadeCleanupService({
|
||||
runOpsWriters: [prisma17, prisma14 as unknown as RunOpsPrismaClient],
|
||||
controlPlaneWriter: prisma14,
|
||||
}).cleanupEnvironment(newScope.environmentId);
|
||||
|
||||
expectSubgraphAllZero(
|
||||
await subgraphCountsForScope(prisma17 as unknown as SeedClient, newScope)
|
||||
);
|
||||
expect(result.bulkActionItem).toBeGreaterThanOrEqual(0);
|
||||
}
|
||||
);
|
||||
|
||||
// Env cleanup over both writers empties the subgraph on BOTH DBs + BulkActionItem on the
|
||||
// control-plane DB; a sibling scope survives.
|
||||
heteroPostgresTest(
|
||||
"cleanupEnvironment empties the subgraph across both writers, isolating a sibling env",
|
||||
async ({ prisma14, prisma17 }) => {
|
||||
await dropCrossSeamFks(prisma17, SUBGRAPH_CROSS_SEAM_FKS);
|
||||
await dropCrossSeamFks(prisma17, BULK_ACTION_CROSS_SEAM_FKS);
|
||||
|
||||
const target14 = await seedScope(prisma14);
|
||||
const target17 = await seedScope(prisma17);
|
||||
const targetRuns14 = await seedRunOpsSubgraph(prisma14, target14);
|
||||
await seedRunOpsSubgraph(prisma17, target17);
|
||||
await seedBulkActionItem(prisma14, targetRuns14);
|
||||
|
||||
const sibling14 = await seedScope(prisma14);
|
||||
const sibling17 = await seedScope(prisma17);
|
||||
const siblingRuns14 = await seedRunOpsSubgraph(prisma14, sibling14);
|
||||
await seedRunOpsSubgraph(prisma17, sibling17);
|
||||
await seedBulkActionItem(prisma14, siblingRuns14);
|
||||
|
||||
await new RunOpsCascadeCleanupService({
|
||||
runOpsWriters: [prisma17, prisma14],
|
||||
controlPlaneWriter: prisma14,
|
||||
}).cleanupEnvironment(target14.environmentId);
|
||||
await new RunOpsCascadeCleanupService({
|
||||
runOpsWriters: [prisma17, prisma14],
|
||||
controlPlaneWriter: prisma14,
|
||||
}).cleanupEnvironment(target17.environmentId);
|
||||
|
||||
expectSubgraphAllZero(await subgraphCountsForScope(prisma14, target14));
|
||||
expectSubgraphAllZero(await subgraphCountsForScope(prisma17, target17));
|
||||
expect(await bulkActionItemCountForScope(prisma14, target14)).toBe(0);
|
||||
expectSubgraphAllNonZero(await subgraphCountsForScope(prisma14, sibling14));
|
||||
expectSubgraphAllNonZero(await subgraphCountsForScope(prisma17, sibling17));
|
||||
expect(await bulkActionItemCountForScope(prisma14, sibling14)).toBeGreaterThan(0);
|
||||
}
|
||||
);
|
||||
|
||||
// Project cleanup over both writers.
|
||||
heteroPostgresTest(
|
||||
"cleanupProject empties the subgraph across both writers, isolating a sibling project",
|
||||
async ({ prisma14, prisma17 }) => {
|
||||
await dropCrossSeamFks(prisma17, SUBGRAPH_CROSS_SEAM_FKS);
|
||||
await dropCrossSeamFks(prisma17, BULK_ACTION_CROSS_SEAM_FKS);
|
||||
|
||||
const target14 = await seedScope(prisma14);
|
||||
const target17 = await seedScope(prisma17);
|
||||
const targetRuns14 = await seedRunOpsSubgraph(prisma14, target14);
|
||||
await seedRunOpsSubgraph(prisma17, target17);
|
||||
await seedBulkActionItem(prisma14, targetRuns14);
|
||||
|
||||
const sibling14 = await seedScope(prisma14);
|
||||
const sibling17 = await seedScope(prisma17);
|
||||
const siblingRuns14 = await seedRunOpsSubgraph(prisma14, sibling14);
|
||||
await seedRunOpsSubgraph(prisma17, sibling17);
|
||||
await seedBulkActionItem(prisma14, siblingRuns14);
|
||||
|
||||
await new RunOpsCascadeCleanupService({
|
||||
runOpsWriters: [prisma17, prisma14],
|
||||
controlPlaneWriter: prisma14,
|
||||
}).cleanupProject(target14.projectId);
|
||||
await new RunOpsCascadeCleanupService({
|
||||
runOpsWriters: [prisma17, prisma14],
|
||||
controlPlaneWriter: prisma14,
|
||||
}).cleanupProject(target17.projectId);
|
||||
|
||||
expectSubgraphAllZero(await subgraphCountsForScope(prisma14, target14));
|
||||
expectSubgraphAllZero(await subgraphCountsForScope(prisma17, target17));
|
||||
expect(await bulkActionItemCountForScope(prisma14, target14)).toBe(0);
|
||||
expectSubgraphAllNonZero(await subgraphCountsForScope(prisma14, sibling14));
|
||||
expectSubgraphAllNonZero(await subgraphCountsForScope(prisma17, sibling17));
|
||||
expect(await bulkActionItemCountForScope(prisma14, sibling14)).toBeGreaterThan(0);
|
||||
}
|
||||
);
|
||||
|
||||
// Idempotency — a second cleanup returns all-zero counts and does not throw on either DB.
|
||||
heteroPostgresTest(
|
||||
"cleanupEnvironment is idempotent on a re-run across both FK configs",
|
||||
async ({ prisma14, prisma17 }) => {
|
||||
await dropCrossSeamFks(prisma17, SUBGRAPH_CROSS_SEAM_FKS);
|
||||
await dropCrossSeamFks(prisma17, BULK_ACTION_CROSS_SEAM_FKS);
|
||||
|
||||
const t14 = await seedScope(prisma14);
|
||||
const t17 = await seedScope(prisma17);
|
||||
const runs14 = await seedRunOpsSubgraph(prisma14, t14);
|
||||
await seedRunOpsSubgraph(prisma17, t17);
|
||||
await seedBulkActionItem(prisma14, runs14);
|
||||
|
||||
const svc14 = new RunOpsCascadeCleanupService({
|
||||
runOpsWriters: [prisma14],
|
||||
controlPlaneWriter: prisma14,
|
||||
});
|
||||
const svc17 = new RunOpsCascadeCleanupService({
|
||||
runOpsWriters: [prisma17],
|
||||
controlPlaneWriter: prisma14,
|
||||
});
|
||||
await svc14.cleanupEnvironment(t14.environmentId);
|
||||
await svc17.cleanupEnvironment(t17.environmentId);
|
||||
|
||||
const second14 = await svc14.cleanupEnvironment(t14.environmentId);
|
||||
const second17 = await svc17.cleanupEnvironment(t17.environmentId);
|
||||
|
||||
for (const result of [second14, second17]) {
|
||||
for (const count of Object.values(result)) {
|
||||
expect(count).toBe(0);
|
||||
}
|
||||
}
|
||||
expectSubgraphAllZero(await subgraphCountsForScope(prisma14, t14));
|
||||
expectSubgraphAllZero(await subgraphCountsForScope(prisma17, t17));
|
||||
}
|
||||
);
|
||||
|
||||
// FK-retained vs FK-dropped fixtures reach an identical run-subgraph end-state.
|
||||
heteroPostgresTest(
|
||||
"FK-retained and FK-dropped fixtures reach an identical end-state after cleanup",
|
||||
async ({ prisma14, prisma17 }) => {
|
||||
await dropCrossSeamFks(prisma17, SUBGRAPH_CROSS_SEAM_FKS);
|
||||
|
||||
const s14 = await seedScope(prisma14);
|
||||
const s17 = await seedScope(prisma17);
|
||||
await seedRunOpsSubgraph(prisma14, s14);
|
||||
await seedRunOpsSubgraph(prisma17, s17);
|
||||
|
||||
await new RunOpsCascadeCleanupService({
|
||||
runOpsWriters: [prisma14],
|
||||
controlPlaneWriter: prisma14,
|
||||
}).cleanupEnvironment(s14.environmentId);
|
||||
await new RunOpsCascadeCleanupService({
|
||||
runOpsWriters: [prisma17],
|
||||
controlPlaneWriter: prisma17,
|
||||
}).cleanupEnvironment(s17.environmentId);
|
||||
|
||||
const counts14 = await subgraphCountsForScope(prisma14, s14);
|
||||
const counts17 = await subgraphCountsForScope(prisma17, s17);
|
||||
expect(counts17).toEqual(counts14);
|
||||
}
|
||||
);
|
||||
|
||||
// Single-DB mode — the same client passed twice de-dups so the pass runs once.
|
||||
heteroPostgresTest(
|
||||
"single-DB: the same client passed twice de-dups so the delete pass runs exactly once",
|
||||
async ({ prisma14 }) => {
|
||||
const scope = await seedScope(prisma14);
|
||||
await seedRunOpsSubgraph(prisma14, scope);
|
||||
|
||||
const before = await subgraphCountsForScope(prisma14, scope);
|
||||
|
||||
// Wrap the real client with a $extends query hook that counts deleteMany calls per model. NOT
|
||||
// a mock — the query still runs against the container. If de-dup failed, the loop would run
|
||||
// twice against this same client and taskRun.deleteMany would fire twice.
|
||||
let taskRunDeleteManyCalls = 0;
|
||||
const counting = prisma14.$extends({
|
||||
query: {
|
||||
taskRun: {
|
||||
async deleteMany({ args, query }) {
|
||||
taskRunDeleteManyCalls++;
|
||||
return query(args);
|
||||
},
|
||||
},
|
||||
},
|
||||
}) as unknown as typeof prisma14;
|
||||
|
||||
const result = await new RunOpsCascadeCleanupService({
|
||||
runOpsWriters: [counting, counting],
|
||||
controlPlaneWriter: counting,
|
||||
}).cleanupEnvironment(scope.environmentId);
|
||||
|
||||
// De-dup ran the pass exactly once: one taskRun.deleteMany, count not double-summed.
|
||||
expect(taskRunDeleteManyCalls).toBe(1);
|
||||
expect(result.taskRun).toBe(before.taskRun);
|
||||
expectSubgraphAllZero(await subgraphCountsForScope(prisma14, scope));
|
||||
}
|
||||
);
|
||||
|
||||
// The two-writer split — an env whose rows straddle both DBs (cuid runs on the LEGACY DB,
|
||||
// ksuid runs on the NEW DB) is fully cleaned by one call; a single-writer service leaks orphans.
|
||||
heteroPostgresTest(
|
||||
"two-writer fan-out cleans a split env on both DBs; single-writer leaves orphans",
|
||||
async ({ prisma14, prisma17 }) => {
|
||||
await dropCrossSeamFks(prisma17, SUBGRAPH_CROSS_SEAM_FKS);
|
||||
|
||||
// One logical env that exists on both DBs (control-plane prereqs seeded on each), with the
|
||||
// SAME env id, modelling the reference-equal control-plane row. We force a shared id
|
||||
// by creating the legacy scope first, then mirroring its env id onto the new DB.
|
||||
const legacy = await seedScope(prisma14);
|
||||
const newOrg = await prisma17.organization.create({
|
||||
data: { id: legacy.organizationId, title: "mirror", slug: `mirror-${seedCounter++}` },
|
||||
});
|
||||
const newProject = await prisma17.project.create({
|
||||
data: {
|
||||
id: legacy.projectId,
|
||||
name: "mirror",
|
||||
slug: `mirror-${seedCounter++}`,
|
||||
externalRef: `mirror_${seedCounter++}`,
|
||||
organizationId: newOrg.id,
|
||||
},
|
||||
});
|
||||
const newEnv = await prisma17.runtimeEnvironment.create({
|
||||
data: {
|
||||
id: legacy.environmentId,
|
||||
type: "PRODUCTION",
|
||||
slug: `mirror-${seedCounter++}`,
|
||||
projectId: newProject.id,
|
||||
organizationId: newOrg.id,
|
||||
apiKey: `tr_${seedCounter++}`,
|
||||
pkApiKey: `pk_${seedCounter++}`,
|
||||
shortcode: `sc_${seedCounter++}`,
|
||||
},
|
||||
});
|
||||
const newWorker = await prisma17.backgroundWorker.create({
|
||||
data: {
|
||||
friendlyId: `w_${seedCounter++}`,
|
||||
contentHash: "h",
|
||||
projectId: newProject.id,
|
||||
runtimeEnvironmentId: newEnv.id,
|
||||
version: `2024.2.${seedCounter++}`,
|
||||
metadata: {},
|
||||
engine: "V2",
|
||||
},
|
||||
});
|
||||
const newTask = await prisma17.backgroundWorkerTask.create({
|
||||
data: {
|
||||
friendlyId: `t_${seedCounter++}`,
|
||||
slug: `s-${seedCounter++}`,
|
||||
filePath: "index.ts",
|
||||
exportName: "myTask",
|
||||
workerId: newWorker.id,
|
||||
runtimeEnvironmentId: newEnv.id,
|
||||
projectId: newProject.id,
|
||||
},
|
||||
});
|
||||
const newQueue = await prisma17.taskQueue.create({
|
||||
data: {
|
||||
friendlyId: `q_${seedCounter++}`,
|
||||
name: `task/s-${seedCounter++}`,
|
||||
runtimeEnvironmentId: newEnv.id,
|
||||
projectId: newProject.id,
|
||||
},
|
||||
});
|
||||
|
||||
const newScope = {
|
||||
projectId: newProject.id,
|
||||
environmentId: newEnv.id,
|
||||
organizationId: newOrg.id,
|
||||
backgroundWorkerId: newWorker.id,
|
||||
workerTaskId: newTask.id,
|
||||
queueId: newQueue.id,
|
||||
};
|
||||
|
||||
// Pre-cutover (LEGACY DB) and post-cutover (NEW DB) run-ops rows for the SAME env.
|
||||
await seedRunOpsSubgraph(prisma14, legacy);
|
||||
await seedRunOpsSubgraph(prisma17, newScope);
|
||||
|
||||
// Two-writer fan-out: one call cleans BOTH DBs.
|
||||
await new RunOpsCascadeCleanupService({
|
||||
runOpsWriters: [prisma17, prisma14],
|
||||
controlPlaneWriter: prisma14,
|
||||
}).cleanupEnvironment(legacy.environmentId);
|
||||
|
||||
expectSubgraphAllZero(await subgraphCountsForScope(prisma14, legacy));
|
||||
expectSubgraphAllZero(await subgraphCountsForScope(prisma17, newScope));
|
||||
|
||||
// The orphan-leak guard: re-seed and run a mis-built SINGLE-writer service; it must leave the
|
||||
// OTHER DB's rows behind.
|
||||
await seedRunOpsSubgraph(prisma14, legacy);
|
||||
await seedRunOpsSubgraph(prisma17, newScope);
|
||||
|
||||
await new RunOpsCascadeCleanupService({
|
||||
runOpsWriters: [prisma17],
|
||||
controlPlaneWriter: prisma14,
|
||||
}).cleanupEnvironment(legacy.environmentId);
|
||||
|
||||
// NEW DB cleaned, LEGACY DB orphans remain — proving a one-handle delete leaks.
|
||||
expectSubgraphAllZero(await subgraphCountsForScope(prisma17, newScope));
|
||||
const leaked = await subgraphCountsForScope(prisma14, legacy);
|
||||
expect(leaked.taskRun).toBeGreaterThan(0);
|
||||
}
|
||||
);
|
||||
});
|
||||
@@ -31,6 +31,8 @@ RUN --mount=type=cache,id=pnpm,target=/root/.local/share/pnpm/store pnpm install
|
||||
# Generate Prisma client here where all deps are installed
|
||||
COPY --from=pruner --chown=node:node /triggerdotdev/internal-packages/database/prisma/schema.prisma /triggerdotdev/internal-packages/database/prisma/schema.prisma
|
||||
RUN pnpx prisma@6.14.0 generate --schema /triggerdotdev/internal-packages/database/prisma/schema.prisma
|
||||
COPY --from=pruner --chown=node:node /triggerdotdev/internal-packages/run-ops-database/prisma/schema.prisma /triggerdotdev/internal-packages/run-ops-database/prisma/schema.prisma
|
||||
RUN pnpx prisma@6.14.0 generate --schema /triggerdotdev/internal-packages/run-ops-database/prisma/schema.prisma
|
||||
|
||||
## Production deps
|
||||
FROM base AS production-deps
|
||||
@@ -85,6 +87,9 @@ COPY --from=pruner --chown=node:node /triggerdotdev/out/full/ .
|
||||
COPY --from=production-deps --chown=node:node /triggerdotdev .
|
||||
# Copy generated Prisma client from dev-deps
|
||||
COPY --from=dev-deps --chown=node:node /triggerdotdev/internal-packages/database/generated ./internal-packages/database/generated
|
||||
# Run-ops Prisma client (query engine + client). Only constructed when the split is enabled,
|
||||
# but the image must carry it so a split-on deployment doesn't hit a missing query engine.
|
||||
COPY --from=dev-deps --chown=node:node /triggerdotdev/internal-packages/run-ops-database/generated ./internal-packages/run-ops-database/generated
|
||||
COPY --from=builder --chown=node:node /triggerdotdev/apps/webapp/build/server.js ./apps/webapp/build/server.js
|
||||
COPY --from=builder --chown=node:node /triggerdotdev/apps/webapp/build ./apps/webapp/build
|
||||
COPY --from=builder --chown=node:node /triggerdotdev/apps/webapp/public ./apps/webapp/public
|
||||
|
||||
+5
@@ -0,0 +1,5 @@
|
||||
-- Fail fast instead of queueing behind a long txn/VACUUM for the ACCESS EXCLUSIVE lock.
|
||||
SET lock_timeout = '5s';
|
||||
|
||||
-- DropForeignKey
|
||||
ALTER TABLE "BulkActionItem" DROP CONSTRAINT IF EXISTS "BulkActionItem_groupId_fkey";
|
||||
+5
@@ -0,0 +1,5 @@
|
||||
-- Fail fast instead of queueing behind a long txn/VACUUM for the ACCESS EXCLUSIVE lock.
|
||||
SET lock_timeout = '5s';
|
||||
|
||||
-- DropForeignKey
|
||||
ALTER TABLE "BatchTaskRun" DROP CONSTRAINT IF EXISTS "BatchTaskRun_runtimeEnvironmentId_fkey";
|
||||
+29
@@ -0,0 +1,29 @@
|
||||
-- Run-ops DB split: drop run-ops tables' cross-database FKs to control-plane tables (they
|
||||
-- can't be enforced on the run-ops DB, which has no control-plane tables). Mirrors the
|
||||
-- earlier TaskRun/BatchTaskRun drops; IF EXISTS so it's idempotent across both databases.
|
||||
|
||||
-- Fail fast instead of queueing behind a long txn/VACUUM for the ACCESS EXCLUSIVE lock.
|
||||
SET lock_timeout = '5s';
|
||||
|
||||
-- Waitpoint
|
||||
ALTER TABLE "Waitpoint" DROP CONSTRAINT IF EXISTS "Waitpoint_projectId_fkey";
|
||||
ALTER TABLE "Waitpoint" DROP CONSTRAINT IF EXISTS "Waitpoint_environmentId_fkey";
|
||||
|
||||
-- TaskRunWaitpoint
|
||||
ALTER TABLE "TaskRunWaitpoint" DROP CONSTRAINT IF EXISTS "TaskRunWaitpoint_projectId_fkey";
|
||||
|
||||
-- TaskRunCheckpoint
|
||||
ALTER TABLE "TaskRunCheckpoint" DROP CONSTRAINT IF EXISTS "TaskRunCheckpoint_projectId_fkey";
|
||||
ALTER TABLE "TaskRunCheckpoint" DROP CONSTRAINT IF EXISTS "TaskRunCheckpoint_runtimeEnvironmentId_fkey";
|
||||
|
||||
-- TaskRunAttempt
|
||||
ALTER TABLE "TaskRunAttempt" DROP CONSTRAINT IF EXISTS "TaskRunAttempt_backgroundWorkerId_fkey";
|
||||
ALTER TABLE "TaskRunAttempt" DROP CONSTRAINT IF EXISTS "TaskRunAttempt_backgroundWorkerTaskId_fkey";
|
||||
ALTER TABLE "TaskRunAttempt" DROP CONSTRAINT IF EXISTS "TaskRunAttempt_runtimeEnvironmentId_fkey";
|
||||
|
||||
-- TaskRunTag
|
||||
ALTER TABLE "TaskRunTag" DROP CONSTRAINT IF EXISTS "TaskRunTag_projectId_fkey";
|
||||
|
||||
-- WaitpointTag
|
||||
ALTER TABLE "WaitpointTag" DROP CONSTRAINT IF EXISTS "WaitpointTag_projectId_fkey";
|
||||
ALTER TABLE "WaitpointTag" DROP CONSTRAINT IF EXISTS "WaitpointTag_environmentId_fkey";
|
||||
+10
@@ -0,0 +1,10 @@
|
||||
-- Run-ops DB split: drop ProjectAlert's cross-DB FKs into the run subgraph (taskRunId -> TaskRun,
|
||||
-- taskRunAttemptId -> TaskRunAttempt). A ksuid run lives only on the dedicated run-ops DB so the FK
|
||||
-- can't be enforced; scalar columns are kept and the run is resolved via runStore.findRun. Mirrors
|
||||
-- ca1a4e18e; IF EXISTS so it's idempotent across both databases.
|
||||
|
||||
-- Fail fast instead of queueing behind a long txn/VACUUM for the ACCESS EXCLUSIVE lock.
|
||||
SET lock_timeout = '5s';
|
||||
|
||||
ALTER TABLE "ProjectAlert" DROP CONSTRAINT IF EXISTS "ProjectAlert_taskRunId_fkey";
|
||||
ALTER TABLE "ProjectAlert" DROP CONSTRAINT IF EXISTS "ProjectAlert_taskRunAttemptId_fkey";
|
||||
+12
@@ -0,0 +1,12 @@
|
||||
/*
|
||||
Warnings:
|
||||
|
||||
- You are about to drop the `_TaskRunToTaskRunTag` table. If the table is not empty, all the data it contains will be lost.
|
||||
|
||||
*/
|
||||
|
||||
-- Fail fast instead of queueing behind a long txn/VACUUM for the ACCESS EXCLUSIVE lock.
|
||||
SET lock_timeout = '5s';
|
||||
|
||||
-- DropTable
|
||||
DROP TABLE IF EXISTS "_TaskRunToTaskRunTag" CASCADE;
|
||||
@@ -369,7 +369,6 @@ model RuntimeEnvironment {
|
||||
backgroundWorkerTasks BackgroundWorkerTask[]
|
||||
taskRuns TaskRun[]
|
||||
taskQueues TaskQueue[]
|
||||
batchTaskRuns BatchTaskRun[]
|
||||
environmentVariableValues EnvironmentVariableValue[]
|
||||
checkpoints Checkpoint[]
|
||||
workerDeployments WorkerDeployment[]
|
||||
@@ -939,7 +938,6 @@ model TaskRun {
|
||||
updatedAt DateTime @updatedAt
|
||||
|
||||
attempts TaskRunAttempt[] @relation("attempts")
|
||||
tags TaskRunTag[]
|
||||
|
||||
/// Denormized column that holds the raw tags
|
||||
runTags String[]
|
||||
@@ -1004,8 +1002,6 @@ model TaskRun {
|
||||
CheckpointRestoreEvent CheckpointRestoreEvent[]
|
||||
executionSnapshots TaskRunExecutionSnapshot[]
|
||||
|
||||
alerts ProjectAlert[]
|
||||
|
||||
scheduleInstanceId String?
|
||||
scheduleId String?
|
||||
|
||||
@@ -1574,8 +1570,6 @@ model TaskRunTag {
|
||||
|
||||
friendlyId String @unique
|
||||
|
||||
runs TaskRun[]
|
||||
|
||||
project Project @relation(fields: [projectId], references: [id], onDelete: Cascade, onUpdate: Cascade)
|
||||
projectId String
|
||||
|
||||
@@ -1674,7 +1668,6 @@ model TaskRunAttempt {
|
||||
checkpoints Checkpoint[]
|
||||
batchTaskRunItems BatchTaskRunItem[]
|
||||
CheckpointRestoreEvent CheckpointRestoreEvent[]
|
||||
alerts ProjectAlert[]
|
||||
childRuns TaskRun[] @relation("TaskParentRunAttempt")
|
||||
|
||||
@@unique([taskRunId, number])
|
||||
@@ -1872,7 +1865,6 @@ model BatchTaskRun {
|
||||
friendlyId String @unique
|
||||
idempotencyKey String?
|
||||
idempotencyKeyExpiresAt DateTime?
|
||||
runtimeEnvironment RuntimeEnvironment @relation(fields: [runtimeEnvironmentId], references: [id], onDelete: Cascade, onUpdate: Cascade)
|
||||
status BatchTaskRunStatus @default(PENDING)
|
||||
runtimeEnvironmentId String
|
||||
/// This only includes new runs, not idempotent runs.
|
||||
@@ -2373,10 +2365,10 @@ model ProjectAlert {
|
||||
|
||||
type ProjectAlertType
|
||||
|
||||
taskRunAttempt TaskRunAttempt? @relation(fields: [taskRunAttemptId], references: [id], onDelete: Cascade, onUpdate: Cascade)
|
||||
// Run-ops split: these reference run-subgraph rows that live on the dedicated run-ops DB for ksuid
|
||||
// runs, so the cross-DB FK can't hold. Scalar-only; the run is resolved via runStore.findRun.
|
||||
taskRunAttemptId String?
|
||||
|
||||
taskRun TaskRun? @relation(fields: [taskRunId], references: [id], onDelete: Cascade, onUpdate: Cascade)
|
||||
taskRunId String?
|
||||
|
||||
workerDeployment WorkerDeployment? @relation(fields: [workerDeploymentId], references: [id], onDelete: Cascade, onUpdate: Cascade)
|
||||
@@ -2488,7 +2480,6 @@ model BulkActionGroup {
|
||||
environmentId String?
|
||||
|
||||
type BulkActionType
|
||||
items BulkActionItem[]
|
||||
|
||||
/// When the group is created it's pending. After we've processed all the items it's completed. This does not mean the associated runs are completed.
|
||||
status BulkActionStatus @default(PENDING)
|
||||
@@ -2556,7 +2547,6 @@ model BulkActionItem {
|
||||
/// @deprecated not used in new BulkActions
|
||||
friendlyId String?
|
||||
|
||||
group BulkActionGroup @relation(fields: [groupId], references: [id], onDelete: Cascade, onUpdate: Cascade)
|
||||
groupId String
|
||||
|
||||
type BulkActionType
|
||||
|
||||
@@ -0,0 +1,133 @@
|
||||
// ProjectAlert.taskRunId/taskRunAttemptId FKs point INTO the run subgraph. A ksuid run lives ONLY
|
||||
// on the dedicated run-ops DB (prisma17), so `projectAlert.create({ taskRunId: <ksuid> })` on
|
||||
// control-plane (prisma14) violates the FK and the alert is silently dropped. After the FK drop +
|
||||
// @relation removal the create succeeds; the read path resolves the run via runStore.findRun.
|
||||
// Asserts the create succeeds: it fails with an FK violation before the fix and succeeds after.
|
||||
|
||||
import { heteroRunOpsPostgresTest } from "@internal/testcontainers";
|
||||
import type { PrismaClient } from "@trigger.dev/database";
|
||||
import type { RunOpsPrismaClient } from "@internal/run-ops-database";
|
||||
import { describe, expect } from "vitest";
|
||||
|
||||
// 27-char internal id → ksuid → NEW (lives only on the dedicated run-ops DB).
|
||||
const KSUID_27 = "k".repeat(27);
|
||||
|
||||
async function seedControlPlaneAlertPrereqs(prisma: PrismaClient, suffix: string) {
|
||||
const organization = await prisma.organization.create({
|
||||
data: { title: `Org ${suffix}`, slug: `org-${suffix}` },
|
||||
});
|
||||
const project = await prisma.project.create({
|
||||
data: {
|
||||
name: `Project ${suffix}`,
|
||||
slug: `project-${suffix}`,
|
||||
externalRef: `proj_${suffix}`,
|
||||
organizationId: organization.id,
|
||||
},
|
||||
});
|
||||
const environment = await prisma.runtimeEnvironment.create({
|
||||
data: {
|
||||
type: "PRODUCTION",
|
||||
slug: "prod",
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
apiKey: `tr_prod_${suffix}`,
|
||||
pkApiKey: `pk_prod_${suffix}`,
|
||||
shortcode: `short_${suffix}`,
|
||||
},
|
||||
});
|
||||
const channel = await prisma.projectAlertChannel.create({
|
||||
data: {
|
||||
friendlyId: `alert_channel_${suffix}`,
|
||||
type: "EMAIL",
|
||||
name: "Email",
|
||||
properties: { type: "EMAIL", email: "alerts@example.com" },
|
||||
alertTypes: ["TASK_RUN"],
|
||||
projectId: project.id,
|
||||
},
|
||||
});
|
||||
return { organization, project, environment, channel };
|
||||
}
|
||||
|
||||
describe("ProjectAlert control-plane → run-subgraph FK reconciliation", () => {
|
||||
heteroRunOpsPostgresTest(
|
||||
"creating a TASK_RUN alert with a ksuid taskRunId (run only on the run-ops DB) succeeds on control-plane",
|
||||
async ({ prisma14, prisma17 }) => {
|
||||
const suffix = "alert-ksuid";
|
||||
const { project, environment, channel } = await seedControlPlaneAlertPrereqs(
|
||||
prisma14,
|
||||
suffix
|
||||
);
|
||||
|
||||
// The run exists ONLY on the dedicated run-ops DB (prisma17), never on control-plane.
|
||||
await (prisma17 as RunOpsPrismaClient).taskRun.create({
|
||||
data: {
|
||||
id: KSUID_27,
|
||||
friendlyId: `run_${suffix}`,
|
||||
engine: "V2",
|
||||
status: "COMPLETED_WITH_ERRORS",
|
||||
taskIdentifier: "my-task",
|
||||
payload: "{}",
|
||||
payloadType: "application/json",
|
||||
traceId: `trace_${suffix}`,
|
||||
spanId: `span_${suffix}`,
|
||||
queue: "task/my-task",
|
||||
runtimeEnvironmentId: environment.id,
|
||||
projectId: project.id,
|
||||
organizationId: project.organizationId,
|
||||
environmentType: "PRODUCTION",
|
||||
},
|
||||
});
|
||||
|
||||
// Control-plane has no TaskRun row for KSUID_27. With the FK present this throws P2003;
|
||||
// after the FK is dropped + the @relation removed it succeeds.
|
||||
const alert = await prisma14.projectAlert.create({
|
||||
data: {
|
||||
friendlyId: `alert_${suffix}`,
|
||||
channelId: channel.id,
|
||||
projectId: project.id,
|
||||
environmentId: environment.id,
|
||||
status: "PENDING",
|
||||
type: "TASK_RUN",
|
||||
taskRunId: KSUID_27,
|
||||
},
|
||||
});
|
||||
|
||||
expect(alert.taskRunId).toBe(KSUID_27);
|
||||
|
||||
// The scalar round-trips and can be re-read off the control-plane row (the read path resolves
|
||||
// the actual run via runStore.findRun against the run-ops DB).
|
||||
const reread = await prisma14.projectAlert.findUniqueOrThrow({ where: { id: alert.id } });
|
||||
expect(reread.taskRunId).toBe(KSUID_27);
|
||||
},
|
||||
120_000
|
||||
);
|
||||
|
||||
heteroRunOpsPostgresTest(
|
||||
"creating a TASK_RUN_ATTEMPT alert with a ksuid taskRunAttemptId (attempt only on the run-ops DB) succeeds on control-plane",
|
||||
async ({ prisma14 }) => {
|
||||
const suffix = "alert-ksuid-attempt";
|
||||
const { project, environment, channel } = await seedControlPlaneAlertPrereqs(
|
||||
prisma14,
|
||||
suffix
|
||||
);
|
||||
|
||||
// A ksuid attempt id with no matching control-plane TaskRunAttempt row. With the FK present
|
||||
// this throws P2003; after the FK is dropped it succeeds.
|
||||
const attemptId = "a".repeat(27);
|
||||
const alert = await prisma14.projectAlert.create({
|
||||
data: {
|
||||
friendlyId: `alert_${suffix}`,
|
||||
channelId: channel.id,
|
||||
projectId: project.id,
|
||||
environmentId: environment.id,
|
||||
status: "PENDING",
|
||||
type: "TASK_RUN_ATTEMPT",
|
||||
taskRunAttemptId: attemptId,
|
||||
},
|
||||
});
|
||||
|
||||
expect(alert.taskRunAttemptId).toBe(attemptId);
|
||||
},
|
||||
120_000
|
||||
);
|
||||
});
|
||||
Reference in New Issue
Block a user