github-featured/triggerdotdev--trigger.dev
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

feat(webapp): completing spans server-side no longer write-after-read, improving efficiency and perf (#2530)

Browse Source 
* Cancel run events which then propogate cancellation status to span ancestors

* WIP

* convert closing cached run spans to new system

* converted expired complete span event to new method

* move v3 over to new methods

* Convert getDetailedTraceSummary to use the new ancestor override stuff

* remove debug logs

* Don't return UNSPECIFIED task events in getRunEvents

* fix the call site for cancelling run event in v3

* Add changeset

* remove methods
This commit is contained in:
Eric Allam
2025-09-19 13:39:48 +01:00
committed by GitHub
parent 365adc24a6
commit 87b3603b23
18 changed files with 1120 additions and 726 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/angry-trainers-perform.md
+5
View File
@@ -0,0 +1,5 @@
---
"trigger.dev": patch
---
Stop failing attempt spans when a run is cancelled
apps/webapp/app/presenters/v3/SpanPresenter.server.ts
+1
View File
@@ -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,
apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.spans.$spanParam/route.tsx
+8
View File
@@ -840,6 +840,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>
apps/webapp/app/runEngine/concerns/idempotencyKeys.server.ts
+8 -1
View File
@@ -90,11 +90,18 @@ export class IdempotencyKeyConcern {
isError: associatedWaitpoint.outputIsError,
},
async (event) => {
const spanId =
request.options?.parentAsLinkType === "replay"
? event.spanId
: event.traceparent?.spanId
? `${event.traceparent.spanId}:${event.spanId}`
: event.spanId;
//block run with waitpoint
await this.engine.blockRunWithWaitpoint({
runId: RunId.fromFriendlyId(parentRunId),
waitpoints: associatedWaitpoint.id,
spanIdToComplete: event.spanId,
spanIdToComplete: spanId,
batch: request.options?.batchId
? {
id: request.options.batchId,
apps/webapp/app/v3/eventRepository.server.ts
+636 -307
View File
File diff suppressed because it is too large Load Diff
apps/webapp/app/v3/runEngineHandlers.server.ts
+308 -244
View File
@@ -1,59 +1,66 @@
import { $replica, prisma } from "~/db.server";
import { tryCatch } from "@trigger.dev/core/utils";
import { createJsonErrorObject, sanitizeError } from "@trigger.dev/core/v3";
import { RunId } from "@trigger.dev/core/v3/isomorphic";
import { $replica } from "~/db.server";
import { env } from "~/env.server";
import { findEnvironmentFromRun } from "~/models/runtimeEnvironment.server";
import { logger } from "~/services/logger.server";
import { updateMetadataService } from "~/services/metadata/updateMetadataInstance.server";
import { reportInvocationUsage } from "~/services/platform.v3.server";
import { MetadataTooLargeError } from "~/utils/packets";
import {
createExceptionPropertiesFromError,
eventRepository,
recordRunDebugLog,
} from "./eventRepository.server";
import { createJsonErrorObject, sanitizeError } from "@trigger.dev/core/v3";
import { logger } from "~/services/logger.server";
import { safeJsonParse } from "~/utils/json";
import type { Attributes } from "@opentelemetry/api";
import { reportInvocationUsage } from "~/services/platform.v3.server";
import { roomFromFriendlyRunId, socketIo } from "./handleSocketIo.server";
import { engine } from "./runEngine.server";
import { PerformTaskRunAlertsService } from "./services/alerts/performTaskRunAlerts.server";
import { RunId } from "@trigger.dev/core/v3/isomorphic";
import { updateMetadataService } from "~/services/metadata/updateMetadataInstance.server";
import { findEnvironmentFromRun } from "~/models/runtimeEnvironment.server";
import { env } from "~/env.server";
import { getTaskEventStoreTableForRun } from "./taskEventStore.server";
import { MetadataTooLargeError } from "~/utils/packets";
export function registerRunEngineEventBusHandlers() {
engine.eventBus.on("runSucceeded", async ({ time, run }) => {
try {
const completedEvent = await eventRepository.completeEvent(
getTaskEventStoreTableForRun(run),
run.spanId,
run.createdAt,
run.completedAt ?? undefined,
{
endTime: time,
attributes: {
isError: false,
output:
run.outputType === "application/store" || run.outputType === "text/plain"
? run.output
: run.output
? (safeJsonParse(run.output) as Attributes)
: undefined,
outputType: run.outputType,
},
}
);
const [taskRunError, taskRun] = await tryCatch(
$replica.taskRun.findFirstOrThrow({
where: {
id: run.id,
},
select: {
id: true,
friendlyId: true,
traceId: true,
spanId: true,
parentSpanId: true,
createdAt: true,
completedAt: true,
taskIdentifier: true,
projectId: true,
runtimeEnvironmentId: true,
environmentType: true,
isTest: true,
organizationId: true,
},
})
);
if (!completedEvent) {
logger.error("[runSucceeded] Failed to complete event for unknown reason", {
runId: run.id,
spanId: run.spanId,
});
return;
}
} catch (error) {
logger.error("[runSucceeded] Failed to complete event", {
error: error instanceof Error ? error.message : error,
if (taskRunError) {
logger.error("[runSucceeded] Failed to find task run", {
error: taskRunError,
runId: run.id,
});
return;
}
const [completeSuccessfulRunEventError] = await tryCatch(
eventRepository.completeSuccessfulRunEvent({
run: taskRun,
endTime: time,
})
);
if (completeSuccessfulRunEventError) {
logger.error("[runSucceeded] Failed to complete successful run event", {
error: completeSuccessfulRunEventError,
runId: run.id,
spanId: run.spanId,
});
}
});
@@ -73,149 +80,181 @@ export function registerRunEngineEventBusHandlers() {
// Handle events
engine.eventBus.on("runFailed", async ({ time, run }) => {
try {
const sanitizedError = sanitizeError(run.error);
const exception = createExceptionPropertiesFromError(sanitizedError);
const sanitizedError = sanitizeError(run.error);
const exception = createExceptionPropertiesFromError(sanitizedError);
const eventStore = getTaskEventStoreTableForRun(run);
const completedEvent = await eventRepository.completeEvent(
eventStore,
run.spanId,
run.createdAt,
run.completedAt ?? undefined,
{
endTime: time,
attributes: {
isError: true,
},
events: [
{
name: "exception",
time,
properties: {
exception,
},
},
],
}
);
if (!completedEvent) {
logger.error("[runFailed] Failed to complete event for unknown reason", {
runId: run.id,
spanId: run.spanId,
});
return;
}
const inProgressEvents = await eventRepository.queryIncompleteEvents(
eventStore,
{
runId: completedEvent?.runId,
const [taskRunError, taskRun] = await tryCatch(
$replica.taskRun.findFirstOrThrow({
where: {
id: run.id,
},
run.createdAt,
run.completedAt ?? undefined
);
select: {
id: true,
friendlyId: true,
traceId: true,
spanId: true,
parentSpanId: true,
createdAt: true,
completedAt: true,
taskIdentifier: true,
projectId: true,
runtimeEnvironmentId: true,
environmentType: true,
isTest: true,
organizationId: true,
},
})
);
await Promise.all(
inProgressEvents.map((event) => {
try {
const completedEvent = eventRepository.completeEvent(
eventStore,
run.spanId,
run.createdAt,
run.completedAt ?? undefined,
{
endTime: time,
attributes: {
isError: true,
},
events: [
{
name: "exception",
time,
properties: {
exception,
},
},
],
}
);
if (!completedEvent) {
logger.error("[runFailed] Failed to complete in-progress event for unknown reason", {
runId: run.id,
spanId: run.spanId,
eventId: event.id,
});
return;
}
} catch (error) {
logger.error("[runFailed] Failed to complete in-progress event", {
error: error instanceof Error ? error.message : error,
runId: run.id,
spanId: run.spanId,
eventId: event.id,
});
}
})
);
} catch (error) {
logger.error("[runFailed] Failed to complete event", {
error: error instanceof Error ? error.message : error,
if (taskRunError) {
logger.error("[runFailed] Failed to find task run", {
error: taskRunError,
runId: run.id,
});
return;
}
const [completeFailedRunEventError] = await tryCatch(
eventRepository.completeFailedRunEvent({
run: taskRun,
endTime: time,
exception,
})
);
if (completeFailedRunEventError) {
logger.error("[runFailed] Failed to complete failed run event", {
error: completeFailedRunEventError,
runId: run.id,
spanId: run.spanId,
});
}
});
engine.eventBus.on("runAttemptFailed", async ({ time, run }) => {
try {
const sanitizedError = sanitizeError(run.error);
const exception = createExceptionPropertiesFromError(sanitizedError);
const eventStore = getTaskEventStoreTableForRun(run);
const sanitizedError = sanitizeError(run.error);
const exception = createExceptionPropertiesFromError(sanitizedError);
const inProgressEvents = await eventRepository.queryIncompleteEvents(
eventStore,
{
runId: RunId.toFriendlyId(run.id),
spanId: {
not: run.spanId,
},
const [taskRunError, taskRun] = await tryCatch(
$replica.taskRun.findFirstOrThrow({
where: {
id: run.id,
},
run.createdAt,
run.completedAt ?? undefined
);
select: {
id: true,
friendlyId: true,
traceId: true,
spanId: true,
parentSpanId: true,
createdAt: true,
completedAt: true,
taskIdentifier: true,
projectId: true,
runtimeEnvironmentId: true,
environmentType: true,
isTest: true,
organizationId: true,
},
})
);
await Promise.all(
inProgressEvents.map((event) => {
return eventRepository.crashEvent({
event: event,
crashedAt: time,
exception,
});
})
);
} catch (error) {
logger.error("[runAttemptFailed] Failed to complete event", {
error: error instanceof Error ? error.message : error,
if (taskRunError) {
logger.error("[runAttemptFailed] Failed to find task run", {
error: taskRunError,
runId: run.id,
});
return;
}
const [createAttemptFailedRunEventError] = await tryCatch(
eventRepository.createAttemptFailedRunEvent({
run: taskRun,
endTime: time,
attemptNumber: run.attemptNumber,
exception,
})
);
if (createAttemptFailedRunEventError) {
logger.error("[runAttemptFailed] Failed to create attempt failed run event", {
error: createAttemptFailedRunEventError,
runId: run.id,
spanId: run.spanId,
});
}
});
engine.eventBus.on("cachedRunCompleted", async ({ time, span, blockedRunId, hasError }) => {
try {
const blockedRun = await $replica.taskRun.findFirst({
select: {
taskEventStore: true,
},
where: {
id: blockedRunId,
},
});
engine.eventBus.on(
"cachedRunCompleted",
async ({ time, span, blockedRunId, hasError, cachedRunId }) => {
const [parentSpanId, spanId] = span.id.split(":");
if (!spanId || !parentSpanId) {
logger.debug("[cachedRunCompleted] Invalid span id", {
spanId,
parentSpanId,
});
return;
}
const [cachedRunError, cachedRun] = await tryCatch(
$replica.taskRun.findFirstOrThrow({
where: {
id: cachedRunId,
},
select: {
id: true,
friendlyId: true,
traceId: true,
spanId: true,
parentSpanId: true,
createdAt: true,
completedAt: true,
taskIdentifier: true,
projectId: true,
runtimeEnvironmentId: true,
environmentType: true,
isTest: true,
organizationId: true,
},
})
);
if (cachedRunError) {
logger.error("[cachedRunCompleted] Failed to find cached run", {
error: cachedRunError,
cachedRunId,
});
return;
}
const [blockedRunError, blockedRun] = await tryCatch(
$replica.taskRun.findFirst({
where: {
id: blockedRunId,
},
select: {
id: true,
friendlyId: true,
traceId: true,
spanId: true,
parentSpanId: true,
createdAt: true,
completedAt: true,
taskIdentifier: true,
projectId: true,
runtimeEnvironmentId: true,
environmentType: true,
isTest: true,
organizationId: true,
},
})
);
if (blockedRunError) {
logger.error("[cachedRunCompleted] Failed to find blocked run", {
error: blockedRunError,
blockedRunId,
});
}
if (!blockedRun) {
logger.error("[cachedRunCompleted] Blocked run not found", {
@@ -224,100 +263,125 @@ export function registerRunEngineEventBusHandlers() {
return;
}
const eventStore = getTaskEventStoreTableForRun(blockedRun);
const completedEvent = await eventRepository.completeEvent(
eventStore,
span.id,
span.createdAt,
time,
{
const [completeCachedRunEventError] = await tryCatch(
eventRepository.completeCachedRunEvent({
run: cachedRun,
blockedRun,
spanId,
parentSpanId,
spanCreatedAt: span.createdAt,
isError: hasError,
endTime: time,
attributes: {
isError: hasError,
},
}
})
);
if (!completedEvent) {
logger.error("[cachedRunCompleted] Failed to complete event for unknown reason", {
span,
if (completeCachedRunEventError) {
logger.error("[cachedRunCompleted] Failed to complete cached run event", {
error: completeCachedRunEventError,
cachedRunId,
});
return;
}
} catch (error) {
logger.error("[cachedRunCompleted] Failed to complete event for unknown reason", {
error: error instanceof Error ? error.message : error,
span,
});
}
});
);
engine.eventBus.on("runExpired", async ({ time, run }) => {
try {
const eventStore = getTaskEventStoreTableForRun(run);
if (!run.ttl) {
return;
}
const completedEvent = await eventRepository.completeEvent(
eventStore,
run.spanId,
run.createdAt,
run.completedAt ?? undefined,
{
endTime: time,
attributes: {
isError: true,
},
events: [
{
name: "exception",
time,
properties: {
exception: {
message: `Run expired because the TTL (${run.ttl}) was reached`,
},
},
},
],
}
);
const [taskRunError, taskRun] = await tryCatch(
$replica.taskRun.findFirstOrThrow({
where: {
id: run.id,
},
select: {
id: true,
friendlyId: true,
traceId: true,
spanId: true,
parentSpanId: true,
createdAt: true,
completedAt: true,
taskIdentifier: true,
projectId: true,
runtimeEnvironmentId: true,
environmentType: true,
isTest: true,
organizationId: true,
},
})
);
if (!completedEvent) {
logger.error("[runFailed] Failed to complete event for unknown reason", {
runId: run.id,
spanId: run.spanId,
});
return;
}
} catch (error) {
logger.error("[runExpired] Failed to complete event", {
error: error instanceof Error ? error.message : error,
if (taskRunError) {
logger.error("[runExpired] Failed to find task run", {
error: taskRunError,
runId: run.id,
});
return;
}
const [completeExpiredRunEventError] = await tryCatch(
eventRepository.completeExpiredRunEvent({
run: taskRun,
endTime: time,
ttl: run.ttl,
})
);
if (completeExpiredRunEventError) {
logger.error("[runExpired] Failed to complete expired run event", {
error: completeExpiredRunEventError,
runId: run.id,
spanId: run.spanId,
});
}
});
engine.eventBus.on("runCancelled", async ({ time, run }) => {
try {
const eventStore = getTaskEventStoreTableForRun(run);
const inProgressEvents = await eventRepository.queryIncompleteEvents(
eventStore,
{
runId: run.friendlyId,
const [taskRunError, taskRun] = await tryCatch(
$replica.taskRun.findFirstOrThrow({
where: {
id: run.id,
},
run.createdAt,
run.completedAt ?? undefined
);
select: {
id: true,
friendlyId: true,
traceId: true,
spanId: true,
parentSpanId: true,
createdAt: true,
completedAt: true,
taskIdentifier: true,
projectId: true,
runtimeEnvironmentId: true,
environmentType: true,
isTest: true,
organizationId: true,
},
})
);
const error = createJsonErrorObject(run.error);
await eventRepository.cancelEvents(inProgressEvents, time, error.message);
} catch (error) {
logger.error("[runCancelled] Failed to cancel event", {
error: error instanceof Error ? error.message : error,
if (taskRunError) {
logger.error("[runCancelled] Task run not found", {
error: taskRunError,
runId: run.id,
});
return;
}
const error = createJsonErrorObject(run.error);
const [cancelRunEventError] = await tryCatch(
eventRepository.cancelRunEvent({
reason: error.message,
run: taskRun,
cancelledAt: time,
})
);
if (cancelRunEventError) {
logger.error("[runCancelled] Failed to cancel run event", {
error: cancelRunEventError,
runId: run.id,
spanId: run.spanId,
});
}
});
apps/webapp/app/v3/services/cancelAttempt.server.ts
-19
View File
@@ -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);
})
);
});
}
}
apps/webapp/app/v3/services/cancelTaskRunV1.server.ts
+16 -16
View File
@@ -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,20 @@ 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,
})
);
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) {
apps/webapp/app/v3/services/completeAttempt.server.ts
+50 -85
View File
@@ -33,6 +33,7 @@ import { CancelAttemptService } from "./cancelAttempt.server";
import { CreateCheckpointService } from "./createCheckpoint.server";
import { FinalizeTaskRunService } from "./finalizeTaskRun.server";
import { RetryAttemptService } from "./retryAttempt.server";
import { tryCatch } from "@trigger.dev/core/utils";
type FoundAttempt = Awaited<ReturnType<typeof findAttempt>>;
@@ -164,27 +165,20 @@ export class CompleteAttemptService extends BaseService {
env,
});
// Now we need to "complete" the task run event/span
await eventRepository.completeEvent(
getTaskEventStoreTableForRun(taskRunAttempt.taskRun),
taskRunAttempt.taskRun.spanId,
taskRunAttempt.taskRun.createdAt,
taskRunAttempt.taskRun.completedAt ?? undefined,
{
const [completeSuccessfulRunEventError] = await tryCatch(
eventRepository.completeSuccessfulRunEvent({
run: taskRunAttempt.taskRun,
endTime: new Date(),
attributes: {
isError: false,
output:
completion.outputType === "application/store" || completion.outputType === "text/plain"
? completion.output
: completion.output
? (safeJsonParse(completion.output) as Attributes)
: undefined,
outputType: completion.outputType,
},
}
})
);
if (completeSuccessfulRunEventError) {
logger.error("[CompleteAttemptService] Failed to complete successful run event", {
error: completeSuccessfulRunEventError,
runId: taskRunAttempt.taskRunId,
});
}
return "COMPLETED";
}
@@ -322,29 +316,21 @@ export class CompleteAttemptService extends BaseService {
exitRun(taskRunAttempt.taskRunId);
}
// Now we need to "complete" the task run event/span
await eventRepository.completeEvent(
getTaskEventStoreTableForRun(taskRunAttempt.taskRun),
taskRunAttempt.taskRun.spanId,
taskRunAttempt.taskRun.createdAt,
taskRunAttempt.taskRun.completedAt ?? undefined,
{
const [completeFailedRunEventError] = await tryCatch(
eventRepository.completeFailedRunEvent({
run: taskRunAttempt.taskRun,
endTime: failedAt,
attributes: {
isError: true,
},
events: [
{
name: "exception",
time: failedAt,
properties: {
exception: createExceptionPropertiesFromError(sanitizedError),
},
},
],
}
exception: createExceptionPropertiesFromError(sanitizedError),
})
);
if (completeFailedRunEventError) {
logger.error("[CompleteAttemptService] Failed to complete failed run event", {
error: completeFailedRunEventError,
runId: taskRunAttempt.taskRunId,
});
}
await this._prisma.taskRun.update({
where: {
id: taskRunAttempt.taskRunId,
@@ -385,64 +371,43 @@ export class CompleteAttemptService extends BaseService {
return "COMPLETED";
}
const inProgressEvents = await eventRepository.queryIncompleteEvents(
getTaskEventStoreTableForRun(taskRunAttempt.taskRun),
{
runId: taskRunAttempt.taskRun.friendlyId,
},
taskRunAttempt.taskRun.createdAt,
taskRunAttempt.taskRun.completedAt ?? undefined
);
// Handle in-progress events
switch (status) {
case "CRASHED": {
logger.debug("[CompleteAttemptService] Crashing in-progress events", {
inProgressEvents: inProgressEvents.map((event) => event.id),
});
await Promise.all(
inProgressEvents.map((event) => {
return eventRepository.crashEvent({
event,
crashedAt: failedAt,
exception: createExceptionPropertiesFromError(sanitizedError),
});
const [createAttemptFailedEventError] = await tryCatch(
eventRepository.createAttemptFailedRunEvent({
run: taskRunAttempt.taskRun,
endTime: failedAt,
attemptNumber: taskRunAttempt.number,
exception: createExceptionPropertiesFromError(sanitizedError),
})
);
if (createAttemptFailedEventError) {
logger.error("[CompleteAttemptService] Failed to create attempt failed run event", {
error: createAttemptFailedEventError,
runId: taskRunAttempt.taskRunId,
});
}
break;
}
case "SYSTEM_FAILURE": {
logger.debug("[CompleteAttemptService] Failing in-progress events", {
inProgressEvents: inProgressEvents.map((event) => event.id),
});
await Promise.all(
inProgressEvents.map((event) => {
return eventRepository.completeEvent(
getTaskEventStoreTableForRun(taskRunAttempt.taskRun),
event.spanId,
taskRunAttempt.taskRun.createdAt,
taskRunAttempt.taskRun.completedAt ?? undefined,
{
endTime: failedAt,
attributes: {
isError: true,
},
events: [
{
name: "exception",
time: failedAt,
properties: {
exception: createExceptionPropertiesFromError(sanitizedError),
},
},
],
}
);
const [createAttemptFailedEventError] = await tryCatch(
eventRepository.createAttemptFailedRunEvent({
run: taskRunAttempt.taskRun,
endTime: failedAt,
attemptNumber: taskRunAttempt.number,
exception: createExceptionPropertiesFromError(sanitizedError),
})
);
if (createAttemptFailedEventError) {
logger.error("[CompleteAttemptService] Failed to create attempt failed run event", {
error: createAttemptFailedEventError,
runId: taskRunAttempt.taskRunId,
});
}
}
}
apps/webapp/app/v3/services/crashTaskRun.server.ts
+17 -25
View File
@@ -8,6 +8,7 @@ import { sanitizeError, TaskRunErrorCodes, TaskRunInternalError } from "@trigger
import { FinalizeTaskRunService } from "./finalizeTaskRun.server";
import { FailedTaskRunRetryHelper } from "../failedTaskRun.server";
import { getTaskEventStoreTableForRun } from "../taskEventStore.server";
import { tryCatch } from "@trigger.dev/core/utils";
export type CrashTaskRunServiceOptions = {
reason?: string;
@@ -120,34 +121,25 @@ export class CrashTaskRunService extends BaseService {
},
});
const inProgressEvents = await eventRepository.queryIncompleteEvents(
getTaskEventStoreTableForRun(taskRun),
{
runId: taskRun.friendlyId,
},
taskRun.createdAt,
taskRun.completedAt ?? undefined,
options?.overrideCompletion
);
logger.debug("[CrashTaskRunService] Crashing in-progress events", {
inProgressEvents: inProgressEvents.map((event) => event.id),
});
await Promise.all(
inProgressEvents.map((event) => {
return eventRepository.crashEvent({
event: event,
crashedAt: opts.crashedAt,
exception: {
type: opts.errorCode ?? TaskRunErrorCodes.TASK_RUN_CRASHED,
message: opts.reason,
stacktrace: opts.logs,
},
});
const [createAttemptFailedEventError] = await tryCatch(
eventRepository.completeFailedRunEvent({
run: crashedTaskRun,
endTime: opts.crashedAt,
exception: {
type: opts.errorCode ?? TaskRunErrorCodes.TASK_RUN_CRASHED,
message: opts.reason,
stacktrace: opts.logs,
},
})
);
if (createAttemptFailedEventError) {
logger.error("[CrashTaskRunService] Failed to complete failed run event", {
error: createAttemptFailedEventError,
runId: crashedTaskRun.id,
});
}
if (!opts.crashAttempts) {
return;
}
apps/webapp/app/v3/services/expireEnqueuedRun.server.ts
+16 -22
View File
@@ -5,6 +5,7 @@ import { eventRepository } from "../eventRepository.server";
import { BaseService } from "./baseService.server";
import { FinalizeTaskRunService } from "./finalizeTaskRun.server";
import { getTaskEventStoreTableForRun } from "../taskEventStore.server";
import { tryCatch } from "@trigger.dev/core/utils";
export class ExpireEnqueuedRunService extends BaseService {
public static async ack(runId: string, tx?: PrismaClientOrTransaction) {
@@ -78,28 +79,21 @@ export class ExpireEnqueuedRunService extends BaseService {
},
});
await eventRepository.completeEvent(
getTaskEventStoreTableForRun(run),
run.spanId,
run.createdAt,
run.completedAt ?? undefined,
{
endTime: new Date(),
attributes: {
isError: true,
},
events: [
{
name: "exception",
time: new Date(),
properties: {
exception: {
message: `Run expired because the TTL (${run.ttl}) was reached`,
},
},
},
],
if (run.ttl) {
const [completeExpiredRunEventError] = await tryCatch(
eventRepository.completeExpiredRunEvent({
run,
endTime: new Date(),
ttl: run.ttl,
})
);
if (completeExpiredRunEventError) {
logger.error("[ExpireEnqueuedRunService] Failed to complete expired run event", {
error: completeExpiredRunEventError,
runId: run.id,
});
}
);
}
}
}
apps/webapp/app/v3/taskEventStore.server.ts
+10 -4
View File
@@ -21,6 +21,7 @@ export type TraceEvent = Pick<
| "events"
| "environmentType"
| "kind"
| "attemptNumber"
>;
export type DetailedTraceEvent = Pick<
@@ -47,6 +48,7 @@ export type DetailedTraceEvent = Pick<
| "machinePreset"
| "properties"
| "output"
| "attemptNumber"
>;
export type TaskEventStoreTable = "taskEvent" | "taskEventPartitioned";
@@ -188,7 +190,8 @@ export class TaskEventStore {
level,
events,
"environmentType",
"kind"
"kind",
"attemptNumber"
FROM "TaskEventPartitioned"
WHERE
"traceId" = ${traceId}
@@ -220,7 +223,8 @@ export class TaskEventStore {
level,
events,
"environmentType",
"kind"
"kind",
"attemptNumber"
FROM "TaskEvent"
WHERE "traceId" = ${traceId}
${
@@ -273,7 +277,8 @@ export class TaskEventStore {
"queueName",
"machinePreset",
properties,
output
output,
"attemptNumber"
FROM "TaskEventPartitioned"
WHERE
"traceId" = ${traceId}
@@ -311,7 +316,8 @@ export class TaskEventStore {
"queueName",
"machinePreset",
properties,
output
output,
"attemptNumber"
FROM "TaskEvent"
WHERE "traceId" = ${traceId}
${
internal-packages/run-engine/src/engine/eventBus.ts
+1
View File
@@ -287,6 +287,7 @@ export type EventBusEvents = {
};
hasError: boolean;
blockedRunId: string;
cachedRunId?: string;
},
];
runMetadataUpdated: [
internal-packages/run-engine/src/engine/systems/waitpointSystem.ts
+1
View File
@@ -160,6 +160,7 @@ export class WaitpointSystem {
},
blockedRunId: run.taskRunId,
hasError: output?.isError ?? false,
cachedRunId: waitpoint.completedByTaskRunId ?? undefined,
});
}
}
packages/core/src/v3/schemas/openTelemetry.ts
+22 -1
View File
@@ -28,6 +28,18 @@ export const CancellationSpanEvent = z.object({
export type CancellationSpanEvent = z.infer<typeof CancellationSpanEvent>;
export const AttemptFailedSpanEvent = z.object({
name: z.literal("attempt_failed"),
time: z.coerce.date(),
properties: z.object({
exception: ExceptionEventProperties,
attemptNumber: z.number(),
runId: z.string(),
}),
});
export type AttemptFailedSpanEvent = z.infer<typeof AttemptFailedSpanEvent>;
export const OtherSpanEvent = z.object({
name: z.string(),
time: z.coerce.date(),
@@ -36,7 +48,12 @@ export const OtherSpanEvent = z.object({
export type OtherSpanEvent = z.infer<typeof OtherSpanEvent>;
export const SpanEvent = z.union([ExceptionSpanEvent, CancellationSpanEvent, OtherSpanEvent]);
export const SpanEvent = z.union([
ExceptionSpanEvent,
CancellationSpanEvent,
AttemptFailedSpanEvent,
OtherSpanEvent,
]);
export type SpanEvent = z.infer<typeof SpanEvent>;
@@ -52,6 +69,10 @@ export function isCancellationSpanEvent(event: SpanEvent): event is Cancellation
return event.name === "cancellation";
}
export function isAttemptFailedSpanEvent(event: SpanEvent): event is AttemptFailedSpanEvent {
return event.name === "attempt_failed";
}
export const SpanMessagingEvent = z.object({
system: z.string().optional(),
client_id: z.string().optional(),
packages/core/src/v3/workers/taskExecutor.ts
+1 -2
View File
@@ -352,8 +352,7 @@ export class TaskExecutor {
? runTimelineMetrics.convertMetricsToSpanEvents()
: undefined,
},
traceContext.extractContext(),
signal
traceContext.extractContext()
);
return { result };
references/hello-world/src/trigger/example.ts
+1
View File
@@ -1,6 +1,7 @@
import { batch, logger, task, tasks, timeout, wait } from "@trigger.dev/sdk";
import { setTimeout } from "timers/promises";
import { ResourceMonitor } from "../resourceMonitor.js";
import { fixedLengthTask } from "./batches.js";
export const helloWorldTask = task({
id: "hello-world",
references/hello-world/src/trigger/idempotency.ts
+19
View File
@@ -282,3 +282,22 @@ export const idempotencyTriggerByTaskAndWait = task({
logger.log("Results 2", { results2 });
},
});
export const idempotencyTriggerAndWaitWithInProgressRun = task({
id: "idempotency-trigger-and-wait-with-in-progress-run",
maxDuration: 60,
run: async () => {
await childTask.trigger(
{ message: "Hello, world!", duration: 5000, failureChance: 100 },
{
idempotencyKey: "b",
}
);
await childTask.triggerAndWait(
{ message: "Hello, world!", duration: 5000, failureChance: 0 },
{
idempotencyKey: "b",
}
);
},
});