Skip to content

Commit f3c9d90

Browse files
committed
fix(webapp,run-store): correct run-ops routing in finalization, resume, and edge cleanup
- Skip the legacy attempt-create when a run-ops-resident run reaches the no-attempt branch during finalization; it is already finalized on its own database, so minting a legacy attempt would orphan a row on the wrong one. - Cancel and finalize the same run by taking the run id from the fetched attempt rather than a caller-supplied id. - When resuming dependent parents, read the owning primary for the run and attempts just finalized in the same operation so replica lag cannot skip a resume, and stop when the environment cannot be resolved instead of treating it as production. - Delete run/waitpoint edges from both stores so a cross-database edge cannot leave a run blocked. - Drop a redundant read client on the bulk-action source-run lookup so it uses the residency-routed replica.
1 parent eef1801 commit f3c9d90

5 files changed

Lines changed: 67 additions & 35 deletions

File tree

apps/webapp/app/v3/services/bulk/performBulkAction.server.ts

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,8 @@ export class PerformBulkActionService extends BaseService {
2727
}
2828

2929
// Fetch the source run through the store (it may reside in a different DB than the item).
30-
const sourceRun = await this.runStore.findRunOrThrow({ id: item.sourceRunId }, this._prisma);
30+
// No client: routing keys off the run id, and this pre-existing run needs no read-your-writes.
31+
const sourceRun = await this.runStore.findRunOrThrow({ id: item.sourceRunId });
3132

3233
switch (item.type) {
3334
case "REPLAY": {

apps/webapp/app/v3/services/cancelAttempt.server.ts

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -71,7 +71,7 @@ export class CancelAttemptService extends BaseService {
7171

7272
const finalizeService = new FinalizeTaskRunService();
7373
await finalizeService.call({
74-
id: taskRunId,
74+
id: taskRunAttempt.taskRun.id,
7575
status: isCancellable ? "INTERRUPTED" : undefined,
7676
completedAt: isCancellable ? cancelledAt : undefined,
7777
attemptStatus: isCancellable ? "CANCELED" : undefined,

apps/webapp/app/v3/services/finalizeTaskRun.server.ts

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
import { type FlushedRunMetadata, type TaskRunError, sanitizeError } from "@trigger.dev/core/v3";
2+
import { ownerEngine } from "@trigger.dev/core/v3/isomorphic";
23
import { type Prisma, type TaskRun } from "@trigger.dev/database";
34
import { runOpsLegacyPrisma } from "~/db.server";
45
import { findQueueInEnvironment } from "~/models/taskQueue.server";
@@ -314,6 +315,17 @@ export class FinalizeTaskRunService extends BaseService {
314315
error,
315316
});
316317

318+
// TaskRunAttempt is V1-residual, minted only for legacy (cuid) runs on the legacy client below.
319+
// A NEW-resident run reaching here (via CompleteAttemptService's fan-out lookup) is already
320+
// finalized on its owning store above, so skip the orphan legacy attempt.
321+
if (ownerEngine(run.id) === "NEW") {
322+
logger.error(
323+
"FinalizeTaskRunService: reached no-attempt branch for a run-ops (NEW-resident) run; skipping legacy attempt-create",
324+
{ runId: run.id, status: run.status }
325+
);
326+
return;
327+
}
328+
317329
if (!run.lockedById) {
318330
// This happens when a run is expired or was cancelled before an attempt, it's not a problem
319331
logger.info(

apps/webapp/app/v3/services/resumeDependentParents.server.ts

Lines changed: 49 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -55,11 +55,14 @@ type Dependency = NonNullable<RunWithDependency["dependency"]>;
5555
export class ResumeDependentParentsService extends BaseService {
5656
public async call({ id }: { id: string }): Promise<Output> {
5757
try {
58+
// Read-your-writes: the caller just committed this run's final status, so read the owning
59+
// primary (writer) — a lagging replica could show a non-final status and skip resumption.
5860
const run = await this.runStore.findRun(
5961
{ id },
6062
{
6163
select: runWithDependencySelect,
62-
}
64+
},
65+
this._prisma
6366
);
6467

6568
const dependency = run?.dependency ?? null;
@@ -83,7 +86,21 @@ export class ResumeDependentParentsService extends BaseService {
8386

8487
const environment = await findEnvironmentById(run.runtimeEnvironmentId);
8588

86-
if (environment?.type === "DEVELOPMENT") {
89+
if (!environment) {
90+
// Bail rather than fall through: an unresolved env would otherwise be treated as
91+
// production and mutate dependency state.
92+
logger.error("ResumeDependentParentsService: environment not found", {
93+
runId: id,
94+
runtimeEnvironmentId: run.runtimeEnvironmentId,
95+
});
96+
97+
return {
98+
success: false,
99+
error: `Environment not found for run ${id}`,
100+
};
101+
}
102+
103+
if (environment.type === "DEVELOPMENT") {
87104
return {
88105
success: true,
89106
action: "dev",
@@ -137,18 +154,22 @@ export class ResumeDependentParentsService extends BaseService {
137154
}
138155
);
139156

140-
const lastAttempt = await this.runStore.findTaskRunAttempt({
141-
select: {
142-
id: true,
143-
status: true,
144-
},
145-
where: {
146-
taskRunId: dependency.taskRunId,
147-
},
148-
orderBy: {
149-
id: "desc",
157+
// Read-your-writes: the just-finalized attempt was written by the same finalize operation.
158+
const lastAttempt = await this.runStore.findTaskRunAttempt(
159+
{
160+
select: {
161+
id: true,
162+
status: true,
163+
},
164+
where: {
165+
taskRunId: dependency.taskRunId,
166+
},
167+
orderBy: {
168+
id: "desc",
169+
},
150170
},
151-
});
171+
this._prisma
172+
);
152173

153174
if (!lastAttempt) {
154175
logger.error(
@@ -210,18 +231,22 @@ export class ResumeDependentParentsService extends BaseService {
210231
};
211232
}
212233

213-
const lastAttempt = await this.runStore.findTaskRunAttempt({
214-
select: {
215-
id: true,
216-
status: true,
217-
},
218-
where: {
219-
taskRunId: dependency.taskRunId,
220-
},
221-
orderBy: {
222-
id: "desc",
234+
// Read-your-writes: the just-finalized attempt was written by the same finalize operation.
235+
const lastAttempt = await this.runStore.findTaskRunAttempt(
236+
{
237+
select: {
238+
id: true,
239+
status: true,
240+
},
241+
where: {
242+
taskRunId: dependency.taskRunId,
243+
},
244+
orderBy: {
245+
id: "desc",
246+
},
223247
},
224-
});
248+
this._prisma
249+
);
225250

226251
if (!lastAttempt) {
227252
logger.error(

internal-packages/run-store/src/runOpsStore.ts

Lines changed: 3 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1362,15 +1362,9 @@ export class RoutingRunStore implements RunStore {
13621362
args: Prisma.TaskRunWaitpointDeleteManyArgs,
13631363
tx?: PrismaClientOrTransaction
13641364
): Promise<Prisma.BatchPayload> {
1365-
const where = args.where as { waitpointId?: unknown } | undefined;
1366-
const waitpointId = typeof where?.waitpointId === "string" ? where.waitpointId : undefined;
1367-
if (waitpointId !== undefined) {
1368-
const store = await this.#resolveWaitpointStore(waitpointId);
1369-
return store.deleteManyTaskRunWaitpoints(args, undefined);
1370-
}
1371-
// Keyed by taskRunId (or other): a run's edges may straddle DBs mid-drain, so delete from both.
1372-
// The caller's `tx` is never forwarded — each leg deletes on its own store's client, so neither
1373-
// leg is tied to a caller-supplied transaction.
1365+
// Edges co-locate with their RUN, not their waitpoint, so a waitpointId/taskRunId predicate may
1366+
// match edges on either store; delete from both so no cross-DB edge keeps a run blocked. The
1367+
// caller's `tx` is never forwarded — each leg deletes on its own store's client.
13741368
const [fromNew, fromLegacy] = await Promise.all([
13751369
this.#new.deleteManyTaskRunWaitpoints(args),
13761370
this.#legacy.deleteManyTaskRunWaitpoints(args),

0 commit comments

Comments
 (0)