From 34ca7667d354d9bd43dab52ed35587d4990dab13 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Wed, 26 Jun 2024 15:22:35 +0100 Subject: [PATCH] Various perf improvements to prevent event loop lag (#1186) * WIP * Handle tasks that have failed but are being auto yielded * Limit trace view to 25k event records, add a download run logs button Also added two new indexes to TaskEvent: ``` /// Used on eventRepository.getTraceSummary() @@index([traceId, startTime]) // Used for getting all logs for a run @@index([runId]) ``` * perf improvements on eventRepository.getSpan() * v2: Add a 5 minute timeout for run execution requests in dev * v3: Include presigned urls for downloading large payloads and outputs when using runs.retrieve * v3: better handle large task payloads and outputs * Change to 512KB * v2: paginate trigger schedules endpoint * v3: add 3MB limit on batch and single payloads * Update task payload and output limits --- .changeset/thick-carrots-sneeze.md | 6 + apps/webapp/app/env.server.ts | 3 + .../app/presenters/RunListPresenter.server.ts | 4 - .../ScheduledTriggersPresenter.server.ts | 98 +- .../v3/ApiRetrieveRunPresenter.server.ts | 33 +- .../route.tsx | 26 +- apps/webapp/app/routes/api.v1.packets.$.ts | 75 +- .../CompleteRunTaskService.server.ts | 124 +- .../route.ts | 53 +- .../FailRunTaskService.server.ts | 4 - .../routes/api.v1.runs.$runId.tasks/route.ts | 42 +- .../app/routes/api.v1.tasks.$taskId.batch.ts | 7 + .../routes/api.v1.tasks.$taskId.trigger.ts | 35 +- .../route.tsx | 26 +- .../resources.packets.$environmentId.$.ts | 33 +- .../resources.runs.$runParam.logs.download.ts | 112 + .../webapp/app/services/endpointApi.server.ts | 17 +- .../runs/performRunExecutionV3.server.ts | 31 +- .../app/services/tasks/runTask.server.ts | 255 +- .../app/utils/parseRequestJson.server.ts | 29 + apps/webapp/app/utils/pathBuilder.ts | 4 + apps/webapp/app/v3/eventRepository.server.ts | 316 +- apps/webapp/app/v3/r2.server.ts | 92 +- .../app/v3/services/triggerTask.server.ts | 36 +- apps/webapp/app/v3/tracer.server.ts | 4 +- docs/v3-openapi.yaml | 8 + docs/v3/limits.mdx | 10 + docs/v3/triggering.mdx | 83 + packages/core/package.json | 8 + packages/core/src/v3/schemas/api.ts | 2 + packages/core/src/v3/utils/ioSerialization.ts | 11 +- packages/core/tsup.config.ts | 1 + .../migration.sql | 2 + .../migration.sql | 2 + packages/database/prisma/schema.prisma | 4 + references/job-catalog/fixtures/large.json | 70010 +++++++ references/job-catalog/fixtures/toolarge.json | 140022 +++++++++++++++ references/job-catalog/src/stressTest.ts | 37 +- references/v3-catalog/package.json | 3 +- references/v3-catalog/src/env.ts | 32 - .../v3-catalog/src/trigger/concurrency.ts | 4 +- .../v3-catalog/src/trigger/longRunning.ts | 19 + references/v3-catalog/src/trigger/simple.ts | 9 + references/v3-catalog/src/trigger/subtasks.ts | 42 + .../v3-catalog/src/triggerWithLargePayload.ts | 37 + references/v3-catalog/tsconfig.json | 4 +- 46 files changed, 211316 insertions(+), 499 deletions(-) create mode 100644 .changeset/thick-carrots-sneeze.md create mode 100644 apps/webapp/app/routes/resources.runs.$runParam.logs.download.ts create mode 100644 apps/webapp/app/utils/parseRequestJson.server.ts create mode 100644 packages/database/prisma/migrations/20240625091632_add_trace_id_start_time_index_on_task_events/migration.sql create mode 100644 packages/database/prisma/migrations/20240625095006_add_run_id_index_to_task_events/migration.sql create mode 100644 references/job-catalog/fixtures/large.json create mode 100644 references/job-catalog/fixtures/toolarge.json delete mode 100644 references/v3-catalog/src/env.ts create mode 100644 references/v3-catalog/src/triggerWithLargePayload.ts diff --git a/.changeset/thick-carrots-sneeze.md b/.changeset/thick-carrots-sneeze.md new file mode 100644 index 000000000..9e82d0a56 --- /dev/null +++ b/.changeset/thick-carrots-sneeze.md @@ -0,0 +1,6 @@ +--- +"@trigger.dev/core": patch +"@trigger.dev/sdk": patch +--- + +v3: Include presigned urls for downloading large payloads and outputs when using runs.retrieve diff --git a/apps/webapp/app/env.server.ts b/apps/webapp/app/env.server.ts index ff0a30b9c..9552aac4a 100644 --- a/apps/webapp/app/env.server.ts +++ b/apps/webapp/app/env.server.ts @@ -206,6 +206,9 @@ const EnvironmentSchema = z.object({ USAGE_OPEN_METER_BASE_URL: z.string().optional(), EVENT_LOOP_MONITOR_ENABLED: z.string().default("1"), MAXIMUM_LIVE_RELOADING_EVENTS: z.coerce.number().int().default(1000), + MAXIMUM_TRACE_SUMMARY_VIEW_COUNT: z.coerce.number().int().default(25_000), + TASK_PAYLOAD_OFFLOAD_THRESHOLD: z.coerce.number().int().default(524_288), // 512KB + TASK_PAYLOAD_MAXIMUM_SIZE: z.coerce.number().int().default(3_145_728), // 3MB }); export type Environment = z.infer; diff --git a/apps/webapp/app/presenters/RunListPresenter.server.ts b/apps/webapp/app/presenters/RunListPresenter.server.ts index cb2be6f5e..91c1e5a98 100644 --- a/apps/webapp/app/presenters/RunListPresenter.server.ts +++ b/apps/webapp/app/presenters/RunListPresenter.server.ts @@ -1,11 +1,9 @@ -import { z } from "zod"; import { Direction, FilterableEnvironment, FilterableStatus, filterableStatuses, } from "~/components/runs/RunStatuses"; -import { PrismaClient, prisma } from "~/db.server"; import { getUsername } from "~/utils/username"; import { BasePresenter } from "./v3/basePresenter.server"; @@ -29,8 +27,6 @@ const DEFAULT_PAGE_SIZE = 20; export type RunList = Awaited>; export class RunListPresenter extends BasePresenter { - - public async call({ userId, eventId, diff --git a/apps/webapp/app/presenters/ScheduledTriggersPresenter.server.ts b/apps/webapp/app/presenters/ScheduledTriggersPresenter.server.ts index 6bd59d5d4..bf72bdd67 100644 --- a/apps/webapp/app/presenters/ScheduledTriggersPresenter.server.ts +++ b/apps/webapp/app/presenters/ScheduledTriggersPresenter.server.ts @@ -1,27 +1,52 @@ -import { User } from "@trigger.dev/database"; import { ScheduleMetadataSchema } from "@trigger.dev/core"; -import { PrismaClient, prisma } from "~/db.server"; +import { User } from "@trigger.dev/database"; import { Organization } from "~/models/organization.server"; import { Project } from "~/models/project.server"; import { calculateNextScheduledEvent } from "~/services/schedules/nextScheduledEvent.server"; +import { BasePresenter } from "./v3/basePresenter.server"; -export class ScheduledTriggersPresenter { - #prismaClient: PrismaClient; - - constructor(prismaClient: PrismaClient = prisma) { - this.#prismaClient = prismaClient; - } +const DEFAULT_PAGE_SIZE = 20; +export class ScheduledTriggersPresenter extends BasePresenter { public async call({ userId, projectSlug, organizationSlug, + direction = "forward", + pageSize = DEFAULT_PAGE_SIZE, + cursor, }: { userId: User["id"]; projectSlug: Project["slug"]; organizationSlug: Organization["slug"]; + direction?: "forward" | "backward"; + pageSize?: number; + cursor?: string; }) { - const scheduled = await this.#prismaClient.scheduleSource.findMany({ + const organization = await this._replica.organization.findFirstOrThrow({ + select: { + id: true, + }, + where: { + slug: organizationSlug, + members: { some: { userId } }, + }, + }); + + // Find the project scoped to the organization + const project = await this._replica.project.findFirstOrThrow({ + select: { + id: true, + }, + where: { + slug: projectSlug, + organizationId: organization.id, + }, + }); + + const directionMultiplier = direction === "forward" ? 1 : -1; + + const scheduled = await this._replica.scheduleSource.findMany({ select: { id: true, key: true, @@ -50,23 +75,50 @@ export class ScheduledTriggersPresenter { }, }, ], - organization: { - slug: organizationSlug, - members: { - some: { - userId, - }, - }, - }, - project: { - slug: projectSlug, - }, + projectId: project.id, }, }, + orderBy: [{ id: "desc" }], + //take an extra record to tell if there are more + take: directionMultiplier * (pageSize + 1), + //skip the cursor if there is one + skip: cursor ? 1 : 0, + cursor: cursor + ? { + id: cursor, + } + : undefined, }); + const hasMore = scheduled.length > pageSize; + + //get cursors for next and previous pages + let next: string | undefined; + let previous: string | undefined; + switch (direction) { + case "forward": + previous = cursor ? scheduled.at(0)?.id : undefined; + if (hasMore) { + next = scheduled[pageSize - 1]?.id; + } + break; + case "backward": + if (hasMore) { + previous = scheduled[1]?.id; + next = scheduled[pageSize]?.id; + } else { + next = scheduled[pageSize - 1]?.id; + } + break; + } + + const scheduledToReturn = + direction === "backward" && hasMore + ? scheduled.slice(1, pageSize + 1) + : scheduled.slice(0, pageSize); + return { - scheduled: scheduled.map((s) => { + scheduled: scheduledToReturn.map((s) => { const schedule = ScheduleMetadataSchema.parse(s.schedule); const nextEventTimestamp = s.active ? calculateNextScheduledEvent(schedule, s.lastEventTimestamp) @@ -78,6 +130,10 @@ export class ScheduledTriggersPresenter { nextEventTimestamp, }; }), + pagination: { + next, + previous, + }, }; } } diff --git a/apps/webapp/app/presenters/v3/ApiRetrieveRunPresenter.server.ts b/apps/webapp/app/presenters/v3/ApiRetrieveRunPresenter.server.ts index 7e8e05d54..1129810c6 100644 --- a/apps/webapp/app/presenters/v3/ApiRetrieveRunPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/ApiRetrieveRunPresenter.server.ts @@ -12,6 +12,7 @@ import { import { Prisma, TaskRunAttemptStatus, TaskRunStatus } from "@trigger.dev/database"; import assertNever from "assert-never"; import { AuthenticatedEnvironment } from "~/services/apiAuth.server"; +import { generatePresignedUrl } from "~/v3/r2.server"; import { BasePresenter } from "./basePresenter.server"; export class ApiRetrieveRunPresenter extends BasePresenter { @@ -44,7 +45,9 @@ export class ApiRetrieveRunPresenter extends BasePresenter { } let $payload: any; + let $payloadPresignedUrl: string | undefined; let $output: any; + let $outputPresignedUrl: string | undefined; if (showSecretDetails) { const payloadPacket = await conditionallyImportPacket({ @@ -52,7 +55,19 @@ export class ApiRetrieveRunPresenter extends BasePresenter { dataType: taskRun.payloadType, }); - $payload = await parsePacket(payloadPacket); + if ( + payloadPacket.dataType === "application/store" && + typeof payloadPacket.data === "string" + ) { + $payloadPresignedUrl = await generatePresignedUrl( + env.project.externalRef, + env.slug, + payloadPacket.data, + "GET" + ); + } else { + $payload = await parsePacket(payloadPacket); + } if (taskRun.status === "COMPLETED_SUCCESSFULLY") { const completedAttempt = taskRun.attempts.find( @@ -65,7 +80,19 @@ export class ApiRetrieveRunPresenter extends BasePresenter { dataType: completedAttempt.outputType, }); - $output = await parsePacket(outputPacket); + if ( + outputPacket.dataType === "application/store" && + typeof outputPacket.data === "string" + ) { + $outputPresignedUrl = await generatePresignedUrl( + env.project.externalRef, + env.slug, + outputPacket.data, + "GET" + ); + } else { + $output = await parsePacket(outputPacket); + } } } } @@ -85,7 +112,9 @@ export class ApiRetrieveRunPresenter extends BasePresenter { ? taskRun.updatedAt : undefined, payload: $payload, + payloadPresignedUrl: $payloadPresignedUrl, output: $output, + outputPresignedUrl: $outputPresignedUrl, isTest: taskRun.isTest, schedule: taskRun.schedule ? { diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.triggers.scheduled/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.triggers.scheduled/route.tsx index e9cadb62f..245bd7392 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.triggers.scheduled/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.triggers.scheduled/route.tsx @@ -2,6 +2,8 @@ import { NoSymbolIcon } from "@heroicons/react/20/solid"; import { CheckCircleIcon, XCircleIcon } from "@heroicons/react/24/solid"; import type { LoaderFunctionArgs } from "@remix-run/server-runtime"; import { typedjson, useTypedLoaderData } from "remix-typedjson"; +import { z } from "zod"; +import { ListPagination } from "~/components/ListPagination"; import { EnvironmentLabel } from "~/components/environments/EnvironmentLabel"; import { DateTime } from "~/components/primitives/DateTime"; import { LabelValueStack } from "~/components/primitives/LabelValueStack"; @@ -17,30 +19,38 @@ import { TableRow, } from "~/components/primitives/Table"; import { TextLink } from "~/components/primitives/TextLink"; -import { useOrganization } from "~/hooks/useOrganizations"; -import { useProject } from "~/hooks/useProject"; +import { DirectionSchema } from "~/components/runs/RunStatuses"; import { ScheduledTriggersPresenter } from "~/presenters/ScheduledTriggersPresenter.server"; import { requireUserId } from "~/services/session.server"; import { ProjectParamSchema, docsPath } from "~/utils/pathBuilder"; +const SearchSchema = z.object({ + cursor: z.string().optional(), + direction: DirectionSchema.optional(), +}); + export const loader = async ({ request, params }: LoaderFunctionArgs) => { const userId = await requireUserId(request); const { organizationSlug, projectParam } = ProjectParamSchema.parse(params); + const url = new URL(request.url); + const s = Object.fromEntries(url.searchParams.entries()); + const searchParams = SearchSchema.parse(s); + const presenter = new ScheduledTriggersPresenter(); const data = await presenter.call({ userId, organizationSlug, projectSlug: projectParam, + direction: searchParams.direction, + cursor: searchParams.cursor, }); return typedjson(data); }; -export default function Integrations() { - const { scheduled } = useTypedLoaderData(); - const organization = useOrganization(); - const project = useProject(); +export default function Route() { + const { scheduled, pagination } = useTypedLoaderData(); return ( <> @@ -49,6 +59,10 @@ export default function Integrations() { expression or an interval. + {scheduled.length > 0 && ( + + )} + diff --git a/apps/webapp/app/routes/api.v1.packets.$.ts b/apps/webapp/app/routes/api.v1.packets.$.ts index eee457be9..9cb72d30e 100644 --- a/apps/webapp/app/routes/api.v1.packets.$.ts +++ b/apps/webapp/app/routes/api.v1.packets.$.ts @@ -1,10 +1,8 @@ import type { ActionFunctionArgs } from "@remix-run/server-runtime"; import { json } from "@remix-run/server-runtime"; import { z } from "zod"; -import { env } from "~/env.server"; import { authenticateApiRequest } from "~/services/apiAuth.server"; -import { logger } from "~/services/logger.server"; -import { r2 } from "~/v3/r2.server"; +import { generatePresignedUrl } from "~/v3/r2.server"; const ParamsSchema = z.object({ "*": z.string(), @@ -26,34 +24,19 @@ export async function action({ request, params }: ActionFunctionArgs) { const parsedParams = ParamsSchema.parse(params); const filename = parsedParams["*"]; - if (!env.OBJECT_STORE_BASE_URL) { - return json({ error: "Object store base URL is not set" }, { status: 500 }); - } - - if (!r2) { - return json({ error: "Object store credentials are not set" }, { status: 500 }); - } - - const url = new URL(env.OBJECT_STORE_BASE_URL); - url.pathname = `/packets/${authenticationResult.environment.project.externalRef}/${authenticationResult.environment.slug}/${filename}`; - url.searchParams.set("X-Amz-Expires", "300"); // 5 minutes - - const signed = await r2.sign( - new Request(url, { - method: "PUT", - }), - { - aws: { signQuery: true }, - } + const presignedUrl = await generatePresignedUrl( + authenticationResult.environment.project.externalRef, + authenticationResult.environment.slug, + filename, + "PUT" ); - logger.debug("Generated presigned URL", { - url: signed.url, - headers: Object.fromEntries(signed.headers), - }); + if (!presignedUrl) { + return json({ error: "Failed to generate presigned URL" }, { status: 500 }); + } // Caller can now use this URL to upload to that object. - return json({ presignedUrl: signed.url }); + return json({ presignedUrl }); } export async function loader({ request, params }: ActionFunctionArgs) { @@ -67,35 +50,17 @@ export async function loader({ request, params }: ActionFunctionArgs) { const parsedParams = ParamsSchema.parse(params); const filename = parsedParams["*"]; - if (!env.OBJECT_STORE_BASE_URL) { - return json({ error: "Object store base URL is not set" }, { status: 500 }); - } - - if (!r2) { - return json({ error: "Object store credentials are not set" }, { status: 500 }); - } - - const url = new URL(env.OBJECT_STORE_BASE_URL); - url.pathname = `/packets/${authenticationResult.environment.project.externalRef}/${authenticationResult.environment.slug}/${filename}`; - url.searchParams.set("X-Amz-Expires", "300"); // 5 minutes - - const signed = await r2.sign( - new Request(url, { - method: request.method, - }), - { - aws: { signQuery: true }, - } + const presignedUrl = await generatePresignedUrl( + authenticationResult.environment.project.externalRef, + authenticationResult.environment.slug, + filename, + "GET" ); - logger.debug("Generated presigned URL", { - url: signed.url, - headers: Object.fromEntries(signed.headers), - }); + if (!presignedUrl) { + return json({ error: "Failed to generate presigned URL" }, { status: 500 }); + } - const getUrl = new URL(url.href); - getUrl.searchParams.delete("X-Amz-Expires"); - - // Caller can now use this URL to upload to that object. - return json({ presignedUrl: signed.url }); + // Caller can now use this URL to fetch that object. + return json({ presignedUrl }); } diff --git a/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.complete/CompleteRunTaskService.server.ts b/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.complete/CompleteRunTaskService.server.ts index 0937bd9ac..eafbacd10 100644 --- a/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.complete/CompleteRunTaskService.server.ts +++ b/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.complete/CompleteRunTaskService.server.ts @@ -3,6 +3,7 @@ import { PrismaClientOrTransaction, prisma } from "~/db.server"; import { taskWithAttemptsToServerTask } from "~/models/task.server"; import type { AuthenticatedEnvironment } from "~/services/apiAuth.server"; import { logger } from "~/services/logger.server"; +import { startActiveSpan } from "~/v3/tracer.server"; export class CompleteRunTaskService { #prismaClient: PrismaClientOrTransaction; @@ -17,76 +18,81 @@ export class CompleteRunTaskService { id: string, taskBody: CompleteTaskBodyOutput ): Promise { - const existingTask = await this.#prismaClient.task.findUnique({ - where: { - id, - }, - include: { - run: true, - attempts: { - where: { - status: "PENDING", - }, - orderBy: { - number: "desc", - }, - take: 1, + return startActiveSpan("CompleteRunTaskService.call", async (span) => { + span.setAttribute("runId", runId); + span.setAttribute("taskId", id); + + const existingTask = await this.#prismaClient.task.findUnique({ + where: { + id, + }, + include: { + run: true, + attempts: { + where: { + status: "PENDING", + }, + orderBy: { + number: "desc", + }, + take: 1, + }, }, - }, - }); - - if (!existingTask) { - return; - } - - if (existingTask.runId !== runId) { - return; - } - - if (existingTask.run.environmentId !== environment.id) { - return; - } - - if ( - existingTask.status === "COMPLETED" || - existingTask.status === "ERRORED" || - existingTask.status === "CANCELED" - ) { - logger.debug("Task already completed", { - existingTask, }); - return taskWithAttemptsToServerTask(existingTask); - } + if (!existingTask) { + return; + } - if (existingTask.attempts.length === 1) { - await this.#prismaClient.taskAttempt.update({ + if (existingTask.runId !== runId) { + return; + } + + if (existingTask.run.environmentId !== environment.id) { + return; + } + + if ( + existingTask.status === "COMPLETED" || + existingTask.status === "ERRORED" || + existingTask.status === "CANCELED" + ) { + logger.debug("Task already completed", { + taskId: id, + }); + + return taskWithAttemptsToServerTask(existingTask); + } + + if (existingTask.attempts.length === 1) { + await this.#prismaClient.taskAttempt.update({ + where: { + id: existingTask.attempts[0].id, + }, + data: { + status: "COMPLETED", + }, + }); + } + + const updatedTask = await this.#prismaClient.task.update({ where: { - id: existingTask.attempts[0].id, + id, }, data: { status: "COMPLETED", + output: taskBody.output as any, + outputIsUndefined: typeof taskBody.output === "undefined", + completedAt: new Date(), + outputProperties: taskBody.properties, + }, + include: { + attempts: true, + run: true, }, }); - } - const updatedTask = await this.#prismaClient.task.update({ - where: { - id, - }, - data: { - status: "COMPLETED", - output: taskBody.output as any, - outputIsUndefined: typeof taskBody.output === "undefined", - completedAt: new Date(), - outputProperties: taskBody.properties, - }, - include: { - attempts: true, - run: true, - }, + return taskWithAttemptsToServerTask(updatedTask); }); - - return taskWithAttemptsToServerTask(updatedTask); } } diff --git a/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.complete/route.ts b/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.complete/route.ts index 3eab7b0cb..618f015f2 100644 --- a/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.complete/route.ts +++ b/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.complete/route.ts @@ -9,8 +9,10 @@ import { import { z } from "zod"; import type { AuthenticatedEnvironment } from "~/services/apiAuth.server"; import { authenticateApiRequest } from "~/services/apiAuth.server"; -import { logger } from "~/services/logger.server"; import { CompleteRunTaskService } from "./CompleteRunTaskService.server"; +import { startActiveSpan } from "~/v3/tracer.server"; +import { parseRequestJsonAsync } from "~/utils/parseRequestJson.server"; +import { FailRunTaskService } from "../api.v1.runs.$runId.tasks.$id.fail/FailRunTaskService.server"; const ParamsSchema = z.object({ runId: z.string(), @@ -44,46 +46,51 @@ export async function action({ request, params }: ActionFunctionArgs) { return json({ error: "Invalid headers" }, { status: 400 }); } + // Check the content size of the request and make sure it's not too large + const contentLength = request.headers.get("content-length"); + + if (!contentLength || parseInt(contentLength) > 3 * 1024 * 1024) { + const service = new FailRunTaskService(); + + await service.call(authenticatedEnv, runId, id, { + error: { + message: "Task output is too large. The limit is 3MB", + }, + }); + + return json({ error: "Task output is too large. The limit is 3MB" }, { status: 413 }); + } + const { "trigger-version": triggerVersion } = headers.data; // Now parse the request body - const anyBody = await request.json(); - - logger.debug("CompleteRunTaskService.call() request body", { - body: anyBody, - runId, - id, - }); + const anyBody = await parseRequestJsonAsync(request, { runId }); if (triggerVersion === API_VERSIONS.SERIALIZED_TASK_OUTPUT) { - const body = CompleteTaskBodyV2InputSchema.safeParse(anyBody); + const body = await startActiveSpan("CompleteTaskBodyV2InputSchema.safeParse()", async () => { + return CompleteTaskBodyV2InputSchema.safeParse(anyBody); + }); if (!body.success) { return json({ error: "Invalid request body" }, { status: 400 }); } - // Make sure the length of the output is less than 3MB - if (body.data.output && body.data.output.length > 3 * 1024 * 1024) { - return json({ error: "Output must be less than 3MB" }, { status: 400 }); - } - return await completeRunTask(authenticatedEnv, runId, id, { ...body.data, output: body.data.output ? (JSON.parse(body.data.output) as any) : undefined, }); } else { - const body = CompleteTaskBodyInputSchema.safeParse(anyBody); + const body = await startActiveSpan("CompleteTaskBodyInputSchema.safeParse()", async () => { + return CompleteTaskBodyInputSchema.omit({ output: true }).safeParse(anyBody); + }); if (!body.success) { return json({ error: "Invalid request body" }, { status: 400 }); } - // Make sure the length of the output is less than 3MB - if (JSON.stringify(body.data.output).length > 3 * 1024 * 1024) { - return json({ error: "Output must be less than 3MB" }, { status: 400 }); - } + const output = (anyBody as any).output; - return await completeRunTask(authenticatedEnv, runId, id, body.data); + return await completeRunTask(authenticatedEnv, runId, id, { ...body.data, output }); } } @@ -98,12 +105,6 @@ async function completeRunTask( try { const task = await service.call(environment, runId, id, taskBody); - logger.debug("CompleteRunTaskService.call() response body", { - runId, - id, - task, - }); - if (!task) { return json({ message: "Task not found" }, { status: 404 }); } diff --git a/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.fail/FailRunTaskService.server.ts b/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.fail/FailRunTaskService.server.ts index 0551c535a..a1d405ea0 100644 --- a/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.fail/FailRunTaskService.server.ts +++ b/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.fail/FailRunTaskService.server.ts @@ -57,10 +57,6 @@ export class FailRunTaskService { existingTask.status === "ERRORED" || existingTask.status === "CANCELED" ) { - logger.debug("Task already completed", { - existingTask, - }); - return existingTask; } diff --git a/apps/webapp/app/routes/api.v1.runs.$runId.tasks/route.ts b/apps/webapp/app/routes/api.v1.runs.$runId.tasks/route.ts index 5d73394f7..fb4f6121d 100644 --- a/apps/webapp/app/routes/api.v1.runs.$runId.tasks/route.ts +++ b/apps/webapp/app/routes/api.v1.runs.$runId.tasks/route.ts @@ -6,6 +6,8 @@ import { authenticateApiRequest } from "~/services/apiAuth.server"; import { logger } from "~/services/logger.server"; import { RunTaskService } from "~/services/tasks/runTask.server"; import { ChangeRequestLazyLoadedCachedTasks } from "./ChangeRequestLazyLoadedCachedTasks.server"; +import { startActiveSpan } from "~/v3/tracer.server"; +import { parseRequestJsonAsync } from "~/utils/parseRequestJson.server"; const ParamsSchema = z.object({ runId: z.string(), @@ -17,6 +19,8 @@ const HeadersSchema = z.object({ "x-cached-tasks-cursor": z.string().optional().nullable(), }); +const BodySchema = RunTaskBodyOutputSchema.omit({ params: true }); + export async function action({ request, params }: ActionFunctionArgs) { // Ensure this is a POST request if (request.method.toUpperCase() !== "POST") { @@ -44,18 +48,26 @@ export async function action({ request, params }: ActionFunctionArgs) { const { runId } = ParamsSchema.parse(params); + const contentLength = request.headers.get("content-length"); + + if (!contentLength || parseInt(contentLength) > 3 * 1024 * 1024) { + return json({ error: "Request body too large" }, { status: 413 }); + } + // Now parse the request body - const anyBody = await request.json(); + const anyBody = await parseRequestJsonAsync(request, { runId }); - logger.debug("RunTaskService.call() request body", { - body: anyBody, - runId, - idempotencyKey, - triggerVersion, - cachedTasksCursor, - }); - - const body = RunTaskBodyOutputSchema.safeParse(anyBody); + const body = await startActiveSpan( + "BodySchema.safeParse", + async () => { + return BodySchema.safeParse(anyBody); + }, + { + attributes: { + runId, + }, + } + ); if (!body.success) { return json({ error: "Invalid request body" }, { status: 400 }); @@ -64,12 +76,9 @@ export async function action({ request, params }: ActionFunctionArgs) { const service = new RunTaskService(); try { - const task = await service.call(runId, idempotencyKey, body.data); - - logger.debug("RunTaskService.call() response body", { - runId, - idempotencyKey, - task, + const task = await service.call(runId, idempotencyKey, { + ...body.data, + params: (anyBody as any).params, }); if (!task) { @@ -84,7 +93,6 @@ export async function action({ request, params }: ActionFunctionArgs) { logger.debug( "RunTaskService.call() response migrating with ChangeRequestLazyLoadedCachedTasks", { - responseBody, cachedTasksCursor, } ); diff --git a/apps/webapp/app/routes/api.v1.tasks.$taskId.batch.ts b/apps/webapp/app/routes/api.v1.tasks.$taskId.batch.ts index ce0b5ad83..e74c904a5 100644 --- a/apps/webapp/app/routes/api.v1.tasks.$taskId.batch.ts +++ b/apps/webapp/app/routes/api.v1.tasks.$taskId.batch.ts @@ -7,6 +7,7 @@ import { authenticateApiRequest } from "~/services/apiAuth.server"; import { logger } from "~/services/logger.server"; import { BatchTriggerTaskService } from "~/v3/services/batchTriggerTask.server"; import { HeadersSchema } from "./api.v1.tasks.$taskId.trigger"; +import { env } from "~/env.server"; const ParamsSchema = z.object({ taskId: z.string(), @@ -43,6 +44,12 @@ export async function action({ request, params }: ActionFunctionArgs) { const { taskId } = ParamsSchema.parse(params); + const contentLength = request.headers.get("content-length"); + + if (!contentLength || parseInt(contentLength) > env.TASK_PAYLOAD_MAXIMUM_SIZE) { + return json({ error: "Request body too large" }, { status: 413 }); + } + // Now parse the request body const anyBody = await request.json(); diff --git a/apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts b/apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts index 0d481b826..9fac30e72 100644 --- a/apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts +++ b/apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts @@ -2,9 +2,12 @@ import type { ActionFunctionArgs } from "@remix-run/server-runtime"; import { json } from "@remix-run/server-runtime"; import { TriggerTaskRequestBody } from "@trigger.dev/core/v3"; import { z } from "zod"; +import { env } from "~/env.server"; import { authenticateApiRequest } from "~/services/apiAuth.server"; import { logger } from "~/services/logger.server"; +import { parseRequestJsonAsync } from "~/utils/parseRequestJson.server"; import { TriggerTaskService } from "~/v3/services/triggerTask.server"; +import { startActiveSpan } from "~/v3/tracer.server"; const ParamsSchema = z.object({ taskId: z.string(), @@ -32,6 +35,12 @@ export async function action({ request, params }: ActionFunctionArgs) { return json({ error: "Invalid or Missing API key" }, { status: 401 }); } + const contentLength = request.headers.get("content-length"); + + if (!contentLength || parseInt(contentLength) > env.TASK_PAYLOAD_MAXIMUM_SIZE) { + return json({ error: "Request body too large" }, { status: 413 }); + } + const rawHeaders = Object.fromEntries(request.headers); const headers = HeadersSchema.safeParse(rawHeaders); @@ -52,9 +61,11 @@ export async function action({ request, params }: ActionFunctionArgs) { const { taskId } = ParamsSchema.parse(params); // Now parse the request body - const anyBody = await request.json(); + const anyBody = await parseRequestJsonAsync(request, { taskId }); - const body = TriggerTaskRequestBody.safeParse(anyBody); + const body = await startActiveSpan("TriggerTaskRequestBody.safeParse()", async (span) => { + return TriggerTaskRequestBody.safeParse(anyBody); + }); if (!body.success) { return json({ error: "Invalid request body" }, { status: 400 }); @@ -76,17 +87,23 @@ export async function action({ request, params }: ActionFunctionArgs) { idempotencyKey, triggerVersion, headers: Object.fromEntries(request.headers), - body: body.data, + options: body.data.options, isFromWorker, traceContext, }); - const run = await service.call(taskId, authenticationResult.environment, body.data, { - idempotencyKey: idempotencyKey ?? undefined, - triggerVersion: triggerVersion ?? undefined, - traceContext, - spanParentAsLink: spanParentAsLink === 1, - }); + const run = await service.call( + taskId, + authenticationResult.environment, + { ...body.data }, + // { ...body.data, payload: (anyBody as any).payload }, + { + idempotencyKey: idempotencyKey ?? undefined, + triggerVersion: triggerVersion ?? undefined, + traceContext, + spanParentAsLink: spanParentAsLink === 1, + } + ); if (!run) { return json({ error: "Task not found" }, { status: 404 }); diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.v3.$projectParam.runs.$runParam.spans.$spanParam/route.tsx b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.v3.$projectParam.runs.$runParam.spans.$spanParam/route.tsx index 195c66d83..4db976ceb 100644 --- a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.v3.$projectParam.runs.$runParam.spans.$spanParam/route.tsx +++ b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.v3.$projectParam.runs.$runParam.spans.$spanParam/route.tsx @@ -33,7 +33,13 @@ import { redirectWithErrorMessage } from "~/models/message.server"; import { Span, SpanPresenter } from "~/presenters/v3/SpanPresenter.server"; import { requireUserId } from "~/services/session.server"; import { cn } from "~/utils/cn"; -import { v3RunPath, v3RunSpanPath, v3SpanParamsSchema, v3TraceSpanPath } from "~/utils/pathBuilder"; +import { + v3RunDownloadLogsPath, + v3RunPath, + v3RunSpanPath, + v3SpanParamsSchema, + v3TraceSpanPath, +} from "~/utils/pathBuilder"; import { SpanLink } from "~/v3/eventRepository.server"; export const loader = async ({ request, params }: LoaderFunctionArgs) => { @@ -256,6 +262,15 @@ function RunActionButtons({ span }: { span: Span }) { if (span.isPartial) { return ( + + Download logs +