Associating runs with waitpoints

matt-aitken · matt-aitken · commit 483246953e94 · 2025-03-25T14:50:30.000Z
diff --git a/apps/webapp/app/components/runs/v3/WaitpointDetails.tsx b/apps/webapp/app/components/runs/v3/WaitpointDetails.tsx
@@ -1,12 +1,28 @@
 import { DateTime, DateTimeAccurate } from "~/components/primitives/DateTime";
+import { Paragraph } from "~/components/primitives/Paragraph";
 import * as Property from "~/components/primitives/PropertyTable";
 import { type WaitpointDetail } from "~/presenters/v3/WaitpointPresenter.server";
 import { ForceTimeout } from "~/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.waitpoints.$waitpointFriendlyId.complete/route";
 import { PacketDisplay } from "./PacketDisplay";
 import { WaitpointStatusCombo } from "./WaitpointStatus";
-import { Paragraph } from "~/components/primitives/Paragraph";
+import { v3WaitpointTokensPath } from "~/utils/pathBuilder";
+import { Link } from "@remix-run/react";
+import { TextLink } from "~/components/primitives/TextLink";
+import { useOrganization } from "~/hooks/useOrganizations";
+import { useProject } from "~/hooks/useProject";
+import { useEnvironment } from "~/hooks/useEnvironment";
+
+export function WaitpointDetailTable({
+  waitpoint,
+  linkToList = false,
+}: {
+  waitpoint: WaitpointDetail;
+  linkToList?: boolean;
+}) {
+  const organization = useOrganization();
+  const project = useProject();
+  const environment = useEnvironment();
 
-export function WaitpointDetailTable({ waitpoint }: { waitpoint: WaitpointDetail }) {
   const hasExpired =
     waitpoint.idempotencyKeyExpiresAt && waitpoint.idempotencyKeyExpiresAt < new Date();
 
@@ -24,7 +40,19 @@ export function WaitpointDetailTable({ waitpoint }: { waitpoint: WaitpointDetail
       </Property.Item>
       <Property.Item>
         <Property.Label>ID</Property.Label>
-        <Property.Value className="whitespace-pre-wrap">{waitpoint.friendlyId}</Property.Value>
+        <Property.Value className="whitespace-pre-wrap">
+          {linkToList ? (
+            <TextLink
+              to={v3WaitpointTokensPath(organization, project, environment, {
+                id: waitpoint.friendlyId,
+              })}
+            >
+              {waitpoint.friendlyId}
+            </TextLink>
+          ) : (
+            waitpoint.friendlyId
+          )}
+        </Property.Value>
       </Property.Item>
       <Property.Item>
         <Property.Label>Idempotency key</Property.Label>
diff --git a/apps/webapp/app/components/runs/v3/WaitpointTokenFilters.tsx b/apps/webapp/app/components/runs/v3/WaitpointTokenFilters.tsx
@@ -65,6 +65,7 @@ export const WaitpointSearchParamsSchema = z.object({
   cursor: z.string().optional(),
   direction: z.enum(["forward", "backward"]).optional(),
 });
+export type WaitpointSearchParams = z.infer<typeof WaitpointSearchParamsSchema>;
 
 type WaitpointTokenFiltersProps = {
   hasFilters: boolean;
@@ -561,7 +562,7 @@ function AppliedWaitpointIdFilter() {
           trigger={
             <Ariakit.Select render={<div className="group cursor-pointer focus-custom" />}>
               <AppliedFilter
-                label="Waitpoint ID"
+                label="ID"
                 value={id}
                 onRemove={() => del(["id", "cursor", "direction"])}
               />
@@ -629,7 +630,7 @@ function IdempotencyKeyDropdown({
           <div className="flex flex-col gap-1">
             <Label>Idempotency key</Label>
             <Input
-              placeholder="run_"
+              placeholder="waitpoint_"
               value={idempotencyKey ?? ""}
               onChange={(e) => setIdempotencyKey(e.target.value)}
               variant="small"
diff --git a/apps/webapp/app/presenters/v3/RunListPresenter.server.ts b/apps/webapp/app/presenters/v3/RunListPresenter.server.ts
@@ -24,7 +24,7 @@ export type RunListOptions = {
   isTest?: boolean;
   rootOnly?: boolean;
   batchId?: string;
-  runId?: string;
+  runIds?: string[];
   //pagination
   direction?: Direction;
   cursor?: string;
@@ -52,7 +52,7 @@ export class RunListPresenter extends BasePresenter {
     isTest,
     rootOnly,
     batchId,
-    runId,
+    runIds,
     from,
     to,
     direction = "forward",
@@ -72,7 +72,7 @@ export class RunListPresenter extends BasePresenter {
       (scheduleId !== undefined && scheduleId !== "") ||
       (tags !== undefined && tags.length > 0) ||
       batchId !== undefined ||
-      runId !== undefined ||
+      (runIds !== undefined && runIds.length > 0) ||
       typeof isTest === "boolean" ||
       rootOnly === true;
 
@@ -182,7 +182,7 @@ export class RunListPresenter extends BasePresenter {
     }
 
     //show all runs if we are filtering by batchId or runId
-    if (batchId || runId || scheduleId || tasks?.length) {
+    if (batchId || runIds?.length || scheduleId || tasks?.length) {
       rootOnly = false;
     }
 
@@ -261,7 +261,7 @@ WHERE
         : Prisma.empty
     }
     -- filters
-    ${runId ? Prisma.sql`AND tr."friendlyId" = ${runId}` : Prisma.empty}
+    ${runIds ? Prisma.sql`AND tr."friendlyId" IN (${Prisma.join(runIds)})` : Prisma.empty}
     ${batchId ? Prisma.sql`AND tr."batchId" = ${batchId}` : Prisma.empty}
     ${
       restrictToRunIds
diff --git a/apps/webapp/app/presenters/v3/WaitpointPresenter.server.ts b/apps/webapp/app/presenters/v3/WaitpointPresenter.server.ts
@@ -1,11 +1,20 @@
 import { isWaitpointOutputTimeout, prettyPrintPacket } from "@trigger.dev/core/v3";
 import { logger } from "~/services/logger.server";
 import { BasePresenter } from "./basePresenter.server";
+import { RunListItem, RunListPresenter } from "./RunListPresenter.server";
 
 export type WaitpointDetail = NonNullable<Awaited<ReturnType<WaitpointPresenter["call"]>>>;
 
 export class WaitpointPresenter extends BasePresenter {
-  public async call({ friendlyId, environmentId }: { friendlyId: string; environmentId: string }) {
+  public async call({
+    friendlyId,
+    environmentId,
+    projectId,
+  }: {
+    friendlyId: string;
+    environmentId: string;
+    projectId: string;
+  }) {
     const waitpoint = await this._replica.waitpoint.findFirst({
       where: {
         friendlyId,
@@ -23,6 +32,12 @@ export class WaitpointPresenter extends BasePresenter {
         outputType: true,
         outputIsError: true,
         completedAfter: true,
+        connectedRuns: {
+          select: {
+            friendlyId: true,
+          },
+          take: 5,
+        },
       },
     });
 
@@ -47,6 +62,20 @@ export class WaitpointPresenter extends BasePresenter {
       }
     }
 
+    const connectedRunIds = waitpoint.connectedRuns.map((run) => run.friendlyId);
+    const connectedRuns: RunListItem[] = [];
+
+    if (connectedRunIds.length > 0) {
+      const runPresenter = new RunListPresenter();
+      const { runs } = await runPresenter.call({
+        projectId: projectId,
+        environments: [environmentId],
+        runIds: connectedRunIds,
+        pageSize: 5,
+      });
+      connectedRuns.push(...runs);
+    }
+
     return {
       friendlyId: waitpoint.friendlyId,
       type: waitpoint.type,
@@ -60,6 +89,7 @@ export class WaitpointPresenter extends BasePresenter {
       outputIsError: waitpoint.outputIsError,
       completedAfter: waitpoint.completedAfter,
       isTimeout,
+      connectedRuns,
     };
   }
 }
diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs._index/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs._index/route.tsx
@@ -134,7 +134,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
     from,
     to,
     batchId,
-    runId,
+    runIds: runId ? [runId] : undefined,
     scheduleId,
     rootOnly,
     direction: direction,
diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.waitpoints.tokens.$waitpointParam/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.waitpoints.tokens.$waitpointParam/route.tsx
@@ -4,7 +4,7 @@ import { typedjson, useTypedLoaderData } from "remix-typedjson";
 import { z } from "zod";
 import { ExitIcon } from "~/assets/icons/ExitIcon";
 import { LinkButton } from "~/components/primitives/Buttons";
-import { Header2 } from "~/components/primitives/Headers";
+import { Header2, Header3 } from "~/components/primitives/Headers";
 import { useEnvironment } from "~/hooks/useEnvironment";
 import { useOrganization } from "~/hooks/useOrganizations";
 import { useProject } from "~/hooks/useProject";
@@ -16,6 +16,8 @@ import { cn } from "~/utils/cn";
 import { EnvironmentParamSchema, v3WaitpointTokensPath } from "~/utils/pathBuilder";
 import { CompleteWaitpointForm } from "../resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.waitpoints.$waitpointFriendlyId.complete/route";
 import { WaitpointDetailTable } from "~/components/runs/v3/WaitpointDetails";
+import { TaskRunsTable } from "~/components/runs/v3/TaskRunsTable";
+import { InfoIconTooltip } from "~/components/primitives/Tooltip";
 
 const Params = EnvironmentParamSchema.extend({
   waitpointParam: z.string(),
@@ -46,6 +48,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
     const result = await presenter.call({
       friendlyId: waitpointParam,
       environmentId: environment.id,
+      projectId: project.id,
     });
 
     if (!result) {
@@ -94,8 +97,31 @@ export default function Page() {
           className="pl-1"
         />
       </div>
-      <div className="overflow-y-auto px-3 pt-3 scrollbar-thin scrollbar-track-transparent scrollbar-thumb-charcoal-600">
-        <WaitpointDetailTable waitpoint={waitpoint} />
+      <div className="overflow-y-auto pt-3 scrollbar-thin scrollbar-track-transparent scrollbar-thumb-charcoal-600">
+        <div className="px-3">
+          <WaitpointDetailTable waitpoint={waitpoint} />
+        </div>
+        <div className="flex flex-col gap-1 pt-6">
+          <div className="mb-1 flex items-center gap-1 pl-3">
+            <Header3>5 related runs</Header3>
+            <InfoIconTooltip content="These runs have been blocked by this waitpoint." />
+          </div>
+          <TaskRunsTable
+            total={waitpoint.connectedRuns.length}
+            hasFilters={false}
+            filters={{
+              tasks: [],
+              versions: [],
+              statuses: [],
+              environments: [],
+              from: undefined,
+              to: undefined,
+            }}
+            runs={waitpoint.connectedRuns}
+            isLoading={false}
+            variant="bright"
+          />
+        </div>
       </div>
       {waitpoint.status === "PENDING" && (
         <div>
diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.waitpoints.tokens/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.waitpoints.tokens/route.tsx
@@ -203,7 +203,7 @@ export default function Page() {
                                   "–"
                                 )}
                               </TableCell>
-                              <TableCell to={path} actionClassName="py-1" className="pr-16">
+                              <TableCell to={path} actionClassName="py-1">
                                 <div className="flex gap-1">
                                   {token.tags.map((tag) => <RunTag key={tag} tag={tag} />) || "–"}
                                 </div>
diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.spans.$spanParam/route.tsx b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.spans.$spanParam/route.tsx
@@ -950,7 +950,7 @@ function SpanEntity({ span }: { span: Span }) {
                 <TextLink to={docsPath("wait")}>View docs</TextLink>.
               </Paragraph>
             </div>
-            <WaitpointDetailTable waitpoint={span.entity.object} />
+            <WaitpointDetailTable waitpoint={span.entity.object} linkToList />
           </div>
           {span.entity.object.status === "PENDING" && (
             <div>
diff --git a/apps/webapp/app/utils/pathBuilder.ts b/apps/webapp/app/utils/pathBuilder.ts
@@ -4,6 +4,10 @@ import { type TaskRunListSearchFilters } from "~/components/runs/v3/RunFilters";
 import type { Organization } from "~/models/organization.server";
 import type { Project } from "~/models/project.server";
 import { objectToSearchParams } from "./searchParams";
+import {
+  WaitpointFilterStatus,
+  WaitpointSearchParams,
+} from "~/components/runs/v3/WaitpointTokenFilters";
 
 export type OrgForPath = Pick<Organization, "slug">;
 export type ProjectForPath = Pick<Project, "slug">;
@@ -314,9 +318,12 @@ export function v3QueuesPath(
 export function v3WaitpointTokensPath(
   organization: OrgForPath,
   project: ProjectForPath,
-  environment: EnvironmentForPath
+  environment: EnvironmentForPath,
+  filters?: WaitpointSearchParams
 ) {
-  return `${v3EnvironmentPath(organization, project, environment)}/waitpoints/tokens`;
+  const searchParams = objectToSearchParams(filters);
+  const query = searchParams ? `?${searchParams.toString()}` : "";
+  return `${v3EnvironmentPath(organization, project, environment)}/waitpoints/tokens${query}`;
 }
 
 export function v3WaitpointTokenPath(
diff --git a/internal-packages/database/prisma/migrations/20250325124348_added_connected_runs_to_waitpoints/migration.sql b/internal-packages/database/prisma/migrations/20250325124348_added_connected_runs_to_waitpoints/migration.sql
@@ -0,0 +1,15 @@
+-- CreateTable
+CREATE TABLE
+    "_WaitpointRunConnections" ("A" TEXT NOT NULL, "B" TEXT NOT NULL);
+
+-- CreateIndex
+CREATE UNIQUE INDEX "_WaitpointRunConnections_AB_unique" ON "_WaitpointRunConnections" ("A", "B");
+
+-- CreateIndex
+CREATE INDEX "_WaitpointRunConnections_B_index" ON "_WaitpointRunConnections" ("B");
+
+-- AddForeignKey
+ALTER TABLE "_WaitpointRunConnections" ADD CONSTRAINT "_WaitpointRunConnections_A_fkey" FOREIGN KEY ("A") REFERENCES "TaskRun" ("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "_WaitpointRunConnections" ADD CONSTRAINT "_WaitpointRunConnections_B_fkey" FOREIGN KEY ("B") REFERENCES "Waitpoint" ("id") ON DELETE CASCADE ON UPDATE CASCADE;
diff --git a/internal-packages/database/prisma/schema.prisma b/internal-packages/database/prisma/schema.prisma
@@ -1794,11 +1794,14 @@ model TaskRun {
   oneTimeUseToken String?
 
   ///When this run is finished, the waitpoint will be marked as completed
-  associatedWaitpoint Waitpoint?
+  associatedWaitpoint Waitpoint? @relation("CompletingRun")
 
   ///If there are any blocked waitpoints, the run won't be executed
   blockedByWaitpoints TaskRunWaitpoint[]
 
+  /// All waitpoints that blocked this run at some point, used for display purposes
+  connectedWaitpoints Waitpoint[] @relation("WaitpointRunConnections")
+
   /// Where the logs are stored
   taskEventStore String @default("taskEvent")
 
@@ -2109,7 +2112,7 @@ model Waitpoint {
 
   /// If it's a RUN type waitpoint, this is the associated run
   completedByTaskRunId String?  @unique
-  completedByTaskRun   TaskRun? @relation(fields: [completedByTaskRunId], references: [id], onDelete: SetNull)
+  completedByTaskRun   TaskRun? @relation("CompletingRun", fields: [completedByTaskRunId], references: [id], onDelete: SetNull)
 
   /// If it's a DATETIME type waitpoint, this is the date.
   /// If it's a MANUAL waitpoint, this can be set as the `timeout`.
@@ -2122,6 +2125,9 @@ model Waitpoint {
   /// The runs this waitpoint is blocking
   blockingTaskRuns TaskRunWaitpoint[]
 
+  /// All runs that have ever been blocked by this waitpoint, used for display purposes
+  connectedRuns TaskRun[] @relation("WaitpointRunConnections")
+
   /// When a waitpoint is complete
   completedExecutionSnapshots TaskRunExecutionSnapshot[] @relation("completedWaitpoints")
 
diff --git a/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts b/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts
@@ -399,6 +399,13 @@ export class WaitpointSystem {
           WHERE w.id IN (${Prisma.join($waitpoints)})
           ON CONFLICT DO NOTHING
           RETURNING "waitpointId"
+        ),
+        connected_runs AS (
+          INSERT INTO "_WaitpointRunConnections" ("A", "B")
+          SELECT ${runId}, w.id
+          FROM "Waitpoint" w
+          WHERE w.id IN (${Prisma.join($waitpoints)})
+          ON CONFLICT DO NOTHING
         )
         SELECT COUNT(*) as pending_count
         FROM inserted i
diff --git a/references/hello-world/src/trigger/tags.ts b/references/hello-world/src/trigger/tags.ts
@@ -0,0 +1,13 @@
+import { logger, runs, task, tasks } from "@trigger.dev/sdk/v3";
+import { fixedLengthTask } from "./prioritize-continuing.js";
+
+type Payload = {
+  tags: string | string[];
+};
+
+export const triggerRunsWithTags = task({
+  id: "tags",
+  run: async (payload: Payload, { ctx }) => {
+    const { id } = await fixedLengthTask.trigger({ waitSeconds: 5 }, { tags: payload.tags });
+  },
+});
