Cleanup context and execution creation, cache stuff, add parent and root task run ids

Author: ericallam

apps/webapp/app/presenters/v3/SpanPresenter.server.ts

@@ -1,14 +1,9 @@
-import {
-  type MachinePresetName,
-  prettyPrintPacket,
-  SemanticInternalAttributes,
-  TaskRunError,
-} from "@trigger.dev/core/v3";
+import { prettyPrintPacket, SemanticInternalAttributes, TaskRunError } from "@trigger.dev/core/v3";
 import { getMaxDuration } from "@trigger.dev/core/v3/isomorphic";
 import { RUNNING_STATUSES } from "~/components/runs/v3/TaskRunStatus";
 import { logger } from "~/services/logger.server";
 import { eventRepository, rehydrateAttribute } from "~/v3/eventRepository.server";
-import { machinePresetFromName, machinePresetFromRun } from "~/v3/machinePresets.server";
+import { machinePresetFromRun } from "~/v3/machinePresets.server";
 import { getTaskEventStoreTableForRun, type TaskEventStoreTable } from "~/v3/taskEventStore.server";
 import { isFailedRunStatus, isFinalRunStatus } from "~/v3/taskStatus";
 import { BasePresenter } from "./basePresenter.server";
@@ -132,11 +127,7 @@ export class SpanPresenter extends BasePresenter {
         isTest: true,
         maxDurationInSeconds: true,
         taskEventStore: true,
-        tags: {
-          select: {
-            name: true,
-          },
-        },
+        runTags: true,
         machinePreset: true,
         lockedToVersion: {
           select: {
@@ -274,12 +265,12 @@ export class SpanPresenter extends BasePresenter {
     const context = {
       task: {
         id: run.taskIdentifier,
-        filePath: run.lockedBy?.filePath,
+        filePath: run.lockedBy?.filePath ?? "",
       },
       run: {
         id: run.friendlyId,
         createdAt: run.createdAt,
-        tags: run.tags.map((tag) => tag.name),
+        tags: run.runTags,
         isTest: run.isTest,
         idempotencyKey: run.idempotencyKey ?? undefined,
         startedAt: run.startedAt ?? run.createdAt,
@@ -292,6 +283,7 @@ export class SpanPresenter extends BasePresenter {
       },
       queue: {
         name: run.queue,
+        id: run.queue,
       },
       environment: {
         id: run.runtimeEnvironment.id,
@@ -342,7 +334,7 @@ export class SpanPresenter extends BasePresenter {
         isCustomQueue: !run.queue.startsWith("task/"),
         concurrencyKey: run.concurrencyKey,
       },
-      tags: run.tags.map((tag) => tag.name),
+      tags: run.runTags,
       baseCostInCents: run.baseCostInCents,
       costInCents: run.costInCents,
       totalCostInCents: run.costInCents + run.baseCostInCents,

apps/webapp/app/routes/resources.runs.$runParam.ts

@@ -76,6 +76,30 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
       lockedBy: {
         select: {
           filePath: true,
+          worker: {
+            select: {
+              deployment: {
+                select: {
+                  friendlyId: true,
+                  shortCode: true,
+                  version: true,
+                  runtime: true,
+                  runtimeVersion: true,
+                  git: true,
+                },
+              },
+            },
+          },
+        },
+      },
+      parentTaskRun: {
+        select: {
+          friendlyId: true,
+        },
+      },
+      rootTaskRun: {
+        select: {
+          friendlyId: true,
         },
       },
     },
@@ -163,6 +187,8 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
       baseCostInCents: run.baseCostInCents,
       maxAttempts: run.maxAttempts ?? undefined,
       version: run.lockedToVersion?.version,
+      parentTaskRunId: run.parentTaskRun?.friendlyId ?? undefined,
+      rootTaskRunId: run.rootTaskRun?.friendlyId ?? undefined,
     },
     queue: {
       name: run.queue,
@@ -184,6 +210,16 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
       name: run.project.name,
     },
     machine: run.machinePreset ? machinePresetFromRun(run) : undefined,
+    deployment: run.lockedBy?.worker.deployment
+      ? {
+          id: run.lockedBy.worker.deployment.friendlyId,
+          shortCode: run.lockedBy.worker.deployment.shortCode,
+          version: run.lockedBy.worker.deployment.version,
+          runtime: run.lockedBy.worker.deployment.runtime,
+          runtimeVersion: run.lockedBy.worker.deployment.runtimeVersion,
+          git: run.lockedBy.worker.deployment.git,
+        }
+      : undefined,
   };
 
   return typedjson({

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

@@ -4,6 +4,7 @@ import {
   TaskRunExecution,
   TaskRunExecutionRetry,
   TaskRunFailedExecutionResult,
+  V3TaskRunExecution,
 } from "@trigger.dev/core/v3";
 import type { Prisma, TaskRun } from "@trigger.dev/database";
 import * as semver from "semver";
@@ -129,7 +130,7 @@ export class FailedTaskRunRetryHelper extends BaseService {
   async #getRetriableAttemptExecution(
     run: TaskRunWithAttempts,
     completion: TaskRunFailedExecutionResult
-  ): Promise<TaskRunExecution | undefined> {
+  ): Promise<V3TaskRunExecution | undefined> {
     let attempt = run.attempts[0];
 
     // We need to create an attempt if:

apps/webapp/app/v3/marqs/devQueueConsumer.server.ts

@@ -1,6 +1,6 @@
 import { Context, ROOT_CONTEXT, Span, SpanKind, context, trace } from "@opentelemetry/api";
 import {
-  TaskRunExecution,
+  V3TaskRunExecution,
   TaskRunExecutionLazyAttemptPayload,
   TaskRunExecutionResult,
   TaskRunFailedExecutionResult,
@@ -138,7 +138,7 @@ export class DevQueueConsumer {
   public async taskAttemptCompleted(
     workerId: string,
     completion: TaskRunExecutionResult,
-    execution: TaskRunExecution
+    execution: V3TaskRunExecution
   ) {
     if (completion.ok) {
       this._taskSuccesses++;

apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts

@@ -11,8 +11,8 @@ import {
 import {
   AckCallbackResult,
   MachinePreset,
-  ProdTaskRunExecution,
-  ProdTaskRunExecutionPayload,
+  V3ProdTaskRunExecution,
+  V3ProdTaskRunExecutionPayload,
   TaskRunError,
   TaskRunErrorCodes,
   TaskRunExecution,
@@ -1679,7 +1679,7 @@ class SharedQueueTasks {
   private async _executionFromAttempt(
     attempt: AttemptForExecution,
     machinePreset?: MachinePreset
-  ): Promise<ProdTaskRunExecution> {
+  ): Promise<V3ProdTaskRunExecution> {
     const { backgroundWorkerTask, taskRun, queue } = attempt;
 
     if (!machinePreset) {
@@ -1693,7 +1693,7 @@ class SharedQueueTasks {
       dataType: taskRun.metadataType,
     });
 
-    const execution: ProdTaskRunExecution = {
+    const execution: V3ProdTaskRunExecution = {
       task: {
         id: backgroundWorkerTask.slug,
         filePath: backgroundWorkerTask.filePath,
@@ -1784,7 +1784,7 @@ class SharedQueueTasks {
     setToExecuting?: boolean;
     isRetrying?: boolean;
     skipStatusChecks?: boolean;
-  }): Promise<ProdTaskRunExecutionPayload | undefined> {
+  }): Promise<V3ProdTaskRunExecutionPayload | undefined> {
     const attempt = await prisma.taskRunAttempt.findFirst({
       where: {
         id,
@@ -1874,7 +1874,7 @@ class SharedQueueTasks {
       machinePreset
     );
 
-    const payload: ProdTaskRunExecutionPayload = {
+    const payload: V3ProdTaskRunExecutionPayload = {
       execution,
       traceContext: taskRun.traceContext as Record<string, unknown>,
       environment: variables.reduce((acc: Record<string, string>, curr) => {
@@ -1888,7 +1888,7 @@ class SharedQueueTasks {
 
   async getResumePayload(attemptId: string): Promise<
     | {
-        execution: ProdTaskRunExecution;
+        execution: V3ProdTaskRunExecution;
         completion: TaskRunExecutionResult;
       }
     | undefined
@@ -1927,7 +1927,7 @@ class SharedQueueTasks {
 
   async getResumePayloads(attemptIds: string[]): Promise<
     Array<{
-      execution: ProdTaskRunExecution;
+      execution: V3ProdTaskRunExecution;
       completion: TaskRunExecutionResult;
     }>
   > {
@@ -1985,7 +1985,7 @@ class SharedQueueTasks {
     id: string,
     setToExecuting?: boolean,
     isRetrying?: boolean
-  ): Promise<ProdTaskRunExecutionPayload | undefined> {
+  ): Promise<V3ProdTaskRunExecutionPayload | undefined> {
     const run = await prisma.taskRun.findFirst({
       where: {
         id,

apps/webapp/app/v3/services/completeAttempt.server.ts

@@ -2,13 +2,13 @@ import { Attributes } from "@opentelemetry/api";
 import {
   MachinePresetName,
   TaskRunContext,
-  TaskRunError,
   TaskRunErrorCodes,
   TaskRunExecution,
   TaskRunExecutionResult,
   TaskRunExecutionRetry,
   TaskRunFailedExecutionResult,
   TaskRunSuccessfulExecutionResult,
+  V3TaskRunExecution,
   flattenAttributes,
   isOOMRunError,
   sanitizeError,
@@ -60,7 +60,7 @@ export class CompleteAttemptService extends BaseService {
     checkpoint,
   }: {
     completion: TaskRunExecutionResult;
-    execution: TaskRunExecution;
+    execution: V3TaskRunExecution;
     env?: AuthenticatedEnvironment;
     checkpoint?: CheckpointData;
   }): Promise<"COMPLETED" | "RETRIED"> {
@@ -196,7 +196,7 @@ export class CompleteAttemptService extends BaseService {
     checkpoint,
   }: {
     completion: TaskRunFailedExecutionResult;
-    execution: TaskRunExecution;
+    execution: V3TaskRunExecution;
     taskRunAttempt: NonNullable<FoundAttempt>;
     env?: AuthenticatedEnvironment;
     checkpoint?: CheckpointData;
@@ -559,7 +559,7 @@ export class CompleteAttemptService extends BaseService {
     forceRequeue = false,
     oomMachine,
   }: {
-    execution: TaskRunExecution;
+    execution: V3TaskRunExecution;
     executionRetry: TaskRunExecutionRetry;
     executionRetryInferred: boolean;
     taskRunAttempt: NonNullable<FoundAttempt>;
@@ -648,7 +648,7 @@ export class CompleteAttemptService extends BaseService {
     executionRetryInferred,
     checkpoint,
   }: {
-    execution: TaskRunExecution;
+    execution: V3TaskRunExecution;
     taskRunAttempt: NonNullable<FoundAttempt>;
     executionRetry: TaskRunExecutionRetry;
     executionRetryInferred: boolean;

apps/webapp/app/v3/services/createTaskRunAttempt.server.ts

@@ -1,4 +1,4 @@
-import { parsePacket, TaskRunExecution } from "@trigger.dev/core/v3";
+import { parsePacket, V3TaskRunExecution } from "@trigger.dev/core/v3";
 import { TaskRun, TaskRunAttempt } from "@trigger.dev/database";
 import { MAX_TASK_RUN_ATTEMPTS } from "~/consts";
 import { $transaction, prisma, PrismaClientOrTransaction } from "~/db.server";
@@ -25,7 +25,7 @@ export class CreateTaskRunAttemptService extends BaseService {
     setToExecuting?: boolean;
     startAtZero?: boolean;
   }): Promise<{
-    execution: TaskRunExecution;
+    execution: V3TaskRunExecution;
     run: TaskRun;
     attempt: TaskRunAttempt;
   }> {
@@ -189,7 +189,7 @@ export class CreateTaskRunAttemptService extends BaseService {
         dataType: taskRun.metadataType,
       });
 
-      const execution: TaskRunExecution = {
+      const execution: V3TaskRunExecution = {
         task: {
           id: lockedBy.slug,
           filePath: lockedBy.filePath,

internal-packages/cache/README.md

@@ -0,0 +1,3 @@
+# Redis
+
+This is a simple package that is used to return a valid Redis client and provides an error callback. It will log and swallow errors if they're not handled.

internal-packages/cache/package.json

@@ -0,0 +1,18 @@
+{
+  "name": "@internal/cache",
+  "private": true,
+  "version": "0.0.1",
+  "main": "./src/index.ts",
+  "types": "./src/index.ts",
+  "type": "module",
+  "dependencies": {
+    "@unkey/cache": "^1.5.0",
+    "@unkey/error": "^0.2.0",
+    "@trigger.dev/core": "workspace:*",
+    "@internal/redis": "workspace:*",
+    "superjson": "^2.2.1"
+  },
+  "scripts": {
+    "typecheck": "tsc --noEmit"
+  }
+}

internal-packages/cache/src/index.ts

@@ -0,0 +1,8 @@
+export {
+  createCache,
+  DefaultStatefulContext,
+  Namespace,
+  type Cache as UnkeyCache,
+} from "@unkey/cache";
+export { MemoryStore } from "@unkey/cache/stores";
+export { RedisCacheStore } from "./stores/redis.js";