From 75550f535ca47252f7f39b8a56861ae07523135b Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Mon, 13 Nov 2023 16:18:35 +0000 Subject: [PATCH] Fix undefined task outputs And also fix the SSE connection getting reset on a infinite loop when runs are complete --- apps/webapp/app/components/run/TaskDetail.tsx | 36 ++++--- apps/webapp/app/hooks/useEventSource.tsx | 9 +- apps/webapp/app/models/task.server.ts | 6 +- .../presenters/TaskDetailsPresenter.server.ts | 14 +-- .../route.tsx | 31 +++--- .../route.tsx | 1 + .../route.tsx | 29 +++--- .../api.v1.runs.$runId.tasks.$id.complete.ts | 3 +- .../runs/performRunExecutionV3.server.ts | 1 + apps/webapp/app/utils/sse.server.ts | 30 +++--- .../migration.sql | 2 + packages/database/prisma/schema.prisma | 23 ++--- references/job-catalog/package.json | 1 + references/job-catalog/src/edge-cases.ts | 99 +++++++++++++++++++ 14 files changed, 210 insertions(+), 75 deletions(-) create mode 100644 packages/database/prisma/migrations/20231113151412_add_output_is_undefined/migration.sql create mode 100644 references/job-catalog/src/edge-cases.ts diff --git a/apps/webapp/app/components/run/TaskDetail.tsx b/apps/webapp/app/components/run/TaskDetail.tsx index cd19f077b..bb4c9ec86 100644 --- a/apps/webapp/app/components/run/TaskDetail.tsx +++ b/apps/webapp/app/components/run/TaskDetail.tsx @@ -33,7 +33,19 @@ import { Spinner } from "../primitives/Spinner"; import type { DetailedTask } from "~/routes/_app.orgs.$organizationSlug.projects.$projectParam.jobs.$jobParam.runs.$runParam.tasks.$taskParam/route"; export function TaskDetail({ task }: { task: DetailedTask }) { - const { name, description, icon, status, params, properties, output, style, attempts } = task; + const { + name, + description, + icon, + status, + params, + properties, + output, + outputIsUndefined, + style, + attempts, + noop, + } = task; const startedAt = task.startedAt ? new Date(task.startedAt) : undefined; const completedAt = task.completedAt ? new Date(task.completedAt) : undefined; @@ -140,16 +152,18 @@ export function TaskDetail({ task }: { task: DetailedTask }) { No input )} -
- Output - {output ? ( - }> - {() => } - - ) : ( - No output - )} -
+ {!noop && ( +
+ Output + {output && !outputIsUndefined ? ( + }> + {() => } + + ) : ( + No output + )} +
+ )} ); diff --git a/apps/webapp/app/hooks/useEventSource.tsx b/apps/webapp/app/hooks/useEventSource.tsx index 98ca1c962..8a4c9a430 100644 --- a/apps/webapp/app/hooks/useEventSource.tsx +++ b/apps/webapp/app/hooks/useEventSource.tsx @@ -3,6 +3,7 @@ import { useEffect, useState } from "react"; type EventSourceOptions = { init?: EventSourceInit; event?: string; + disabled?: boolean; }; /** @@ -13,11 +14,15 @@ type EventSourceOptions = { */ export function useEventSource( url: string | URL, - { event = "message", init }: EventSourceOptions = {} + { event = "message", init, disabled }: EventSourceOptions = {} ) { const [data, setData] = useState(null); useEffect(() => { + if (disabled) { + return; + } + const eventSource = new EventSource(url, init); eventSource.addEventListener(event ?? "message", handler); @@ -32,7 +37,7 @@ export function useEventSource( eventSource.removeEventListener(event ?? "message", handler); eventSource.close(); }; - }, [url, event, init]); + }, [url, event, init, disabled]); return data; } diff --git a/apps/webapp/app/models/task.server.ts b/apps/webapp/app/models/task.server.ts index 83e2e46b1..c840e6f86 100644 --- a/apps/webapp/app/models/task.server.ts +++ b/apps/webapp/app/models/task.server.ts @@ -15,7 +15,7 @@ export function taskWithAttemptsToServerTask(task: TaskWithAttempts): ServerTask status: task.status, description: task.description, params: task.params as any, - output: task.output as any, + output: task.outputIsUndefined ? undefined : (task.output as any), context: task.context as any, properties: task.properties as any, style: task.style as any, @@ -32,7 +32,7 @@ export function taskWithAttemptsToServerTask(task: TaskWithAttempts): ServerTask export type TaskForCaching = Pick< Task, - "id" | "status" | "idempotencyKey" | "noop" | "output" | "parentId" + "id" | "status" | "idempotencyKey" | "noop" | "output" | "parentId" | "outputIsUndefined" >; export function prepareTasksForCaching( @@ -105,7 +105,7 @@ function prepareTaskForCaching(task: TaskForCaching): CachedTask { status: task.status, idempotencyKey: task.idempotencyKey, noop: task.noop, - output: task.output as any, + output: task.outputIsUndefined ? undefined : (task.output as any), parentId: task.parentId, }; } diff --git a/apps/webapp/app/presenters/TaskDetailsPresenter.server.ts b/apps/webapp/app/presenters/TaskDetailsPresenter.server.ts index 4d27df4b7..afcf56da9 100644 --- a/apps/webapp/app/presenters/TaskDetailsPresenter.server.ts +++ b/apps/webapp/app/presenters/TaskDetailsPresenter.server.ts @@ -1,5 +1,4 @@ -import { RedactSchema } from "@trigger.dev/core"; -import { StyleSchema } from "@trigger.dev/core"; +import { RedactSchema, StyleSchema } from "@trigger.dev/core"; import { PrismaClient, prisma } from "~/db.server"; import { mergeProperties } from "~/utils/mergeProperties.server"; import { Redactor } from "~/utils/redactor"; @@ -58,6 +57,7 @@ export class TaskDetailsPresenter { outputProperties: true, params: true, output: true, + outputIsUndefined: true, error: true, startedAt: true, completedAt: true, @@ -89,9 +89,11 @@ export class TaskDetailsPresenter { return { ...task, redact: undefined, - output: task.output - ? JSON.stringify(this.#stringifyOutputWithRedactions(task.output, task.redact), null, 2) - : undefined, + output: JSON.stringify( + this.#stringifyOutputWithRedactions(task.output, task.redact), + null, + 2 + ), connection: task.runConnection, params: task.params as Record, properties: mergeProperties(task.properties, task.outputProperties), @@ -101,7 +103,7 @@ export class TaskDetailsPresenter { #stringifyOutputWithRedactions(output: any, redact: unknown): any { if (!output) { - return; + return output; } const parsedRedact = RedactSchema.safeParse(redact); diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.jobs.$jobParam.runs.$runParam.tasks.$taskParam/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.jobs.$jobParam.runs.$runParam.tasks.$taskParam/route.tsx index 52538a357..0f114b1b8 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.jobs.$jobParam.runs.$runParam.tasks.$taskParam/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.jobs.$jobParam.runs.$runParam.tasks.$taskParam/route.tsx @@ -1,6 +1,5 @@ -import { Await, useLoaderData } from "@remix-run/react"; -import { LoaderFunctionArgs, SerializeFrom, defer } from "@remix-run/server-runtime"; -import { Suspense } from "react"; +import { LoaderFunctionArgs } from "@remix-run/server-runtime"; +import { UseDataFunctionReturn, typedjson, useTypedLoaderData } from "remix-typedjson"; import { Spinner } from "~/components/primitives/Spinner"; import { TaskDetail } from "~/components/run/TaskDetail"; import { TaskDetailsPresenter } from "~/presenters/TaskDetailsPresenter.server"; @@ -12,26 +11,28 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => { const { taskParam } = TaskParamsSchema.parse(params); const presenter = new TaskDetailsPresenter(); - const taskPromise = presenter.call({ + const task = await presenter.call({ userId, id: taskParam, }); - return defer({ - taskPromise, + return typedjson({ + task, }); }; -export type DetailedTask = NonNullable["taskPromise"]>>; +export type DetailedTask = NonNullable["task"]>; export default function Page() { - const { taskPromise } = useLoaderData(); + const { task } = useTypedLoaderData(); - return ( - }> - Error loading task!

}> - {(resolvedTask) => resolvedTask && } - - - ); + if (!task) { + return ( +
+ +
+ ); + } + + return ; } diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.jobs.$jobParam.runs.$runParam/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.jobs.$jobParam.runs.$runParam/route.tsx index fdcceb03b..a8a382e3b 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.jobs.$jobParam.runs.$runParam/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.jobs.$jobParam.runs.$runParam/route.tsx @@ -67,6 +67,7 @@ export default function Page() { const revalidator = useRevalidator(); const events = useEventSource(runStreamingPath(organization, project, job, run), { event: "message", + disabled: !!run.completedAt, }); useEffect(() => { if (events !== null) { diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.triggers_.external.$triggerParam_.runs.$runParam.tasks.$taskParam/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.triggers_.external.$triggerParam_.runs.$runParam.tasks.$taskParam/route.tsx index 1e07e6deb..532bafddb 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.triggers_.external.$triggerParam_.runs.$runParam.tasks.$taskParam/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.triggers_.external.$triggerParam_.runs.$runParam.tasks.$taskParam/route.tsx @@ -1,6 +1,5 @@ -import { Await, useLoaderData } from "@remix-run/react"; -import { LoaderFunctionArgs, defer } from "@remix-run/server-runtime"; -import { Suspense } from "react"; +import { LoaderFunctionArgs } from "@remix-run/server-runtime"; +import { typedjson, useTypedLoaderData } from "remix-typedjson"; import { Spinner } from "~/components/primitives/Spinner"; import { TaskDetail } from "~/components/run/TaskDetail"; import { TaskDetailsPresenter } from "~/presenters/TaskDetailsPresenter.server"; @@ -12,24 +11,26 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => { const { taskParam } = TriggerSourceRunTaskParamsSchema.parse(params); const presenter = new TaskDetailsPresenter(); - const taskPromise = presenter.call({ + const task = await presenter.call({ userId, id: taskParam, }); - return defer({ - taskPromise, + return typedjson({ + task, }); }; export default function Page() { - const { taskPromise } = useLoaderData(); + const { task } = useTypedLoaderData(); - return ( - }> - Error loading task!

}> - {(resolvedTask) => resolvedTask && } - - - ); + if (!task) { + return ( +
+ +
+ ); + } + + return ; } diff --git a/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.complete.ts b/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.complete.ts index 74df41dc1..44a5be5e0 100644 --- a/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.complete.ts +++ b/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.complete.ts @@ -191,7 +191,8 @@ export class CompleteRunTaskService { }, data: { status: "COMPLETED", - output: taskBody.output ?? undefined, + output: taskBody.output as any, + outputIsUndefined: typeof taskBody.output === "undefined", completedAt: new Date(), outputProperties: taskBody.properties, }, diff --git a/apps/webapp/app/services/runs/performRunExecutionV3.server.ts b/apps/webapp/app/services/runs/performRunExecutionV3.server.ts index 6beadd365..2472978b9 100644 --- a/apps/webapp/app/services/runs/performRunExecutionV3.server.ts +++ b/apps/webapp/app/services/runs/performRunExecutionV3.server.ts @@ -1220,6 +1220,7 @@ async function findRun(prisma: PrismaClientOrTransaction, id: string) { status: true, noop: true, output: true, + outputIsUndefined: true, parentId: true, }, orderBy: { diff --git a/apps/webapp/app/utils/sse.server.ts b/apps/webapp/app/utils/sse.server.ts index 72ea6d24c..fced1fbaf 100644 --- a/apps/webapp/app/utils/sse.server.ts +++ b/apps/webapp/app/utils/sse.server.ts @@ -22,6 +22,16 @@ export function sse({ request, pingInterval = 1000, updateInterval = 348, run }: return new Response("SSE disabled", { status: 200 }); } + let pinger: NodeJS.Timer | undefined = undefined; + let updater: NodeJS.Timer | undefined = undefined; + let timeout: NodeJS.Timeout | undefined = undefined; + + const abort = () => { + clearInterval(pinger); + clearInterval(updater); + clearTimeout(timeout); + }; + return eventStream(request.signal, (send, close) => { const safeSend = (args: { event?: string; data: string }) => { try { @@ -49,30 +59,26 @@ export function sse({ request, pingInterval = 1000, updateInterval = 348, run }: } }; - const pinger = setInterval(() => { + pinger = setInterval(() => { if (request.signal.aborted) { - return close(); + return abort(); } safeSend({ event: "ping", data: new Date().toISOString() }); }, pingInterval); - const updater = setInterval(() => { + updater = setInterval(() => { if (request.signal.aborted) { - return close(); + return abort(); } - run(safeSend, close); + run(safeSend, abort); }, updateInterval); - const timeout = setTimeout(() => { - close(); + timeout = setTimeout(() => { + close(); // close the connection after 1 minute of inactivity, which will refresh the connection (that's why we aren't using abort) }, 60 * 1000); // 1 minute - return () => { - clearInterval(updater); - clearInterval(pinger); - clearTimeout(timeout); - }; + return abort; }); } diff --git a/packages/database/prisma/migrations/20231113151412_add_output_is_undefined/migration.sql b/packages/database/prisma/migrations/20231113151412_add_output_is_undefined/migration.sql new file mode 100644 index 000000000..f70613b2e --- /dev/null +++ b/packages/database/prisma/migrations/20231113151412_add_output_is_undefined/migration.sql @@ -0,0 +1,2 @@ +-- AlterTable +ALTER TABLE "Task" ADD COLUMN "outputIsUndefined" BOOLEAN NOT NULL DEFAULT false; diff --git a/packages/database/prisma/schema.prisma b/packages/database/prisma/schema.prisma index 3e7fe37ee..2e357db70 100644 --- a/packages/database/prisma/schema.prisma +++ b/packages/database/prisma/schema.prisma @@ -882,17 +882,18 @@ model Task { delayUntil DateTime? noop Boolean @default(false) - description String? - properties Json? - outputProperties Json? - params Json? - output Json? - context Json? - error String? - redact Json? - style Json? - operation String? - callbackUrl String? + description String? + properties Json? + outputProperties Json? + params Json? + output Json? + outputIsUndefined Boolean @default(false) + context Json? + error String? + redact Json? + style Json? + operation String? + callbackUrl String? startedAt DateTime? completedAt DateTime? diff --git a/references/job-catalog/package.json b/references/job-catalog/package.json index f91285120..9ddc68237 100644 --- a/references/job-catalog/package.json +++ b/references/job-catalog/package.json @@ -31,6 +31,7 @@ "cli-example": "nodemon --watch src/cli-example.ts -r tsconfig-paths/register -r dotenv/config src/cli-example.ts", "invoke": "nodemon --watch src/invoke.ts -r tsconfig-paths/register -r dotenv/config src/invoke.ts", "built-ins": "nodemon --watch src/built-ins.ts -r tsconfig-paths/register -r dotenv/config src/built-ins.ts", + "edge-cases": "nodemon --watch src/edge-cases.ts -r tsconfig-paths/register -r dotenv/config src/edge-cases.ts", "dev:trigger": "trigger-cli dev --port 8080" }, "dependencies": { diff --git a/references/job-catalog/src/edge-cases.ts b/references/job-catalog/src/edge-cases.ts new file mode 100644 index 000000000..bf70f7b27 --- /dev/null +++ b/references/job-catalog/src/edge-cases.ts @@ -0,0 +1,99 @@ +import { createExpressServer } from "@trigger.dev/express"; +import { TriggerClient, invokeTrigger } from "@trigger.dev/sdk"; +import fs from "node:fs"; +import fsPromises from "node:fs/promises"; + +export const client = new TriggerClient({ + id: "job-catalog", + apiKey: process.env["TRIGGER_API_KEY"], + apiUrl: process.env["TRIGGER_API_URL"], + verbose: true, + ioLogLocalEnabled: true, +}); + +client.defineJob({ + id: "task-output-edge-cases", + name: "Task Output Edge Cases", + version: "1.0.0", + trigger: invokeTrigger(), + run: async (payload, io, ctx) => { + const result1 = await io.runTask("undefined", async (task) => { + return undefined; + }); + + if (typeof result1 !== "undefined") { + throw new Error(`Expected undefined, got ${typeof result1}: ${JSON.stringify(result1)}`); + } + + const result2 = await io.runTask("null", async (task) => { + return null; + }); + + if (result2 !== null) { + throw new Error(`Expected null, got ${typeof result2}: ${JSON.stringify(result2)}`); + } + + const result3 = await io.runTask("false", async (task) => { + return false; + }); + + if (typeof result3 !== "boolean" && result3 !== false) { + throw new Error(`Expected false, got ${typeof result3}: ${JSON.stringify(result3)}`); + } + + const result4 = await io.runTask("true", async (task) => { + return true; + }); + + if (typeof result4 !== "boolean" && result4 !== true) { + throw new Error(`Expected true, got ${typeof result4}: ${JSON.stringify(result4)}`); + } + + const result5 = await io.runTask("date", async (task) => { + return new Date(); + }); + + if (typeof result5 !== "string" || new Date(result5).toString() === "Invalid Date") { + throw new Error(`Expected string, got ${typeof result5}: ${JSON.stringify(result5)}`); + } + + const result6 = await io.runTask("object", async (task) => { + return { + a: 1, + b: "2", + c: true, + d: new Date(), + e: null, + f: undefined, + }; + }); + + if (typeof result6 !== "object" || result6 === null) { + throw new Error(`Expected object, got ${typeof result6}: ${JSON.stringify(result6)}`); + } + + const result7 = await io.runTask("array", async (task) => { + return [1, "2", true, new Date(), null, undefined]; + }); + + if (!Array.isArray(result7)) { + throw new Error(`Expected array, got ${typeof result7}: ${JSON.stringify(result7)}`); + } + + const result8 = await io.runTask("file", async (task) => { + return fs.createReadStream(__filename) as any; + }); + + const result9 = await io.runTask("read-file", async (task) => { + return fsPromises.readFile(__filename, "utf-8"); + }); + + if (typeof result9 !== "string") { + throw new Error(`Expected string, got ${typeof result9}: ${JSON.stringify(result9)}`); + } + + await io.wait("wait-1", 1); + }, +}); + +createExpressServer(client);