Skip to content

Commit 7432ca5

Browse files
d-csTrigger.dev RepoOps
authored andcommitted
fix(webapp): read a gen-2 batch row from its shard in the batch results API
Fix the batch results API returning 404 for batches whose id routes to a run-ops shard. The batch row is now read from the shard the id names, matching how batch retrieve already finds it. Adds a regression test for the sharded case. Mono-RevId: 26a6a2f1c43ebcbff5fb9820259c4eea906bb32d
1 parent d06ee58 commit 7432ca5

4 files changed

Lines changed: 106 additions & 40 deletions

File tree

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
---
2+
area: webapp
3+
type: fix
4+
---
5+
6+
Fix the batch results API returning "Batch not found" for some batches that the batch retrieve API could see.

‎apps/webapp/app/presenters/v3/ApiBatchResultsPresenter.server.ts‎

Lines changed: 29 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -85,12 +85,20 @@ export class ApiBatchResultsPresenter extends BasePresenter {
8585
});
8686
}
8787

88-
// Passthrough: batch row off the replica, members via the single run store. No legacy read.
89-
async #callPassthrough(
88+
// A gen-2 batch id lives only on its shard: neither the passthrough replica nor the gen-1
89+
// new/legacy probes hold the row, so both paths route the row read by the id first.
90+
#shardReplicaFor(friendlyId: string): PrismaReplicaClient | undefined {
91+
const shardKey = resolveShard(friendlyId);
92+
if (shardKey === "new" || shardKey === "legacy") return undefined;
93+
return (this.readThrough?.shardReplicas ?? runOpsShardReplicas).get(shardKey);
94+
}
95+
96+
#readBatchRow(
97+
client: PrismaClientOrTransaction,
9098
friendlyId: string,
9199
env: AuthenticatedEnvironment
92-
): Promise<BatchTaskRunExecutionResult | undefined> {
93-
const batchRun = await this._replica.batchTaskRun.findFirst({
100+
) {
101+
return client.batchTaskRun.findFirst({
94102
where: {
95103
friendlyId,
96104
runtimeEnvironmentId: env.id,
@@ -103,6 +111,18 @@ export class ApiBatchResultsPresenter extends BasePresenter {
103111
},
104112
},
105113
});
114+
}
115+
116+
// Passthrough: batch row off the replica, members via the single run store. No legacy read.
117+
async #callPassthrough(
118+
friendlyId: string,
119+
env: AuthenticatedEnvironment
120+
): Promise<BatchTaskRunExecutionResult | undefined> {
121+
const batchRun = await this.#readBatchRow(
122+
this.#shardReplicaFor(friendlyId) ?? this._replica,
123+
friendlyId,
124+
env
125+
);
106126

107127
if (!batchRun) {
108128
return undefined;
@@ -150,26 +170,12 @@ export class ApiBatchResultsPresenter extends BasePresenter {
150170
const newClient = (this.readThrough?.newClient ?? this._replica) as PrismaReplicaClient;
151171
const legacyReplica = (this.readThrough?.legacyReplica ?? this._replica) as PrismaReplicaClient;
152172

153-
const readBatch = (client: PrismaClientOrTransaction) =>
154-
client.batchTaskRun.findFirst({
155-
where: {
156-
friendlyId,
157-
runtimeEnvironmentId: env.id,
158-
},
159-
include: {
160-
items: {
161-
select: {
162-
taskRunId: true,
163-
},
164-
},
165-
},
166-
});
167-
168-
let batchRun = await readBatch(newClient);
173+
const batchShard = this.#shardReplicaFor(friendlyId);
174+
let batchRun = await this.#readBatchRow(batchShard ?? newClient, friendlyId, env);
169175

170-
// Legacy READ REPLICA probe, only on a new-probe miss; skipped when past retention.
171-
if (!batchRun && !this.readThrough?.isPastRetention?.(friendlyId)) {
172-
batchRun = await readBatch(legacyReplica);
176+
// Legacy READ REPLICA probe, only on a gen-1 new-probe miss; skipped when past retention.
177+
if (!batchRun && !batchShard && !this.readThrough?.isPastRetention?.(friendlyId)) {
178+
batchRun = await this.#readBatchRow(legacyReplica, friendlyId, env);
173179
}
174180

175181
if (!batchRun) {

‎apps/webapp/app/v3/runOpsMigration/track1-baseline.json‎

Lines changed: 8 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -68,37 +68,28 @@
6868
"WaitpointTag.project"
6969
],
7070
"totals": {
71-
"violations": 5,
72-
"detectorI": 5,
71+
"violations": 4,
72+
"detectorI": 4,
7373
"detectorII": 0,
7474
"detectorIII": 0,
7575
"write": 0,
76-
"read": 5,
76+
"read": 4,
7777
"files": 1,
7878
"legacyAnnotations": 0
7979
},
8080
"violations": [
8181
{
8282
"file": "apps/webapp/app/presenters/v3/ApiBatchResultsPresenter.server.ts",
83-
"line": 93,
83+
"line": 101,
8484
"model": "BatchTaskRun",
8585
"delegate": "batchTaskRun",
8686
"callKind": "read",
8787
"detector": "i",
88-
"snippet": "const batchRun = await this._replica.batchTaskRun.findFirst({"
88+
"snippet": "return client.batchTaskRun.findFirst({"
8989
},
9090
{
9191
"file": "apps/webapp/app/presenters/v3/ApiBatchResultsPresenter.server.ts",
92-
"line": 154,
93-
"model": "BatchTaskRun",
94-
"delegate": "batchTaskRun",
95-
"callKind": "read",
96-
"detector": "i",
97-
"snippet": "client.batchTaskRun.findFirst({"
98-
},
99-
{
100-
"file": "apps/webapp/app/presenters/v3/ApiBatchResultsPresenter.server.ts",
101-
"line": 214,
92+
"line": 220,
10293
"model": "TaskRun",
10394
"delegate": "taskRun",
10495
"callKind": "read",
@@ -107,7 +98,7 @@
10798
},
10899
{
109100
"file": "apps/webapp/app/presenters/v3/ApiBatchResultsPresenter.server.ts",
110-
"line": 224,
101+
"line": 230,
111102
"model": "TaskRun",
112103
"delegate": "taskRun",
113104
"callKind": "read",
@@ -116,7 +107,7 @@
116107
},
117108
{
118109
"file": "apps/webapp/app/presenters/v3/ApiBatchResultsPresenter.server.ts",
119-
"line": 241,
110+
"line": 247,
120111
"model": "TaskRun",
121112
"delegate": "taskRun",
122113
"callKind": "read",

‎apps/webapp/test/apiBatchResultsPresenter.dedicatedSeam.test.ts‎

Lines changed: 63 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ import { heteroRunOpsPostgresTest, makeNShardRunOpsPostgresTest } from "@interna
77
import type { RunOpsPrismaClient } from "@internal/run-ops-database";
88
import type { PrismaClient } from "@trigger.dev/database";
99
import { generateRunOpsId, generateRunOpsIdV2 } from "@trigger.dev/core/v3/isomorphic";
10+
import { PostgresRunStore } from "@internal/run-store";
1011
import { describe, expect, vi } from "vitest";
1112
import type { PrismaReplicaClient } from "~/db.server";
1213
import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
@@ -475,4 +476,66 @@ describe("ApiBatchResultsPresenter split mode — real run-ops dedicated schema
475476
},
476477
180_000
477478
);
479+
480+
// A gen-2 BATCH id is minted onto its shard, so the batch row (not just its members) lives
481+
// only there. Both read paths must route the row by its own id; before this, the gen-1
482+
// probes missed and the endpoint 404'd for a batch that `batch.retrieve` found.
483+
for (const splitEnabled of [true, false]) {
484+
oneShardTest(
485+
`a gen-2 batch whose row lives only on its shard returns its members (split ${
486+
splitEnabled ? "on" : "off"
487+
})`,
488+
async ({ legacyPrisma, newPrisma, shardPrismas }) => {
489+
const shardPrisma = shardPrismas[0]!;
490+
const ctx = await seedLegacyEnv(legacyPrisma, `gen2-batch-row-${splitEnabled}`);
491+
492+
const memberIds = [generateRunOpsIdV2("a"), generateRunOpsIdV2("a")];
493+
for (const [i, id] of memberIds.entries()) {
494+
await seedNewMember(
495+
shardPrisma,
496+
{ envId: ctx.environment.id, orgId: ctx.organization.id, projectId: ctx.project.id },
497+
{
498+
id,
499+
friendlyId: `run_shard_${i}`,
500+
status: "COMPLETED_SUCCESSFULLY",
501+
output: JSON.stringify({ i }),
502+
}
503+
);
504+
}
505+
506+
const batchFriendlyId = `batch_${generateRunOpsIdV2("a")}`;
507+
await seedBatchOnNew(shardPrisma, ctx.environment.id, batchFriendlyId, memberIds);
508+
509+
const shardReplicas = new Map([["a", shardPrisma as unknown as PrismaReplicaClient]]);
510+
const presenter = splitEnabled
511+
? new ApiBatchResultsPresenter(throwingPrisma, throwingPrisma, {
512+
splitEnabled: true,
513+
newClient: newPrisma as unknown as PrismaReplicaClient,
514+
legacyReplica: legacyPrisma as unknown as PrismaReplicaClient,
515+
shardReplicas,
516+
})
517+
: // Passthrough hydrates members through the primary handle; the REPLICA (where the row
518+
// is read without the fix) is the gen-1 store, which does not hold this batch.
519+
new ApiBatchResultsPresenter(
520+
shardPrisma as unknown as PrismaReplicaClient,
521+
newPrisma as unknown as PrismaReplicaClient,
522+
{ splitEnabled: false, shardReplicas },
523+
new PostgresRunStore({
524+
prisma: shardPrisma as any,
525+
readOnlyPrisma: shardPrisma as any,
526+
})
527+
);
528+
529+
const result = await presenter.call(batchFriendlyId, env(ctx));
530+
531+
expect(result).toBeDefined();
532+
expect(result!.id).toBe(batchFriendlyId);
533+
expect(result!.items.map((item) => (item as { output?: string }).output)).toEqual([
534+
JSON.stringify({ i: 0 }),
535+
JSON.stringify({ i: 1 }),
536+
]);
537+
},
538+
180_000
539+
);
540+
}
478541
});

0 commit comments

Comments
 (0)