fix(webapp): stop slow database cleanup on project deletion (#4191)
## Summary Deleting a project triggered an unbounded database cleanup that scanned the project's entire run history, so deleting a project with many runs could be very slow. Project deletion is a soft delete again: run data is retained and the deletion completes quickly. ## Fix Project deletion ran a cascade hard-delete whose `BulkActionItem` step filtered through a relation to `TaskRun` scoped by `projectId`. Prisma compiles that to an `EXISTS`-join over the project's entire `TaskRun` set (a large, hot table with no `projectId` index), and it ran on every project deletion unconditionally. Removing the cascade-cleanup call restores the prior soft-delete behaviour: queues are removed, the project is marked deleted, and run data is retained. The cascade-cleanup service (added in [#4117](https://github.com/triggerdotdev/trigger.dev/pull/4117)) had no other callers, so it and its test are deleted.
This commit is contained in:
Daniel Sutton
@@ -0,0 +1,6 @@
|
||||
---
|
||||
area: webapp
|
||||
type: fix
|
||||
---
|
||||
|
||||
Deleting a project no longer triggers a slow database cleanup that could hang on projects with many runs.
|
||||
@@ -82,8 +82,8 @@ export class ArchiveBranchService {
|
||||
}
|
||||
|
||||
// Branch archive is a SOFT update — do NOT hard-delete run-ops rows here (it would destroy a
|
||||
// retained branch's history). RunOpsCascadeCleanupService.cleanupEnvironment belongs on the
|
||||
// env hard-delete/purge path (owned by the cloud env-purge runbook), which has no site today.
|
||||
// retained branch's history). Any env hard-delete/purge belongs on a dedicated purge path
|
||||
// (owned by the cloud env-purge runbook), which has no site today.
|
||||
const slug = `${environment.slug}-${nanoid(6)}`;
|
||||
const shortcode = slug;
|
||||
|
||||
|
||||
@@ -3,7 +3,6 @@ import { prisma } from "~/db.server";
|
||||
import { marqs } from "~/v3/marqs/index.server";
|
||||
import { engine } from "~/v3/runEngine.server";
|
||||
import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server";
|
||||
import { RunOpsCascadeCleanupService } from "~/v3/runOpsMigration/runOpsCascadeCleanup.server";
|
||||
|
||||
type Options = ({ projectId: string } | { projectSlug: string }) & {
|
||||
userId: string;
|
||||
@@ -51,9 +50,7 @@ export class DeleteProjectService {
|
||||
});
|
||||
}
|
||||
|
||||
// Hard-delete the project's run-ops rows across both run-ops DBs (replaces the cloud-only
|
||||
// dropped cross-seam FK cascades). Idempotent; uses the run-ops writers, not #prismaClient.
|
||||
await new RunOpsCascadeCleanupService().cleanupProject(project.id);
|
||||
// Soft delete only: run-ops rows are intentionally retained (no hard-delete cascade here).
|
||||
|
||||
// Mark the project as deleted (do this last because it makes it impossible to try again)
|
||||
// - This disables all API keys
|
||||
|
||||
@@ -1,275 +0,0 @@
|
||||
import { type PrismaClient } from "@trigger.dev/database";
|
||||
import { type RunOpsPrismaClient } from "@internal/run-ops-database";
|
||||
import { runOpsLegacyPrisma, runOpsNewPrismaClient } from "~/db.server";
|
||||
|
||||
/**
|
||||
* Structural client covering exactly the run-subgraph delegates + WHERE filters the cascade uses on
|
||||
* a run-ops writer. Both `@trigger.dev/database`'s `PrismaClient` (full schema, legacy writer) and
|
||||
* `@internal/run-ops-database`'s `RunOpsPrismaClient` (dedicated SUBSET schema, new writer) are
|
||||
* assignable to it — the two concrete clients are NOT mutually assignable (the subset adds FK-free
|
||||
* join models the full schema lacks), so a shared structural type is the only common ground.
|
||||
*
|
||||
* Crucially it does NOT expose control-plane-resident models (e.g. `bulkActionItem`) nor scalarized
|
||||
* relations that don't exist on the subset (e.g. `TaskRunWaitpoint.taskRun`), so the compiler now
|
||||
* rejects the two bugs an `as unknown as PrismaClient` cast would otherwise mask.
|
||||
*/
|
||||
type CountResult = { count: number };
|
||||
type RunSubgraphCleanupClient = {
|
||||
taskRun: {
|
||||
findMany(args: {
|
||||
where: { runtimeEnvironmentId: string };
|
||||
select: { id: true };
|
||||
}): Promise<Array<{ id: string }>>;
|
||||
deleteMany(args: {
|
||||
where: { runtimeEnvironmentId: string } | { projectId: string };
|
||||
}): Promise<CountResult>;
|
||||
};
|
||||
taskRunAttempt: {
|
||||
deleteMany(args: {
|
||||
where: { runtimeEnvironmentId: string } | { taskRun: { projectId: string } };
|
||||
}): Promise<CountResult>;
|
||||
};
|
||||
taskRunWaitpoint: {
|
||||
deleteMany(args: {
|
||||
where: { taskRunId: { in: string[] } } | { projectId: string };
|
||||
}): Promise<CountResult>;
|
||||
};
|
||||
taskRunCheckpoint: {
|
||||
deleteMany(args: {
|
||||
where: { runtimeEnvironmentId: string } | { projectId: string };
|
||||
}): Promise<CountResult>;
|
||||
};
|
||||
checkpoint: {
|
||||
deleteMany(args: {
|
||||
where: { runtimeEnvironmentId: string } | { projectId: string };
|
||||
}): Promise<CountResult>;
|
||||
};
|
||||
checkpointRestoreEvent: {
|
||||
deleteMany(args: {
|
||||
where: { runtimeEnvironmentId: string } | { projectId: string };
|
||||
}): Promise<CountResult>;
|
||||
};
|
||||
waitpoint: {
|
||||
deleteMany(args: {
|
||||
where: { environmentId: string } | { projectId: string };
|
||||
}): Promise<CountResult>;
|
||||
};
|
||||
batchTaskRun: {
|
||||
deleteMany(args: {
|
||||
where: { runtimeEnvironmentId: string } | { runs: { some: { projectId: string } } };
|
||||
}): Promise<CountResult>;
|
||||
};
|
||||
};
|
||||
|
||||
// Compile-time assertion that both concrete writers satisfy the structural shape.
|
||||
const _newWriterAssignable: RunSubgraphCleanupClient = undefined as unknown as RunOpsPrismaClient;
|
||||
const _legacyWriterAssignable: RunSubgraphCleanupClient = undefined as unknown as PrismaClient;
|
||||
void _newWriterAssignable;
|
||||
void _legacyWriterAssignable;
|
||||
|
||||
/**
|
||||
* RunOpsCascadeCleanupService — application-level env/project-delete cascade-cleanup that replaces
|
||||
* the cloud-only dropped cross-seam `onDelete: Cascade` FKs crossing run-ops -> control-plane.
|
||||
*
|
||||
* Deletes route through the dedicated run-ops write clients (`runOpsNewPrismaClient` +
|
||||
* `runOpsLegacyPrisma`), NOT the control-plane `prisma`. The ordered delete pass runs against BOTH
|
||||
* writers: a migrating env/project's run-ops rows split across the new (run-ops id) and
|
||||
* legacy (cuid) DBs per the per-env cutover + roll-new-forward rollback, and the
|
||||
* cloud DB that lost its physical FK has no cascade to clean the other writer's miss. In single-DB
|
||||
* both handles are reference-equal to the one collapsed client, so de-dup-by-reference runs the
|
||||
* pass once; the FK cascade also fires there, making these deletes idempotent no-ops.
|
||||
*
|
||||
* The NEW run-ops writer is a dedicated `RunOpsPrismaClient` over the run-subgraph SUBSET schema:
|
||||
* it does NOT carry control-plane-resident models. `BulkActionItem` is one such control-plane model
|
||||
* (it lives in `@trigger.dev/database` but NOT in the run-ops subset), so cleaning it on the NEW
|
||||
* writer would dereference an `undefined` delegate at runtime. Its cleanup therefore runs ONLY
|
||||
* against the control-plane writer; the run-subgraph deletes (which DO exist on both schemas) run
|
||||
* per run-ops writer. Typing the run-ops writers as `RunOpsPrismaClient` makes the compiler reject
|
||||
* any future control-plane-only model access on the NEW writer, so this class of bug can't recur.
|
||||
*
|
||||
* Deliberately NOT gated behind `isSplitEnabled()` (cloud relies on it; self-host treats it as
|
||||
* idempotent insurance). Every delete is `deleteMany`, so a zero-row scope is a no-op and rows a
|
||||
* concurrent FK cascade already removed return `count: 0`. Deletes are not wrapped in one
|
||||
* `$transaction` (no cross-DB txn is possible, and a single huge txn risks long locks); a crash
|
||||
* mid-cleanup is recovered by re-running.
|
||||
*/
|
||||
|
||||
/** Per-table deleted row counts, summed across the distinct run-ops writers actually run. */
|
||||
type CascadeCleanupResult = Record<string, number>;
|
||||
|
||||
type CleanupServiceDeps = {
|
||||
/**
|
||||
* Run-ops write clients to run the run-subgraph delete pass against. Defaults to the two
|
||||
* run-ops writers — NOT the control-plane `prisma`. Typed as the structural
|
||||
* `RunSubgraphCleanupClient` so the compiler rejects control-plane-only model access (e.g.
|
||||
* `bulkActionItem`) and subset-absent relations. De-duped by reference so the single-DB
|
||||
* reference-equal collapse runs the pass once.
|
||||
*/
|
||||
runOpsWriters?: RunSubgraphCleanupClient[];
|
||||
/**
|
||||
* Control-plane writer for control-plane-resident models the run-subgraph cascade must also clean
|
||||
* (currently only `BulkActionItem`, which has no env/project column and is NOT in the run-ops
|
||||
* subset schema). Runs exactly once. Defaults to the legacy run-ops writer, which IS the
|
||||
* control-plane client.
|
||||
*/
|
||||
controlPlaneWriter?: PrismaClient;
|
||||
};
|
||||
|
||||
export class RunOpsCascadeCleanupService {
|
||||
#writers: RunSubgraphCleanupClient[];
|
||||
#controlPlaneWriter: PrismaClient;
|
||||
|
||||
constructor(deps: CleanupServiceDeps = {}) {
|
||||
const writers = deps.runOpsWriters ?? [runOpsNewPrismaClient, runOpsLegacyPrisma];
|
||||
this.#writers = Array.from(new Set(writers));
|
||||
this.#controlPlaneWriter = deps.controlPlaneWriter ?? runOpsLegacyPrisma;
|
||||
}
|
||||
|
||||
/** Delete all run-ops rows scoped to one environment, across every distinct run-ops writer. */
|
||||
public async cleanupEnvironment(runtimeEnvironmentId: string): Promise<CascadeCleanupResult> {
|
||||
const result: CascadeCleanupResult = {};
|
||||
await this.#cleanupBulkActionItemsForEnvironment(runtimeEnvironmentId, result);
|
||||
for (const writer of this.#writers) {
|
||||
await this.#cleanupEnvironmentOnWriter(writer, runtimeEnvironmentId, result);
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
/** Delete all run-ops rows scoped to one project, across every distinct run-ops writer. */
|
||||
public async cleanupProject(projectId: string): Promise<CascadeCleanupResult> {
|
||||
const result: CascadeCleanupResult = {};
|
||||
await this.#cleanupBulkActionItemsForProject(projectId, result);
|
||||
for (const writer of this.#writers) {
|
||||
await this.#cleanupProjectOnWriter(writer, projectId, result);
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
// BulkActionItem is control-plane-resident (it exists in @trigger.dev/database, NOT in the
|
||||
// run-ops subset schema), so it is cleaned only on the control-plane writer. It has no env column;
|
||||
// clean via both run relations (destination may differ).
|
||||
async #cleanupBulkActionItemsForEnvironment(
|
||||
runtimeEnvironmentId: string,
|
||||
result: CascadeCleanupResult
|
||||
): Promise<void> {
|
||||
await this.#accumulate(result, "bulkActionItem", async () => {
|
||||
const a = await this.#controlPlaneWriter.bulkActionItem.deleteMany({
|
||||
where: { sourceRun: { runtimeEnvironmentId } },
|
||||
});
|
||||
const b = await this.#controlPlaneWriter.bulkActionItem.deleteMany({
|
||||
where: { destinationRun: { runtimeEnvironmentId } },
|
||||
});
|
||||
return a.count + b.count;
|
||||
});
|
||||
}
|
||||
|
||||
// BulkActionItem has no projectId column; clean via both run relations.
|
||||
async #cleanupBulkActionItemsForProject(
|
||||
projectId: string,
|
||||
result: CascadeCleanupResult
|
||||
): Promise<void> {
|
||||
await this.#accumulate(result, "bulkActionItem", async () => {
|
||||
const a = await this.#controlPlaneWriter.bulkActionItem.deleteMany({
|
||||
where: { sourceRun: { projectId } },
|
||||
});
|
||||
const b = await this.#controlPlaneWriter.bulkActionItem.deleteMany({
|
||||
where: { destinationRun: { projectId } },
|
||||
});
|
||||
return a.count + b.count;
|
||||
});
|
||||
}
|
||||
|
||||
// Child-before-parent ordering: an FK-retained DB never errors on an out-of-order delete, and an
|
||||
// FK-dropped DB leaves no orphans. TaskRun self-relations and TaskRun.batchId are SetNull, so a
|
||||
// single deleteMany of all scoped TaskRuns is order-safe within the table; Waitpoint's run/batch
|
||||
// links are SetNull (nullable) so its position is for tidiness only.
|
||||
async #cleanupEnvironmentOnWriter(
|
||||
writer: RunSubgraphCleanupClient,
|
||||
runtimeEnvironmentId: string,
|
||||
result: CascadeCleanupResult
|
||||
): Promise<void> {
|
||||
await this.#accumulate(result, "checkpointRestoreEvent", () =>
|
||||
writer.checkpointRestoreEvent
|
||||
.deleteMany({ where: { runtimeEnvironmentId } })
|
||||
.then((r) => r.count)
|
||||
);
|
||||
await this.#accumulate(result, "checkpoint", () =>
|
||||
writer.checkpoint.deleteMany({ where: { runtimeEnvironmentId } }).then((r) => r.count)
|
||||
);
|
||||
await this.#accumulate(result, "taskRunCheckpoint", () =>
|
||||
writer.taskRunCheckpoint.deleteMany({ where: { runtimeEnvironmentId } }).then((r) => r.count)
|
||||
);
|
||||
// TaskRunWaitpoint has neither an env column nor (on the subset schema) a `taskRun` relation to
|
||||
// filter through, so resolve the scoped run ids first and delete by the scalar `taskRunId`.
|
||||
await this.#accumulate(result, "taskRunWaitpoint", async () => {
|
||||
const runs = await writer.taskRun.findMany({
|
||||
where: { runtimeEnvironmentId },
|
||||
select: { id: true },
|
||||
});
|
||||
if (runs.length === 0) return 0;
|
||||
const r = await writer.taskRunWaitpoint.deleteMany({
|
||||
where: { taskRunId: { in: runs.map((run) => run.id) } },
|
||||
});
|
||||
return r.count;
|
||||
});
|
||||
// Waitpoint's env column is `environmentId`, NOT `runtimeEnvironmentId`.
|
||||
await this.#accumulate(result, "waitpoint", () =>
|
||||
writer.waitpoint
|
||||
.deleteMany({ where: { environmentId: runtimeEnvironmentId } })
|
||||
.then((r) => r.count)
|
||||
);
|
||||
await this.#accumulate(result, "taskRunAttempt", () =>
|
||||
writer.taskRunAttempt.deleteMany({ where: { runtimeEnvironmentId } }).then((r) => r.count)
|
||||
);
|
||||
await this.#accumulate(result, "batchTaskRun", () =>
|
||||
writer.batchTaskRun.deleteMany({ where: { runtimeEnvironmentId } }).then((r) => r.count)
|
||||
);
|
||||
await this.#accumulate(result, "taskRun", () =>
|
||||
writer.taskRun.deleteMany({ where: { runtimeEnvironmentId } }).then((r) => r.count)
|
||||
);
|
||||
}
|
||||
|
||||
async #cleanupProjectOnWriter(
|
||||
writer: RunSubgraphCleanupClient,
|
||||
projectId: string,
|
||||
result: CascadeCleanupResult
|
||||
): Promise<void> {
|
||||
await this.#accumulate(result, "checkpointRestoreEvent", () =>
|
||||
writer.checkpointRestoreEvent.deleteMany({ where: { projectId } }).then((r) => r.count)
|
||||
);
|
||||
await this.#accumulate(result, "checkpoint", () =>
|
||||
writer.checkpoint.deleteMany({ where: { projectId } }).then((r) => r.count)
|
||||
);
|
||||
await this.#accumulate(result, "taskRunCheckpoint", () =>
|
||||
writer.taskRunCheckpoint.deleteMany({ where: { projectId } }).then((r) => r.count)
|
||||
);
|
||||
await this.#accumulate(result, "taskRunWaitpoint", () =>
|
||||
writer.taskRunWaitpoint.deleteMany({ where: { projectId } }).then((r) => r.count)
|
||||
);
|
||||
await this.#accumulate(result, "waitpoint", () =>
|
||||
writer.waitpoint.deleteMany({ where: { projectId } }).then((r) => r.count)
|
||||
);
|
||||
// TaskRunAttempt has no projectId column; clean via its TaskRun relation.
|
||||
await this.#accumulate(result, "taskRunAttempt", () =>
|
||||
writer.taskRunAttempt.deleteMany({ where: { taskRun: { projectId } } }).then((r) => r.count)
|
||||
);
|
||||
// BatchTaskRun has no projectId column; clean via its TaskRun (`runs`) members.
|
||||
await this.#accumulate(result, "batchTaskRun", () =>
|
||||
writer.batchTaskRun
|
||||
.deleteMany({ where: { runs: { some: { projectId } } } })
|
||||
.then((r) => r.count)
|
||||
);
|
||||
await this.#accumulate(result, "taskRun", () =>
|
||||
writer.taskRun.deleteMany({ where: { projectId } }).then((r) => r.count)
|
||||
);
|
||||
}
|
||||
|
||||
async #accumulate(
|
||||
result: CascadeCleanupResult,
|
||||
table: string,
|
||||
run: () => Promise<number>
|
||||
): Promise<void> {
|
||||
const count = await run();
|
||||
result[table] = (result[table] ?? 0) + count;
|
||||
}
|
||||
}
|
||||
@@ -1,660 +0,0 @@
|
||||
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,
|
||||
// run-ops 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);
|
||||
}
|
||||
);
|
||||
});
|
||||
Reference in New Issue
Block a user