diff --git a/apps/webapp/app/services/runs/performRunExecutionV3.server.ts b/apps/webapp/app/services/runs/performRunExecutionV3.server.ts index 6aaff74b7..fb0aa7279 100644 --- a/apps/webapp/app/services/runs/performRunExecutionV3.server.ts +++ b/apps/webapp/app/services/runs/performRunExecutionV3.server.ts @@ -46,7 +46,7 @@ import { forceYieldCoordinator } from "./forceYieldCoordinator.server"; import { ResumeRunService } from "./resumeRun.server"; type FoundRun = NonNullable>>; -type FoundTask = FoundRun["tasks"][number]; +type FoundTask = NonNullable>>[number]; // We need to limit the cached tasks to not be too large >3.5MB when serialized const TOTAL_CACHED_TASK_BYTE_LIMIT = 3500000; @@ -81,6 +81,14 @@ export class PerformRunExecutionV3Service { } public async call(input: PerformRunExecutionV3Input, driftInMs: number = 0) { + logger.debug("PerformRunExecutionV3Service.call", { input, driftInMs }); + + if (Array.isArray(input.id)) { + logger.error("PerformRunExecutionV3Service.call: input.id is an array", { input }); + + throw new Error("input.id must be a string"); + } + const run = await findRun(this.#prismaClient, input.id); if (!run) { @@ -221,11 +229,13 @@ export class PerformRunExecutionV3Service { }); } + const tasks = await getCompletedTasksForRun(this.#prismaClient, run.id); + const sourceContext = RunSourceContextSchema.safeParse(run.event.sourceContext); const executionBody = await this.#createExecutionBody( run, - run.tasks, + tasks, startedAt, false, connections.auth, @@ -1086,53 +1096,6 @@ export class PerformRunExecutionV3Service { return; } - const runWithLatestTask = await tx.jobRun.findUniqueOrThrow({ - where: { - id: run.id, - }, - select: { - tasks: { - select: { - id: true, - name: true, - status: true, - displayKey: true, - }, - take: 1, - orderBy: { createdAt: "desc" }, - }, - _count: { - select: { - tasks: true, - }, - }, - }, - }); - - if (runWithLatestTask._count.tasks === run._count.tasks) { - const latestTask = runWithLatestTask.tasks[0]; - - const cause = - latestTask?.status === "RUNNING" - ? `This is likely caused by task "${ - latestTask.displayKey ?? latestTask.name - }" execution exceeding the function timeout` - : "This is likely caused by executing code outside of a task that exceeded the function timeout"; - - await this.#failRunExecution( - tx, - run, - { - message: `Function timeout detected in ${ - durationInMs / 1000.0 - }s without any task creation. This is unexpected behavior and could lead to an infinite execution error because the run will never finish. ${cause}`, - }, - "TIMED_OUT", - durationInMs - ); - return; - } - await tx.jobRun.update({ where: { id: run.id, @@ -1254,6 +1217,27 @@ function prepareNoOpTasksBloomFilter(possibleTasks: FoundTask[]): string { return filter.serialize(); } +async function getCompletedTasksForRun(prisma: PrismaClientOrTransaction, runId: string) { + return await prisma.task.findMany({ + where: { + runId, + status: "COMPLETED", + }, + select: { + id: true, + idempotencyKey: true, + status: true, + noop: true, + output: true, + outputIsUndefined: true, + parentId: true, + }, + orderBy: { + id: "asc", + }, + }); +} + async function findRun(prisma: PrismaClientOrTransaction, id: string) { return await prisma.jobRun.findUnique({ where: { id }, @@ -1278,25 +1262,6 @@ async function findRun(prisma: PrismaClientOrTransaction, id: string) { }, }, }, - tasks: { - where: { - status: { - in: ["COMPLETED"], - }, - }, - select: { - id: true, - idempotencyKey: true, - status: true, - noop: true, - output: true, - outputIsUndefined: true, - parentId: true, - }, - orderBy: { - id: "asc", - }, - }, event: true, version: { include: { @@ -1309,11 +1274,6 @@ async function findRun(prisma: PrismaClientOrTransaction, id: string) { recipientMethod: "ENDPOINT", }, }, - _count: { - select: { - tasks: true, - }, - }, }, }); }