Fix controller waitpoint resolution, suspendable state, and snapshot race conditions

.changeset/plenty-dolphins-act.md

@@ -0,0 +1,8 @@
+---
+"trigger.dev": patch
+"@trigger.dev/core": patch
+---
+
+- Correctly resolve waitpoints that come in early
+- Ensure correct state before requesting suspension
+- Fix race conditions in snapshot processing

.changeset/sweet-dolphins-invent.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Always print full deploy logs in CI

.github/workflows/publish-worker-re2.yml

@@ -85,11 +85,11 @@ jobs:
           REPOSITORY: ${{ steps.get_repository.outputs.repo }}
           IMAGE_TAG: ${{ steps.get_tag.outputs.tag }}
 
-      - name: 🐙 Push 'v3' tag to GitHub Container Registry
-        if: steps.get_tag.outputs.is_semver == 'true'
-        run: |
-          docker tag infra_image "$REGISTRY/$REPOSITORY:v3"
-          docker push "$REGISTRY/$REPOSITORY:v3"
-        env:
-          REGISTRY: ghcr.io/triggerdotdev
-          REPOSITORY: ${{ steps.get_repository.outputs.repo }}
+      # - name: 🐙 Push 'v3' tag to GitHub Container Registry
+      #   if: steps.get_tag.outputs.is_semver == 'true'
+      #   run: |
+      #     docker tag infra_image "$REGISTRY/$REPOSITORY:v3"
+      #     docker push "$REGISTRY/$REPOSITORY:v3"
+      #   env:
+      #     REGISTRY: ghcr.io/triggerdotdev
+      #     REPOSITORY: ${{ steps.get_repository.outputs.repo }}

.github/workflows/publish-worker.yml

@@ -77,11 +77,11 @@ jobs:
           REPOSITORY: ${{ steps.get_repository.outputs.repo }}
           IMAGE_TAG: ${{ steps.get_tag.outputs.tag }}
 
-      - name: 🐙 Push 'v3' tag to GitHub Container Registry
-        if: steps.get_tag.outputs.is_semver == 'true'
-        run: |
-          docker tag infra_image "$REGISTRY/$REPOSITORY:v3"
-          docker push "$REGISTRY/$REPOSITORY:v3"
-        env:
-          REGISTRY: ghcr.io/triggerdotdev
-          REPOSITORY: ${{ steps.get_repository.outputs.repo }}
+      # - name: 🐙 Push 'v3' tag to GitHub Container Registry
+      #   if: steps.get_tag.outputs.is_semver == 'true'
+      #   run: |
+      #     docker tag infra_image "$REGISTRY/$REPOSITORY:v3"
+      #     docker push "$REGISTRY/$REPOSITORY:v3"
+      #   env:
+      #     REGISTRY: ghcr.io/triggerdotdev
+      #     REPOSITORY: ${{ steps.get_repository.outputs.repo }}

.github/workflows/publish.yml

@@ -56,15 +56,22 @@ jobs:
     secrets: inherit
 
   publish-webapp:
-    needs: [typecheck]
+    needs: [typecheck, units]
     uses: ./.github/workflows/publish-webapp.yml
     secrets: inherit
     with:
       image_tag: ${{ inputs.image_tag }}
 
   publish-worker:
-    needs: [typecheck]
+    needs: [typecheck, units]
     uses: ./.github/workflows/publish-worker.yml
     secrets: inherit
     with:
       image_tag: ${{ inputs.image_tag }}
+
+  publish-worker-v4:
+    needs: [typecheck, units]
+    uses: ./.github/workflows/publish-worker-v4.yml
+    secrets: inherit
+    with:
+      image_tag: ${{ inputs.image_tag }}

apps/webapp/app/runEngine/validators/triggerTaskValidator.ts

@@ -46,7 +46,7 @@ export class DefaultTriggerTaskValidator implements TriggerTaskValidator {
 
     const result = await getEntitlement(environment.organizationId);
 
-    if (!result || result.hasAccess === false) {
+    if (result && result.hasAccess === false) {
       return {
         ok: false,
         error: new OutOfEntitlementError(),

apps/webapp/app/v3/runEngineHandlers.server.ts

@@ -401,7 +401,7 @@ export function registerRunEngineEventBusHandlers() {
   engine.eventBus.on("executionSnapshotCreated", async ({ time, run, snapshot }) => {
     const eventResult = await recordRunDebugLog(
       run.id,
-      `${snapshot.executionStatus} - ${snapshot.description}`,
+      `[engine] ${snapshot.executionStatus} - ${snapshot.description}`,
       {
         attributes: {
           properties: {
@@ -450,6 +450,7 @@ export function registerRunEngineEventBusHandlers() {
       // Record notification event
       const eventResult = await recordRunDebugLog(
         run.id,
+        // don't prefix this with [engine] - "run:notify" is the correct prefix
         `run:notify platform -> supervisor: ${snapshot.executionStatus}`,
         {
           attributes: {
@@ -479,6 +480,7 @@ export function registerRunEngineEventBusHandlers() {
       // Record notification event
       const eventResult = await recordRunDebugLog(
         run.id,
+        // don't prefix this with [engine] - "run:notify" is the correct prefix
         `run:notify ERROR platform -> supervisor: ${snapshot.executionStatus}`,
         {
           attributes: {
@@ -505,7 +507,7 @@ export function registerRunEngineEventBusHandlers() {
   engine.eventBus.on("incomingCheckpointDiscarded", async ({ time, run, snapshot, checkpoint }) => {
     const eventResult = await recordRunDebugLog(
       run.id,
-      `Checkpoint discarded: ${checkpoint.discardReason}`,
+      `[engine] Checkpoint discarded: ${checkpoint.discardReason}`,
       {
         attributes: {
           properties: {

internal-packages/testcontainers/src/index.ts

@@ -53,7 +53,9 @@ const postgresContainer = async (
   try {
     await use(container);
   } finally {
-    await container.stop();
+    // WARNING: Testcontainers by default will not wait until the container has stopped. It will simply issue the stop command and return immediately.
+    // If you need to wait for the container to be stopped, you can provide a timeout. The unit of timeout option here is second
+    await container.stop({ timeout: 10 });
   }
 };
 
@@ -92,7 +94,9 @@ const redisContainer = async (
   try {
     await use(container);
   } finally {
-    await container.stop();
+    // WARNING: Testcontainers by default will not wait until the container has stopped. It will simply issue the stop command and return immediately.
+    // If you need to wait for the container to be stopped, you can provide a timeout. The unit of timeout option here is second
+    await container.stop({ timeout: 10 });
   }
 };
 
@@ -142,7 +146,9 @@ const electricOrigin = async (
   try {
     await use(origin);
   } finally {
-    await container.stop();
+    // WARNING: Testcontainers by default will not wait until the container has stopped. It will simply issue the stop command and return immediately.
+    // If you need to wait for the container to be stopped, you can provide a timeout. The unit of timeout option here is second
+    await container.stop({ timeout: 10 });
   }
 };
 

packages/cli-v3/package.json

@@ -39,6 +39,9 @@
       "esm"
     ],
     "project": "./tsconfig.src.json",
+    "exclude": [
+      "**/*.test.ts"
+    ],
     "exports": {
       "./package.json": "./package.json",
       ".": "./src/index.ts"
@@ -70,6 +73,7 @@
     "typecheck": "tsc -p tsconfig.src.json --noEmit",
     "build": "tshy && pnpm run update-version",
     "dev": "tshy --watch",
+    "test": "vitest",
     "test:e2e": "vitest --run -c ./e2e/vitest.config.ts",
     "update-version": "tsx ../../scripts/updateVersion.ts"
   },

packages/cli-v3/src/commands/deploy.ts

@@ -1,10 +1,11 @@
-import { intro, outro } from "@clack/prompts";
+import { intro, log, outro } from "@clack/prompts";
 import { prepareDeploymentError } from "@trigger.dev/core/v3";
 import { InitializeDeploymentResponseBody } from "@trigger.dev/core/v3/schemas";
 import { Command, Option as CommandOption } from "commander";
 import { resolve } from "node:path";
 import { x } from "tinyexec";
 import { z } from "zod";
+import { isCI } from "std-env";
 import { CliApiClient } from "../apiClient.js";
 import { buildWorker } from "../build/buildWorker.js";
 import { resolveAlwaysExternal } from "../build/externals.js";
@@ -321,24 +322,24 @@ async function _deployCommand(dir: string, options: DeployCommandOptions) {
 
   const version = deployment.version;
 
-  const deploymentLink = cliLink(
-    "View deployment",
-    `${authorization.dashboardUrl}/projects/v3/${resolvedConfig.project}/deployments/${deployment.shortCode}`
-  );
+  const rawDeploymentLink = `${authorization.dashboardUrl}/projects/v3/${resolvedConfig.project}/deployments/${deployment.shortCode}`;
+  const rawTestLink = `${authorization.dashboardUrl}/projects/v3/${
+    resolvedConfig.project
+  }/test?environment=${options.env === "prod" ? "prod" : "stg"}`;
 
-  const testLink = cliLink(
-    "Test tasks",
-    `${authorization.dashboardUrl}/projects/v3/${resolvedConfig.project}/test?environment=${
-      options.env === "prod" ? "prod" : "stg"
-    }`
-  );
+  const deploymentLink = cliLink("View deployment", rawDeploymentLink);
+  const testLink = cliLink("Test tasks", rawTestLink);
 
   const $spinner = spinner();
 
-  if (isLinksSupported) {
-    $spinner.start(`Building version ${version} ${deploymentLink}`);
+  if (isCI) {
+    log.step(`Building version ${version}\n`);
   } else {
-    $spinner.start(`Building version ${version}`);
+    if (isLinksSupported) {
+      $spinner.start(`Building version ${version} ${deploymentLink}`);
+    } else {
+      $spinner.start(`Building version ${version}`);
+    }
   }
 
   const selfHostedRegistryHost = deployment.registryHost ?? options.registry;
@@ -368,6 +369,11 @@ async function _deployCommand(dir: string, options: DeployCommandOptions) {
     buildEnvVars: buildManifest.build.env,
     network: options.network,
     onLog: (logMessage) => {
+      if (isCI) {
+        console.log(logMessage);
+        return;
+      }
+
       if (isLinksSupported) {
         $spinner.message(`Building version ${version} ${deploymentLink}: ${logMessage}`);
       } else {
@@ -441,10 +447,14 @@ async function _deployCommand(dir: string, options: DeployCommandOptions) {
       }`
     : `${buildResult.image}${buildResult.digest ? `@${buildResult.digest}` : ""}`;
 
-  if (isLinksSupported) {
-    $spinner.message(`Deploying version ${version} ${deploymentLink}`);
+  if (isCI) {
+    log.step(`Deploying version ${version}\n`);
   } else {
-    $spinner.message(`Deploying version ${version}`);
+    if (isLinksSupported) {
+      $spinner.message(`Deploying version ${version} ${deploymentLink}`);
+    } else {
+      $spinner.message(`Deploying version ${version}`);
+    }
   }
 
   const finalizeResponse = await projectClient.client.finalizeDeployment(
@@ -455,6 +465,11 @@ async function _deployCommand(dir: string, options: DeployCommandOptions) {
       skipPromotion: options.skipPromotion,
     },
     (logMessage) => {
+      if (isCI) {
+        console.log(logMessage);
+        return;
+      }
+
       if (isLinksSupported) {
         $spinner.message(`Deploying version ${version} ${deploymentLink}: ${logMessage}`);
       } else {
@@ -475,7 +490,11 @@ async function _deployCommand(dir: string, options: DeployCommandOptions) {
     throw new SkipLoggingError("Failed to finalize deployment");
   }
 
-  $spinner.stop(`Successfully deployed version ${version}`);
+  if (isCI) {
+    log.step(`Successfully deployed version ${version}`);
+  } else {
+    $spinner.stop(`Successfully deployed version ${version}`);
+  }
 
   const taskCount = deploymentWithWorker.worker?.tasks.length ?? 0;
 
@@ -485,6 +504,14 @@ async function _deployCommand(dir: string, options: DeployCommandOptions) {
     }`
   );
 
+  if (!isLinksSupported) {
+    console.log("View deployment");
+    console.log(rawDeploymentLink);
+    console.log(); // new line
+    console.log("Test tasks");
+    console.log(rawTestLink);
+  }
+
   setGithubActionsOutputAndEnvVars({
     envVars: {
       TRIGGER_DEPLOYMENT_VERSION: version,

packages/cli-v3/src/entryPoints/dev-run-worker.ts

@@ -35,7 +35,7 @@ import {
   getEnvVar,
   getNumberEnvVar,
   logLevels,
-  ManagedRuntimeManager,
+  SharedRuntimeManager,
   OtelTaskLogger,
   populateEnv,
   StandardLifecycleHooksManager,
@@ -452,20 +452,11 @@ const zodIpc = new ZodIpcConnection({
         });
       }
     },
-    TASK_RUN_COMPLETED_NOTIFICATION: async () => {
-      await managedWorkerRuntime.completeWaitpoints([]);
-    },
-    WAIT_COMPLETED_NOTIFICATION: async () => {
-      await managedWorkerRuntime.completeWaitpoints([]);
-    },
     FLUSH: async ({ timeoutInMs }, sender) => {
       await flushAll(timeoutInMs);
     },
-    WAITPOINT_CREATED: async ({ wait, waitpoint }) => {
-      managedWorkerRuntime.associateWaitWithWaitpoint(wait.id, waitpoint.id);
-    },
-    WAITPOINT_COMPLETED: async ({ waitpoint }) => {
-      managedWorkerRuntime.completeWaitpoints([waitpoint]);
+    RESOLVE_WAITPOINT: async ({ waitpoint }) => {
+      sharedWorkerRuntime.resolveWaitpoints([waitpoint]);
     },
   },
 });
@@ -537,8 +528,8 @@ async function flushMetadata(timeoutInMs: number = 10_000) {
   };
 }
 
-const managedWorkerRuntime = new ManagedRuntimeManager(zodIpc, showInternalLogs);
-runtime.setGlobalRuntimeManager(managedWorkerRuntime);
+const sharedWorkerRuntime = new SharedRuntimeManager(zodIpc, showInternalLogs);
+runtime.setGlobalRuntimeManager(sharedWorkerRuntime);
 
 process.title = "trigger-managed-worker";
 

packages/cli-v3/src/entryPoints/managed-run-worker.ts

@@ -34,7 +34,7 @@ import {
   getEnvVar,
   getNumberEnvVar,
   logLevels,
-  ManagedRuntimeManager,
+  SharedRuntimeManager,
   OtelTaskLogger,
   populateEnv,
   ProdUsageManager,
@@ -445,20 +445,11 @@ const zodIpc = new ZodIpcConnection({
         });
       }
     },
-    TASK_RUN_COMPLETED_NOTIFICATION: async () => {
-      await managedWorkerRuntime.completeWaitpoints([]);
-    },
-    WAIT_COMPLETED_NOTIFICATION: async () => {
-      await managedWorkerRuntime.completeWaitpoints([]);
-    },
     FLUSH: async ({ timeoutInMs }, sender) => {
       await flushAll(timeoutInMs);
     },
-    WAITPOINT_CREATED: async ({ wait, waitpoint }) => {
-      managedWorkerRuntime.associateWaitWithWaitpoint(wait.id, waitpoint.id);
-    },
-    WAITPOINT_COMPLETED: async ({ waitpoint }) => {
-      managedWorkerRuntime.completeWaitpoints([waitpoint]);
+    RESOLVE_WAITPOINT: async ({ waitpoint }) => {
+      sharedWorkerRuntime.resolveWaitpoints([waitpoint]);
     },
   },
 });
@@ -565,9 +556,9 @@ function initializeUsageManager({
   timeout.setGlobalManager(new UsageTimeoutManager(devUsageManager));
 }
 
-const managedWorkerRuntime = new ManagedRuntimeManager(zodIpc, true);
+const sharedWorkerRuntime = new SharedRuntimeManager(zodIpc, true);
 
-runtime.setGlobalRuntimeManager(managedWorkerRuntime);
+runtime.setGlobalRuntimeManager(sharedWorkerRuntime);
 
 process.title = "trigger-managed-worker";