Durations waits use the API to create/block with a waitpoint, not the runtime

Author: matt-aitken

apps/webapp/app/routes/api.v1.dev.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.wait.duration.ts

@@ -0,0 +1,75 @@
+import { json, TypedResponse } from "@remix-run/server-runtime";
+import { WaitForDurationRequestBody, WaitForDurationResponseBody } from "@trigger.dev/core/v3";
+import { RunId } from "@trigger.dev/core/v3/apps";
+
+import { z } from "zod";
+import { prisma } from "~/db.server";
+import { logger } from "~/services/logger.server";
+import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { resolveIdempotencyKeyTTL } from "~/utils/idempotencyKeys.server";
+import { engine } from "~/v3/runEngine.server";
+
+const { action } = createActionApiRoute(
+  {
+    body: WaitForDurationRequestBody,
+    params: z.object({
+      runFriendlyId: z.string(),
+    }),
+    method: "POST",
+  },
+  async ({ authentication, body, params }): Promise<TypedResponse<WaitForDurationResponseBody>> => {
+    const { runFriendlyId } = params;
+    const runId = RunId.toId(runFriendlyId);
+
+    try {
+      const run = await prisma.taskRun.findFirst({
+        where: {
+          id: runId,
+          runtimeEnvironmentId: authentication.environment.id,
+        },
+      });
+
+      if (!run) {
+        throw new Response("You don't have permissions for this run", { status: 401 });
+      }
+
+      const idempotencyKeyExpiresAt = body.idempotencyKeyTTL
+        ? resolveIdempotencyKeyTTL(body.idempotencyKeyTTL)
+        : undefined;
+
+      const { waitpoint } = await engine.createDateTimeWaitpoint({
+        projectId: authentication.environment.project.id,
+        environmentId: authentication.environment.id,
+        completedAfter: body.date,
+        idempotencyKey: body.idempotencyKey,
+        idempotencyKeyExpiresAt: idempotencyKeyExpiresAt,
+      });
+
+      const waitResult = await engine.blockRunWithWaitpoint({
+        runId: run.id,
+        waitpoints: waitpoint.id,
+        environmentId: authentication.environment.id,
+        projectId: authentication.environment.project.id,
+        organizationId: authentication.environment.organization.id,
+        releaseConcurrency: {
+          releaseQueue: true,
+        },
+      });
+
+      return json({
+        waitUntil: body.date,
+        waitpoint: {
+          id: waitpoint.friendlyId,
+        },
+      });
+    } catch (error) {
+      logger.error("Failed to wait for duration dev", {
+        environmentId: authentication.environment.id,
+        error,
+      });
+      throw error;
+    }
+  }
+);
+
+export { action };

apps/webapp/app/routes/api.v1.worker-actions.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.wait.duration.ts

@@ -52,6 +52,7 @@ const { action } = createActionApiRoute(
         waitpoints: [waitpointId],
         environmentId: authentication.environment.id,
         projectId: authentication.environment.project.id,
+        organizationId: authentication.environment.organization.id,
         failAfter: timeout,
       });
 

apps/webapp/app/routes/engine.v1.runs.$runFriendlyId.wait.duration.ts

@@ -718,24 +718,6 @@ export class AuthenticatedWorkerInstance extends WithRunEngine {
     });
   }
 
-  async waitForDuration({
-    runFriendlyId,
-    snapshotFriendlyId,
-    date,
-  }: {
-    runFriendlyId: string;
-    snapshotFriendlyId: string;
-    date: Date;
-  }): Promise<WaitForDurationResult> {
-    return await this._engine.waitForDuration({
-      runId: fromFriendlyId(runFriendlyId),
-      snapshotId: fromFriendlyId(snapshotFriendlyId),
-      date,
-      workerId: this.workerInstanceId,
-      runnerId: this.runnerId,
-    });
-  }
-
   async getLatestSnapshot({ runFriendlyId }: { runFriendlyId: string }) {
     return await this._engine.getRunExecutionData({
       runId: fromFriendlyId(runFriendlyId),