Cancel run events which then propogate cancellation status to span ancestors

ericallam · ericallam · commit b89e5ab40d20 · 2025-09-19T09:46:18.000+01:00
diff --git a/apps/webapp/app/presenters/v3/SpanPresenter.server.ts b/apps/webapp/app/presenters/v3/SpanPresenter.server.ts
@@ -251,6 +251,7 @@ export class SpanPresenter extends BasePresenter {
       engine: run.engine,
       region,
       workerQueue: run.workerQueue,
+      traceId: run.traceId,
       spanId: run.spanId,
       isCached: !!span.originalRun,
       machinePreset: machine?.name,
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
@@ -772,6 +772,14 @@ function RunBody({
                       <Property.Label>Worker queue</Property.Label>
                       <Property.Value>{run.workerQueue}</Property.Value>
                     </Property.Item>
+                    <Property.Item>
+                      <Property.Label>Trace ID</Property.Label>
+                      <Property.Value>{run.traceId}</Property.Value>
+                    </Property.Item>
+                    <Property.Item>
+                      <Property.Label>Span ID</Property.Label>
+                      <Property.Value>{run.spanId}</Property.Value>
+                    </Property.Item>
                   </div>
                 )}
               </Property.Table>
diff --git a/apps/webapp/app/v3/eventRepository.server.ts b/apps/webapp/app/v3/eventRepository.server.ts
@@ -21,7 +21,14 @@ import {
   unflattenAttributes,
 } from "@trigger.dev/core/v3";
 import { parseTraceparent, serializeTraceparent } from "@trigger.dev/core/v3/isomorphic";
-import { Prisma, TaskEvent, TaskEventKind, TaskEventStatus } from "@trigger.dev/database";
+import {
+  Prisma,
+  RuntimeEnvironmentType,
+  TaskEvent,
+  TaskEventKind,
+  TaskEventStatus,
+  TaskRun,
+} from "@trigger.dev/database";
 import { nanoid } from "nanoid";
 import { createHash } from "node:crypto";
 import { EventEmitter } from "node:stream";
@@ -350,18 +357,44 @@ export class EventRepository {
     return completedEvent;
   }
 
-  async cancelEvent(event: TaskEventRecord, cancelledAt: Date, reason: string) {
-    if (!event.isPartial) {
-      return;
-    }
+  async cancelRunEvent({
+    reason,
+    run,
+    cancelledAt,
+    projectRef,
+    environmentType,
+    organizationId,
+  }: {
+    reason: string;
+    run: TaskRun;
+    cancelledAt: Date;
+    projectRef: string;
+    organizationId: string;
+    environmentType: RuntimeEnvironmentType;
+  }) {
+    const startTime = convertDateToNanoseconds(run.createdAt);
 
     await this.insertImmediate({
-      ...omit(event, "id"),
+      message: run.taskIdentifier,
+      serviceName: "api server",
+      serviceNamespace: "trigger.dev",
+      level: "TRACE",
+      kind: "SERVER",
+      traceId: run.traceId,
+      spanId: run.spanId,
+      parentId: run.parentSpanId,
+      runId: run.id,
+      taskSlug: run.taskIdentifier,
+      projectRef,
+      projectId: run.projectId,
+      environmentId: run.runtimeEnvironmentId,
+      environmentType,
+      organizationId,
       isPartial: false,
-      isError: false,
+      isError: true,
       isCancelled: true,
       status: "ERROR",
-      links: event.links ?? [],
+      runIsTest: run.isTest,
       events: [
         {
           name: "cancellation",
@@ -370,56 +403,18 @@ export class EventRepository {
             reason,
           },
         },
-        ...((event.events as any[]) ?? []),
       ],
-      duration: calculateDurationFromStart(event.startTime, cancelledAt),
-      properties: event.properties as Attributes,
-      metadata: event.metadata as Attributes,
-      style: event.style as Attributes,
-      output: event.output as Attributes,
-      outputType: event.outputType,
-      payload: event.payload as Attributes,
-      payloadType: event.payloadType,
+      startTime,
+      properties: {},
+      metadata: undefined,
+      style: undefined,
+      duration: calculateDurationFromStart(startTime, cancelledAt),
+      output: undefined,
+      payload: undefined,
+      payloadType: undefined,
     });
   }
 
-  async cancelEvents(events: TaskEventRecord[], cancelledAt: Date, reason: string) {
-    const eventsToCancel = events.filter((event) => event.isPartial);
-
-    if (eventsToCancel.length === 0) {
-      return;
-    }
-
-    await this.insertMany(
-      eventsToCancel.map((event) => ({
-        ...omit(event, "id"),
-        isPartial: false,
-        isError: false,
-        isCancelled: true,
-        status: "ERROR",
-        links: event.links ?? [],
-        events: [
-          {
-            name: "cancellation",
-            time: cancelledAt,
-            properties: {
-              reason,
-            },
-          },
-          ...((event.events as any[]) ?? []),
-        ],
-        duration: calculateDurationFromStart(event.startTime, cancelledAt),
-        properties: event.properties as Attributes,
-        metadata: event.metadata as Attributes,
-        style: event.style as Attributes,
-        output: event.output as Attributes,
-        outputType: event.outputType,
-        payload: event.payload as Attributes,
-        payloadType: event.payloadType,
-      }))
-    );
-  }
-
   async crashEvent({
     event,
     crashedAt,
@@ -567,7 +562,19 @@ export class EventRepository {
         }
 
         if (event.isCancelled || !event.isPartial) {
-          eventsBySpanId.set(event.spanId, event);
+          // If we have a cancelled event and an existing partial event,
+          // merge them: use cancelled event data but preserve style from the partial event
+          if (event.isCancelled && existingEvent.isPartial && !existingEvent.isCancelled) {
+            const mergedEvent: PreparedEvent = {
+              ...event, // Use cancelled event as base (has correct timing, status, events)
+              // Preserve style from the original partial event
+              style: existingEvent.style,
+            };
+            eventsBySpanId.set(event.spanId, mergedEvent);
+          } else {
+            // For non-cancelled events or other cases, use the standard replacement logic
+            eventsBySpanId.set(event.spanId, event);
+          }
         }
       }
 
@@ -583,6 +590,9 @@ export class EventRepository {
           event.duration
         );
 
+        const isCancelled =
+          event.isCancelled === true ? true : event.isPartial && ancestorCancelled;
+
         const span = {
           id: event.spanId,
           parentId: event.parentId ?? undefined,
@@ -592,9 +602,9 @@ export class EventRepository {
             message: event.message,
             style: event.style,
             duration,
-            isError: event.isError,
+            isError: isCancelled ? false : event.isError,
             isPartial: ancestorCancelled ? false : event.isPartial,
-            isCancelled: event.isCancelled === true ? true : event.isPartial && ancestorCancelled,
+            isCancelled,
             isDebug: event.kind === TaskEventKind.LOG,
             startTime: getDateFromNanoseconds(event.startTime),
             level: event.level,
@@ -618,6 +628,8 @@ export class EventRepository {
         return;
       }
 
+      logger.debug("[getTraceSummary] result", { rootSpan, spans });
+
       return {
         rootSpan,
         spans,
@@ -1141,11 +1153,6 @@ export class EventRepository {
   ): Promise<TResult> {
     const propagatedContext = extractContextFromCarrier(options.context ?? {});
 
-    logger.debug("[otelContext]", {
-      propagatedContext,
-      options,
-    });
-
     const start = process.hrtime.bigint();
     const startTime = options.startTime ?? getNowInNanoseconds();
 
@@ -1895,6 +1902,10 @@ export function getDateFromNanoseconds(nanoseconds: bigint) {
   return new Date(Number(nanoseconds) / 1_000_000);
 }
 
+function convertDateToNanoseconds(date: Date) {
+  return BigInt(date.getTime()) * BigInt(1_000_000);
+}
+
 function nanosecondsToMilliseconds(nanoseconds: bigint | number): number {
   return Number(nanoseconds) / 1_000_000;
 }
diff --git a/apps/webapp/app/v3/runEngineHandlers.server.ts b/apps/webapp/app/v3/runEngineHandlers.server.ts
@@ -299,20 +299,42 @@ export function registerRunEngineEventBusHandlers() {
 
   engine.eventBus.on("runCancelled", async ({ time, run }) => {
     try {
-      const eventStore = getTaskEventStoreTableForRun(run);
-
-      const inProgressEvents = await eventRepository.queryIncompleteEvents(
-        eventStore,
-        {
-          runId: run.friendlyId,
+      const taskRun = await $replica.taskRun.findFirst({
+        where: {
+          id: run.id,
         },
-        run.createdAt,
-        run.completedAt ?? undefined
-      );
+        include: {
+          project: {
+            select: {
+              externalRef: true,
+            },
+          },
+          runtimeEnvironment: {
+            select: {
+              type: true,
+              organizationId: true,
+            },
+          },
+        },
+      });
+
+      if (!taskRun) {
+        logger.error("[runCancelled] Task run not found", {
+          runId: run.id,
+        });
+        return;
+      }
 
       const error = createJsonErrorObject(run.error);
 
-      await eventRepository.cancelEvents(inProgressEvents, time, error.message);
+      await eventRepository.cancelRunEvent({
+        reason: error.message,
+        run: taskRun,
+        cancelledAt: time,
+        projectRef: taskRun.project.externalRef,
+        organizationId: taskRun.runtimeEnvironment.organizationId,
+        environmentType: taskRun.runtimeEnvironment.type,
+      });
     } catch (error) {
       logger.error("[runCancelled] Failed to cancel event", {
         error: error instanceof Error ? error.message : error,
diff --git a/apps/webapp/app/v3/services/cancelAttempt.server.ts b/apps/webapp/app/v3/services/cancelAttempt.server.ts
@@ -72,25 +72,6 @@ export class CancelAttemptService extends BaseService {
           error: isCancellable ? { type: "STRING_ERROR", raw: reason } : undefined,
         });
       });
-
-      const inProgressEvents = await eventRepository.queryIncompleteEvents(
-        getTaskEventStoreTableForRun(taskRunAttempt.taskRun),
-        {
-          runId: taskRunAttempt.taskRun.friendlyId,
-        },
-        taskRunAttempt.taskRun.createdAt,
-        taskRunAttempt.taskRun.completedAt ?? undefined
-      );
-
-      logger.debug("Cancelling in-progress events", {
-        inProgressEvents: inProgressEvents.map((event) => event.id),
-      });
-
-      await Promise.all(
-        inProgressEvents.map((event) => {
-          return eventRepository.cancelEvent(event, cancelledAt, reason);
-        })
-      );
     });
   }
 }
diff --git a/apps/webapp/app/v3/services/cancelTaskRunV1.server.ts b/apps/webapp/app/v3/services/cancelTaskRunV1.server.ts
@@ -1,4 +1,4 @@
-import { type Prisma, type TaskRun } from "@trigger.dev/database";
+import { type Prisma } from "@trigger.dev/database";
 import assertNever from "assert-never";
 import { logger } from "~/services/logger.server";
 import { eventRepository } from "../eventRepository.server";
@@ -8,9 +8,9 @@ import { CANCELLABLE_ATTEMPT_STATUSES, isCancellableRunStatus } from "../taskSta
 import { BaseService } from "./baseService.server";
 import { CancelAttemptService } from "./cancelAttempt.server";
 import { CancelTaskAttemptDependenciesService } from "./cancelTaskAttemptDependencies.server";
-import { FinalizeTaskRunService } from "./finalizeTaskRun.server";
-import { getTaskEventStoreTableForRun } from "../taskEventStore.server";
 import { CancelableTaskRun } from "./cancelTaskRun.server";
+import { FinalizeTaskRunService } from "./finalizeTaskRun.server";
+import { tryCatch } from "@trigger.dev/core/utils";
 
 type ExtendedTaskRun = Prisma.TaskRunGetPayload<{
   include: {
@@ -92,6 +92,7 @@ export class CancelTaskRunServiceV1 extends BaseService {
         },
         runtimeEnvironment: true,
         lockedToVersion: true,
+        project: true,
       },
       attemptStatus: "CANCELED",
       error: {
@@ -100,21 +101,23 @@ export class CancelTaskRunServiceV1 extends BaseService {
       },
     });
 
-    const inProgressEvents = await eventRepository.queryIncompleteEvents(
-      getTaskEventStoreTableForRun(taskRun),
-      {
-        runId: taskRun.friendlyId,
-      },
-      taskRun.createdAt,
-      taskRun.completedAt ?? undefined
+    const [cancelRunEventError] = await tryCatch(
+      eventRepository.cancelRunEvent({
+        reason: opts.reason,
+        run: cancelledTaskRun,
+        cancelledAt: opts.cancelledAt,
+        projectRef: cancelledTaskRun.project.externalRef,
+        organizationId: cancelledTaskRun.runtimeEnvironment.organizationId,
+        environmentType: cancelledTaskRun.runtimeEnvironment.type,
+      })
     );
 
-    logger.debug("Cancelling in-progress events", {
-      inProgressEvents: inProgressEvents.map((event) => event.id),
-      eventCount: inProgressEvents.length,
-    });
-
-    await eventRepository.cancelEvents(inProgressEvents, opts.cancelledAt, opts.reason);
+    if (cancelRunEventError) {
+      logger.error("[CancelTaskRunServiceV1] Failed to cancel run event", {
+        error: cancelRunEventError,
+        runId: cancelledTaskRun.id,
+      });
+    }
 
     // Cancel any in progress attempts
     if (opts.cancelAttempts) {
diff --git a/packages/core/src/v3/workers/taskExecutor.ts b/packages/core/src/v3/workers/taskExecutor.ts
@@ -352,8 +352,7 @@ export class TaskExecutor {
             ? runTimelineMetrics.convertMetricsToSpanEvents()
             : undefined,
       },
-      traceContext.extractContext(),
-      signal
+      traceContext.extractContext()
     );
 
     return { result };
