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

Merge remote-tracking branch 'origin/main' into feat/ecr-support

Browse Source
This commit is contained in:
nicktrn
2025-07-02 16:31:24 +01:00
parent 512d6827f9 b7134dd769
commit 6202130aa5
58 changed files with 2098 additions and 590 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/polite-badgers-suffer.md
+5
View File
@@ -0,0 +1,5 @@
---
"trigger.dev": patch
---
experimental processKeepAlive
.changeset/wet-dragons-boil.md
+5
View File
@@ -0,0 +1,5 @@
---
"trigger.dev": patch
---
Serialize metadata to prevent invalid metadata from breaking run completions
apps/supervisor/src/util.ts
+10
View File
@@ -1,5 +1,15 @@
import { isMacOS, isWindows } from "std-env";
export function normalizeDockerHostUrl(url: string) {
const $url = new URL(url);
if ($url.hostname === "localhost") {
$url.hostname = getDockerHostDomain();
}
return $url.toString();
}
export function getDockerHostDomain() {
return isMacOS || isWindows ? "host.docker.internal" : "localhost";
}
apps/supervisor/src/workloadManager/docker.ts
+2 -2
View File
@@ -5,7 +5,7 @@ import {
type WorkloadManagerOptions,
} from "./types.js";
import { env } from "../env.js";
import { getDockerHostDomain, getRunnerId } from "../util.js";
import { getDockerHostDomain, getRunnerId, normalizeDockerHostUrl } from "../util.js";
import Docker from "dockerode";
import { tryCatch } from "@trigger.dev/core";
@@ -78,7 +78,7 @@ export class DockerWorkloadManager implements WorkloadManager {
];
if (this.opts.warmStartUrl) {
envVars.push(`TRIGGER_WARM_START_URL=${this.opts.warmStartUrl}`);
envVars.push(`TRIGGER_WARM_START_URL=${normalizeDockerHostUrl(this.opts.warmStartUrl)}`);
}
if (this.opts.metadataUrl) {
apps/webapp/app/env.server.ts
+10
View File
@@ -2,6 +2,7 @@ import { z } from "zod";
import { isValidDatabaseUrl } from "./utils/db";
import { isValidRegex } from "./utils/regex";
import { BoolEnv } from "./utils/boolEnv";
import { OTEL_ATTRIBUTE_PER_LINK_COUNT_LIMIT, OTEL_LINK_COUNT_LIMIT } from "@trigger.dev/core/v3";
const EnvironmentSchema = z.object({
NODE_ENV: z.union([z.literal("development"), z.literal("production"), z.literal("test")]),
@@ -282,6 +283,15 @@ const EnvironmentSchema = z.object({
PROD_OTEL_LOG_EXPORT_TIMEOUT_MILLIS: z.string().default("30000"),
PROD_OTEL_LOG_MAX_QUEUE_SIZE: z.string().default("512"),
TRIGGER_OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT: z.string().default("256"),
TRIGGER_OTEL_LOG_ATTRIBUTE_COUNT_LIMIT: z.string().default("256"),
TRIGGER_OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT: z.string().default("131072"),
TRIGGER_OTEL_LOG_ATTRIBUTE_VALUE_LENGTH_LIMIT: z.string().default("131072"),
TRIGGER_OTEL_SPAN_EVENT_COUNT_LIMIT: z.string().default("10"),
TRIGGER_OTEL_LINK_COUNT_LIMIT: z.string().default("2"),
TRIGGER_OTEL_ATTRIBUTE_PER_LINK_COUNT_LIMIT: z.string().default("10"),
TRIGGER_OTEL_ATTRIBUTE_PER_EVENT_COUNT_LIMIT: z.string().default("10"),
CHECKPOINT_THRESHOLD_IN_MS: z.coerce.number().int().default(30000),
// Internal OTEL environment variables
apps/webapp/app/utils/timelineSpanEvents.ts
-3
View File
@@ -142,9 +142,6 @@ function getFriendlyNameForEvent(event: string, properties?: Record<string, any>
return "Attempt created";
}
case "import": {
if (properties && typeof properties.file === "string") {
return `Importing ${properties.file}`;
}
return "Importing task file";
}
case "lazy_payload": {
apps/webapp/app/v3/environmentVariables/environmentVariablesRepository.server.ts
+90 -1
View File
@@ -810,6 +810,7 @@ export const RuntimeEnvironmentForEnvRepoPayload = {
apiKey: true,
organizationId: true,
branchName: true,
builtInEnvironmentVariableOverrides: true,
},
} as const;
@@ -1025,5 +1026,93 @@ async function resolveBuiltInProdVariables(
async function resolveCommonBuiltInVariables(
runtimeEnvironment: RuntimeEnvironmentForEnvRepo
): Promise<Array<EnvironmentVariable>> {
return [];
return [
{
key: "TRIGGER_OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT",
value: resolveBuiltInEnvironmentVariableOverrides(
"TRIGGER_OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT",
runtimeEnvironment,
String(env.TRIGGER_OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT)
),
},
{
key: "TRIGGER_OTEL_LOG_ATTRIBUTE_COUNT_LIMIT",
value: resolveBuiltInEnvironmentVariableOverrides(
"TRIGGER_OTEL_LOG_ATTRIBUTE_COUNT_LIMIT",
runtimeEnvironment,
String(env.TRIGGER_OTEL_LOG_ATTRIBUTE_COUNT_LIMIT)
),
},
{
key: "TRIGGER_OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT",
value: resolveBuiltInEnvironmentVariableOverrides(
"TRIGGER_OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT",
runtimeEnvironment,
String(env.TRIGGER_OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT)
),
},
{
key: "TRIGGER_OTEL_LOG_ATTRIBUTE_VALUE_LENGTH_LIMIT",
value: resolveBuiltInEnvironmentVariableOverrides(
"TRIGGER_OTEL_LOG_ATTRIBUTE_VALUE_LENGTH_LIMIT",
runtimeEnvironment,
String(env.TRIGGER_OTEL_LOG_ATTRIBUTE_VALUE_LENGTH_LIMIT)
),
},
{
key: "TRIGGER_OTEL_SPAN_EVENT_COUNT_LIMIT",
value: resolveBuiltInEnvironmentVariableOverrides(
"TRIGGER_OTEL_SPAN_EVENT_COUNT_LIMIT",
runtimeEnvironment,
String(env.TRIGGER_OTEL_SPAN_EVENT_COUNT_LIMIT)
),
},
{
key: "TRIGGER_OTEL_LINK_COUNT_LIMIT",
value: resolveBuiltInEnvironmentVariableOverrides(
"TRIGGER_OTEL_LINK_COUNT_LIMIT",
runtimeEnvironment,
String(env.TRIGGER_OTEL_LINK_COUNT_LIMIT)
),
},
{
key: "TRIGGER_OTEL_ATTRIBUTE_PER_LINK_COUNT_LIMIT",
value: resolveBuiltInEnvironmentVariableOverrides(
"TRIGGER_OTEL_ATTRIBUTE_PER_LINK_COUNT_LIMIT",
runtimeEnvironment,
String(env.TRIGGER_OTEL_ATTRIBUTE_PER_LINK_COUNT_LIMIT)
),
},
{
key: "TRIGGER_OTEL_ATTRIBUTE_PER_EVENT_COUNT_LIMIT",
value: resolveBuiltInEnvironmentVariableOverrides(
"TRIGGER_OTEL_ATTRIBUTE_PER_EVENT_COUNT_LIMIT",
runtimeEnvironment,
String(env.TRIGGER_OTEL_ATTRIBUTE_PER_EVENT_COUNT_LIMIT)
),
},
];
}
function resolveBuiltInEnvironmentVariableOverrides(
key: string,
runtimeEnvironment: RuntimeEnvironmentForEnvRepo,
defaultValue: string
) {
const overrides = runtimeEnvironment.builtInEnvironmentVariableOverrides;
if (!overrides) {
return defaultValue;
}
if (
!Array.isArray(overrides) &&
typeof overrides === "object" &&
key in overrides &&
typeof overrides[key] === "string"
) {
return overrides[key];
}
return defaultValue;
}
apps/webapp/app/v3/otlpExporter.server.ts
+213 -88
View File
@@ -39,7 +39,8 @@ class OTLPExporter {
constructor(
private readonly _eventRepository: EventRepository,
private readonly _verbose: boolean
private readonly _verbose: boolean,
private readonly _spanAttributeValueLengthLimit: number
) {
this._tracer = trace.getTracer("otlp-exporter");
}
@@ -52,7 +53,7 @@ class OTLPExporter {
this.#logExportTracesVerbose(request);
const events = this.#filterResourceSpans(request.resourceSpans).flatMap((resourceSpan) => {
return convertSpansToCreateableEvents(resourceSpan);
return convertSpansToCreateableEvents(resourceSpan, this._spanAttributeValueLengthLimit);
});
const enrichedEvents = enrichCreatableEvents(events);
@@ -79,7 +80,7 @@ class OTLPExporter {
this.#logExportLogsVerbose(request);
const events = this.#filterResourceLogs(request.resourceLogs).flatMap((resourceLog) => {
return convertLogsToCreateableEvents(resourceLog);
return convertLogsToCreateableEvents(resourceLog, this._spanAttributeValueLengthLimit);
});
const enrichedEvents = enrichCreatableEvents(events);
@@ -180,10 +181,13 @@ class OTLPExporter {
}
}
function convertLogsToCreateableEvents(resourceLog: ResourceLogs): Array<CreatableEvent> {
function convertLogsToCreateableEvents(
resourceLog: ResourceLogs,
spanAttributeValueLengthLimit: number
): Array<CreatableEvent> {
const resourceAttributes = resourceLog.resource?.attributes ?? [];
const resourceProperties = extractResourceProperties(resourceAttributes);
const resourceProperties = extractEventProperties(resourceAttributes);
return resourceLog.scopeLogs.flatMap((scopeLog) => {
return scopeLog.logRecords
@@ -194,6 +198,11 @@ function convertLogsToCreateableEvents(resourceLog: ResourceLogs): Array<Creatab
return;
}
const logProperties = extractEventProperties(
log.attributes ?? [],
SemanticInternalAttributes.METADATA
);
return {
traceId: binaryToHex(log.traceId),
spanId: eventRepository.generateSpanId(),
@@ -208,14 +217,9 @@ function convertLogsToCreateableEvents(resourceLog: ResourceLogs): Array<Creatab
status: logLevelToEventStatus(log.severityNumber),
startTime: log.timeUnixNano,
properties: {
...convertKeyValueItemsToMap(log.attributes ?? [], [
SemanticInternalAttributes.SPAN_ID,
SemanticInternalAttributes.SPAN_PARTIAL,
]),
...convertKeyValueItemsToMap(
resourceAttributes,
[SemanticInternalAttributes.TRIGGER],
SemanticInternalAttributes.METADATA
truncateAttributes(log.attributes ?? [], spanAttributeValueLengthLimit),
[SemanticInternalAttributes.SPAN_ID, SemanticInternalAttributes.SPAN_PARTIAL]
),
},
style: convertKeyValueItemsToMap(
@@ -236,7 +240,35 @@ function convertLogsToCreateableEvents(resourceLog: ResourceLogs): Array<Creatab
),
SemanticInternalAttributes.PAYLOAD
),
...resourceProperties,
metadata: logProperties.metadata ?? resourceProperties.metadata,
serviceName: logProperties.serviceName ?? resourceProperties.serviceName ?? "unknown",
serviceNamespace:
logProperties.serviceNamespace ?? resourceProperties.serviceNamespace ?? "unknown",
environmentId:
logProperties.environmentId ?? resourceProperties.environmentId ?? "unknown",
environmentType:
logProperties.environmentType ?? resourceProperties.environmentType ?? "DEVELOPMENT",
organizationId:
logProperties.organizationId ?? resourceProperties.organizationId ?? "unknown",
projectId: logProperties.projectId ?? resourceProperties.projectId ?? "unknown",
projectRef: logProperties.projectRef ?? resourceProperties.projectRef ?? "unknown",
runId: logProperties.runId ?? resourceProperties.runId ?? "unknown",
runIsTest: logProperties.runIsTest ?? resourceProperties.runIsTest ?? false,
taskSlug: logProperties.taskSlug ?? resourceProperties.taskSlug ?? "unknown",
taskPath: logProperties.taskPath ?? resourceProperties.taskPath ?? "unknown",
workerId: logProperties.workerId ?? resourceProperties.workerId ?? "unknown",
workerVersion:
logProperties.workerVersion ?? resourceProperties.workerVersion ?? "unknown",
queueId: logProperties.queueId ?? resourceProperties.queueId ?? "unknown",
queueName: logProperties.queueName ?? resourceProperties.queueName ?? "unknown",
batchId: logProperties.batchId ?? resourceProperties.batchId,
idempotencyKey: logProperties.idempotencyKey ?? resourceProperties.idempotencyKey,
machinePreset: logProperties.machinePreset ?? resourceProperties.machinePreset,
machinePresetCpu: logProperties.machinePresetCpu ?? resourceProperties.machinePresetCpu,
machinePresetMemory:
logProperties.machinePresetMemory ?? resourceProperties.machinePresetMemory,
machinePresetCentsPerMs:
logProperties.machinePresetCentsPerMs ?? resourceProperties.machinePresetCentsPerMs,
attemptId:
extractStringAttribute(
log.attributes ?? [],
@@ -255,10 +287,13 @@ function convertLogsToCreateableEvents(resourceLog: ResourceLogs): Array<Creatab
});
}
function convertSpansToCreateableEvents(resourceSpan: ResourceSpans): Array<CreatableEvent> {
function convertSpansToCreateableEvents(
resourceSpan: ResourceSpans,
spanAttributeValueLengthLimit: number
): Array<CreatableEvent> {
const resourceAttributes = resourceSpan.resource?.attributes ?? [];
const resourceProperties = extractResourceProperties(resourceAttributes);
const resourceProperties = extractEventProperties(resourceAttributes);
return resourceSpan.scopeSpans.flatMap((scopeSpan) => {
return scopeSpan.spans
@@ -269,6 +304,11 @@ function convertSpansToCreateableEvents(resourceSpan: ResourceSpans): Array<Crea
return;
}
const spanProperties = extractEventProperties(
span.attributes ?? [],
SemanticInternalAttributes.METADATA
);
return {
traceId: binaryToHex(span.traceId),
spanId: isPartial
@@ -290,14 +330,9 @@ function convertSpansToCreateableEvents(resourceSpan: ResourceSpans): Array<Crea
events: spanEventsToEventEvents(span.events ?? []),
duration: span.endTimeUnixNano - span.startTimeUnixNano,
properties: {
...convertKeyValueItemsToMap(span.attributes ?? [], [
SemanticInternalAttributes.SPAN_ID,
SemanticInternalAttributes.SPAN_PARTIAL,
]),
...convertKeyValueItemsToMap(
resourceAttributes,
[SemanticInternalAttributes.TRIGGER],
SemanticInternalAttributes.METADATA
truncateAttributes(span.attributes ?? [], spanAttributeValueLengthLimit),
[SemanticInternalAttributes.SPAN_ID, SemanticInternalAttributes.SPAN_PARTIAL]
),
},
style: convertKeyValueItemsToMap(
@@ -327,7 +362,35 @@ function convertSpansToCreateableEvents(resourceSpan: ResourceSpans): Array<Crea
span.attributes ?? [],
SemanticInternalAttributes.PAYLOAD_TYPE
) ?? "application/json",
...resourceProperties,
metadata: spanProperties.metadata ?? resourceProperties.metadata,
serviceName: spanProperties.serviceName ?? resourceProperties.serviceName ?? "unknown",
serviceNamespace:
spanProperties.serviceNamespace ?? resourceProperties.serviceNamespace ?? "unknown",
environmentId:
spanProperties.environmentId ?? resourceProperties.environmentId ?? "unknown",
environmentType:
spanProperties.environmentType ?? resourceProperties.environmentType ?? "DEVELOPMENT",
organizationId:
spanProperties.organizationId ?? resourceProperties.organizationId ?? "unknown",
projectId: spanProperties.projectId ?? resourceProperties.projectId ?? "unknown",
projectRef: spanProperties.projectRef ?? resourceProperties.projectRef ?? "unknown",
runId: spanProperties.runId ?? resourceProperties.runId ?? "unknown",
runIsTest: spanProperties.runIsTest ?? resourceProperties.runIsTest ?? false,
taskSlug: spanProperties.taskSlug ?? resourceProperties.taskSlug ?? "unknown",
taskPath: spanProperties.taskPath ?? resourceProperties.taskPath ?? "unknown",
workerId: spanProperties.workerId ?? resourceProperties.workerId ?? "unknown",
workerVersion:
spanProperties.workerVersion ?? resourceProperties.workerVersion ?? "unknown",
queueId: spanProperties.queueId ?? resourceProperties.queueId ?? "unknown",
queueName: spanProperties.queueName ?? resourceProperties.queueName ?? "unknown",
batchId: spanProperties.batchId ?? resourceProperties.batchId,
idempotencyKey: spanProperties.idempotencyKey ?? resourceProperties.idempotencyKey,
machinePreset: spanProperties.machinePreset ?? resourceProperties.machinePreset,
machinePresetCpu: spanProperties.machinePresetCpu ?? resourceProperties.machinePresetCpu,
machinePresetMemory:
spanProperties.machinePresetMemory ?? resourceProperties.machinePresetMemory,
machinePresetCentsPerMs:
spanProperties.machinePresetCentsPerMs ?? resourceProperties.machinePresetCentsPerMs,
attemptId:
extractStringAttribute(
span.attributes ?? [],
@@ -359,67 +422,77 @@ function convertSpansToCreateableEvents(resourceSpan: ResourceSpans): Array<Crea
});
}
function extractResourceProperties(attributes: KeyValue[]) {
function extractEventProperties(attributes: KeyValue[], prefix?: string) {
return {
metadata: convertKeyValueItemsToMap(attributes, [SemanticInternalAttributes.TRIGGER]),
serviceName: extractStringAttribute(
attributes,
SemanticResourceAttributes.SERVICE_NAME,
"unknown"
),
serviceName: extractStringAttribute(attributes, SemanticResourceAttributes.SERVICE_NAME),
serviceNamespace: extractStringAttribute(
attributes,
SemanticResourceAttributes.SERVICE_NAMESPACE,
"unknown"
SemanticResourceAttributes.SERVICE_NAMESPACE
),
environmentId: extractStringAttribute(
attributes,
environmentId: extractStringAttribute(attributes, [
prefix,
SemanticInternalAttributes.ENVIRONMENT_ID,
"unknown"
),
environmentType: extractStringAttribute(
attributes,
]),
environmentType: extractStringAttribute(attributes, [
prefix,
SemanticInternalAttributes.ENVIRONMENT_TYPE,
"unknown"
) as CreatableEventEnvironmentType,
organizationId: extractStringAttribute(
attributes,
]) as CreatableEventEnvironmentType,
organizationId: extractStringAttribute(attributes, [
prefix,
SemanticInternalAttributes.ORGANIZATION_ID,
"unknown"
),
projectId: extractStringAttribute(attributes, SemanticInternalAttributes.PROJECT_ID, "unknown"),
projectRef: extractStringAttribute(
attributes,
]),
projectId: extractStringAttribute(attributes, [prefix, SemanticInternalAttributes.PROJECT_ID]),
projectRef: extractStringAttribute(attributes, [
prefix,
SemanticInternalAttributes.PROJECT_REF,
"unknown"
]),
runId: extractStringAttribute(attributes, [prefix, SemanticInternalAttributes.RUN_ID]),
runIsTest: extractBooleanAttribute(
attributes,
[prefix, SemanticInternalAttributes.RUN_IS_TEST],
false
),
runId: extractStringAttribute(attributes, SemanticInternalAttributes.RUN_ID, "unknown"),
runIsTest: extractBooleanAttribute(attributes, SemanticInternalAttributes.RUN_IS_TEST, false),
attemptId: extractStringAttribute(attributes, SemanticInternalAttributes.ATTEMPT_ID),
attemptNumber: extractNumberAttribute(attributes, SemanticInternalAttributes.ATTEMPT_NUMBER),
taskSlug: extractStringAttribute(attributes, SemanticInternalAttributes.TASK_SLUG, "unknown"),
taskPath: extractStringAttribute(attributes, SemanticInternalAttributes.TASK_PATH),
attemptId: extractStringAttribute(attributes, [prefix, SemanticInternalAttributes.ATTEMPT_ID]),
attemptNumber: extractNumberAttribute(attributes, [
prefix,
SemanticInternalAttributes.ATTEMPT_NUMBER,
]),
taskSlug: extractStringAttribute(attributes, [prefix, SemanticInternalAttributes.TASK_SLUG]),
taskPath: extractStringAttribute(attributes, [prefix, SemanticInternalAttributes.TASK_PATH]),
taskExportName: "@deprecated",
workerId: extractStringAttribute(attributes, SemanticInternalAttributes.WORKER_ID),
workerVersion: extractStringAttribute(attributes, SemanticInternalAttributes.WORKER_VERSION),
queueId: extractStringAttribute(attributes, SemanticInternalAttributes.QUEUE_ID),
queueName: extractStringAttribute(attributes, SemanticInternalAttributes.QUEUE_NAME),
batchId: extractStringAttribute(attributes, SemanticInternalAttributes.BATCH_ID),
idempotencyKey: extractStringAttribute(attributes, SemanticInternalAttributes.IDEMPOTENCY_KEY),
machinePreset: extractStringAttribute(
attributes,
SemanticInternalAttributes.MACHINE_PRESET_NAME
),
workerId: extractStringAttribute(attributes, [prefix, SemanticInternalAttributes.WORKER_ID]),
workerVersion: extractStringAttribute(attributes, [
prefix,
SemanticInternalAttributes.WORKER_VERSION,
]),
queueId: extractStringAttribute(attributes, [prefix, SemanticInternalAttributes.QUEUE_ID]),
queueName: extractStringAttribute(attributes, [prefix, SemanticInternalAttributes.QUEUE_NAME]),
batchId: extractStringAttribute(attributes, [prefix, SemanticInternalAttributes.BATCH_ID]),
idempotencyKey: extractStringAttribute(attributes, [
prefix,
SemanticInternalAttributes.IDEMPOTENCY_KEY,
]),
machinePreset: extractStringAttribute(attributes, [
prefix,
SemanticInternalAttributes.MACHINE_PRESET_NAME,
]),
machinePresetCpu:
extractDoubleAttribute(attributes, SemanticInternalAttributes.MACHINE_PRESET_CPU) ??
extractNumberAttribute(attributes, SemanticInternalAttributes.MACHINE_PRESET_CPU),
extractDoubleAttribute(attributes, [prefix, SemanticInternalAttributes.MACHINE_PRESET_CPU]) ??
extractNumberAttribute(attributes, [prefix, SemanticInternalAttributes.MACHINE_PRESET_CPU]),
machinePresetMemory:
extractDoubleAttribute(attributes, SemanticInternalAttributes.MACHINE_PRESET_MEMORY) ??
extractNumberAttribute(attributes, SemanticInternalAttributes.MACHINE_PRESET_MEMORY),
machinePresetCentsPerMs: extractDoubleAttribute(
attributes,
SemanticInternalAttributes.MACHINE_PRESET_CENTS_PER_MS
),
extractDoubleAttribute(attributes, [
prefix,
SemanticInternalAttributes.MACHINE_PRESET_MEMORY,
]) ??
extractNumberAttribute(attributes, [
prefix,
SemanticInternalAttributes.MACHINE_PRESET_MEMORY,
]),
machinePresetCentsPerMs: extractDoubleAttribute(attributes, [
prefix,
SemanticInternalAttributes.MACHINE_PRESET_CENTS_PER_MS,
]),
};
}
@@ -643,56 +716,92 @@ function convertUnixNanoToDate(unixNano: bigint | number): Date {
return new Date(Number(BigInt(unixNano) / BigInt(1_000_000)));
}
function extractStringAttribute(attributes: KeyValue[], name: string): string | undefined;
function extractStringAttribute(attributes: KeyValue[], name: string, fallback: string): string;
function extractStringAttribute(
attributes: KeyValue[],
name: string,
name: string | Array<string | undefined>
): string | undefined;
function extractStringAttribute(
attributes: KeyValue[],
name: string | Array<string | undefined>,
fallback: string
): string;
function extractStringAttribute(
attributes: KeyValue[],
name: string | Array<string | undefined>,
fallback?: string
): string | undefined {
const attribute = attributes.find((attribute) => attribute.key === name);
const key = Array.isArray(name) ? name.filter(Boolean).join(".") : name;
const attribute = attributes.find((attribute) => attribute.key === key);
if (!attribute) return fallback;
return isStringValue(attribute?.value) ? attribute.value.stringValue : fallback;
}
function extractNumberAttribute(attributes: KeyValue[], name: string): number | undefined;
function extractNumberAttribute(attributes: KeyValue[], name: string, fallback: number): number;
function extractNumberAttribute(
attributes: KeyValue[],
name: string,
name: string | Array<string | undefined>
): number | undefined;
function extractNumberAttribute(
attributes: KeyValue[],
name: string | Array<string | undefined>,
fallback: number
): number;
function extractNumberAttribute(
attributes: KeyValue[],
name: string | Array<string | undefined>,
fallback?: number
): number | undefined {
const attribute = attributes.find((attribute) => attribute.key === name);
const key = Array.isArray(name) ? name.filter(Boolean).join(".") : name;
const attribute = attributes.find((attribute) => attribute.key === key);
if (!attribute) return fallback;
return isIntValue(attribute?.value) ? Number(attribute.value.intValue) : fallback;
}
function extractDoubleAttribute(attributes: KeyValue[], name: string): number | undefined;
function extractDoubleAttribute(attributes: KeyValue[], name: string, fallback: number): number;
function extractDoubleAttribute(
attributes: KeyValue[],
name: string,
name: string | Array<string | undefined>
): number | undefined;
function extractDoubleAttribute(
attributes: KeyValue[],
name: string | Array<string | undefined>,
fallback: number
): number;
function extractDoubleAttribute(
attributes: KeyValue[],
name: string | Array<string | undefined>,
fallback?: number
): number | undefined {
const attribute = attributes.find((attribute) => attribute.key === name);
const key = Array.isArray(name) ? name.filter(Boolean).join(".") : name;
const attribute = attributes.find((attribute) => attribute.key === key);
if (!attribute) return fallback;
return isDoubleValue(attribute?.value) ? Number(attribute.value.doubleValue) : fallback;
}
function extractBooleanAttribute(attributes: KeyValue[], name: string): boolean | undefined;
function extractBooleanAttribute(attributes: KeyValue[], name: string, fallback: boolean): boolean;
function extractBooleanAttribute(
attributes: KeyValue[],
name: string,
name: string | Array<string | undefined>
): boolean | undefined;
function extractBooleanAttribute(
attributes: KeyValue[],
name: string | Array<string | undefined>,
fallback: boolean
): boolean;
function extractBooleanAttribute(
attributes: KeyValue[],
name: string | Array<string | undefined>,
fallback?: boolean
): boolean | undefined {
const attribute = attributes.find((attribute) => attribute.key === name);
const key = Array.isArray(name) ? name.filter(Boolean).join(".") : name;
const attribute = attributes.find((attribute) => attribute.key === key);
if (!attribute) return fallback;
@@ -750,7 +859,23 @@ function binaryToHex(buffer: Buffer | string | undefined): string | undefined {
return Buffer.from(Array.from(buffer)).toString("hex");
}
function truncateAttributes(attributes: KeyValue[], maximumLength: number = 1024): KeyValue[] {
return attributes.map((attribute) => {
return isStringValue(attribute.value)
? {
key: attribute.key,
value: {
stringValue: attribute.value.stringValue.slice(0, maximumLength),
},
}
: attribute;
});
}
export const otlpExporter = new OTLPExporter(
eventRepository,
process.env.OTLP_EXPORTER_VERBOSE === "1"
process.env.OTLP_EXPORTER_VERBOSE === "1",
process.env.SERVER_OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT
? parseInt(process.env.SERVER_OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT, 10)
: 8192
);
apps/webapp/test/timelineSpanEvents.test.ts
+5 -5
View File
@@ -76,7 +76,7 @@ describe("createTimelineSpanEventsFromSpanEvents", () => {
expect(result.some((event) => event.name === "Dequeued")).toBe(true);
expect(result.some((event) => event.name === "Launched")).toBe(true);
expect(result.some((event) => event.name === "Attempt created")).toBe(true);
expect(result.some((event) => event.name === "Importing src/trigger/chat.ts")).toBe(true);
expect(result.some((event) => event.name === "Importing task file")).toBe(true);
});
test("should sort events by timestamp", () => {
@@ -86,7 +86,7 @@ describe("createTimelineSpanEventsFromSpanEvents", () => {
expect(result[0].name).toBe("Dequeued");
expect(result[1].name).toBe("Attempt created");
expect(result[2].name).toBe("Launched");
expect(result[3].name).toBe("Importing src/trigger/chat.ts");
expect(result[3].name).toBe("Importing task file");
});
test("should calculate offsets correctly from the first event", () => {
@@ -176,7 +176,7 @@ describe("createTimelineSpanEventsFromSpanEvents", () => {
expect(result.find((e) => e.name === "Attempt created")?.helpText).toBe(
"An attempt was created for the run"
);
expect(result.find((e) => e.name === "Importing src/trigger/chat.ts")?.helpText).toBe(
expect(result.find((e) => e.name === "Importing task file")?.helpText).toBe(
"A task file was imported"
);
});
@@ -187,7 +187,7 @@ describe("createTimelineSpanEventsFromSpanEvents", () => {
expect(result.find((e) => e.name === "Dequeued")?.duration).toBe(0);
expect(result.find((e) => e.name === "Launched")?.duration).toBe(127);
expect(result.find((e) => e.name === "Attempt created")?.duration).toBe(56);
expect(result.find((e) => e.name === "Importing src/trigger/chat.ts")?.duration).toBe(67);
expect(result.find((e) => e.name === "Importing task file")?.duration).toBe(67);
});
test("should use fallback name for import event without file property", () => {
@@ -214,7 +214,7 @@ describe("createTimelineSpanEventsFromSpanEvents", () => {
// Without fork event, import should also be visible for non-admins
expect(result.length).toBe(2);
expect(result.some((event) => event.name === "Dequeued")).toBe(true);
expect(result.some((event) => event.name === "Importing src/trigger/chat.ts")).toBe(true);
expect(result.some((event) => event.name === "Importing task file")).toBe(true);
// create_attempt should still be admin-only
expect(result.some((event) => event.name === "Attempt created")).toBe(false);
docs/config/config-file.mdx
+5
View File
@@ -6,6 +6,7 @@ description: "This file is used to configure your project and how it's built."
import ScrapingWarning from "/snippets/web-scraping-warning.mdx";
import BundlePackages from "/snippets/bundle-packages.mdx";
import NodeVersions from "/snippets/node-versions.mdx";
The `trigger.config.ts` file is used to configure your Trigger.dev project. It is a TypeScript file at the root of your project that exports a default configuration object. Here's an example:
@@ -245,6 +246,10 @@ export default defineConfig({
See our [Bun guide](/guides/frameworks/bun) for more information.
### Node.js versions
<NodeVersions />
## Default machine
You can specify the default machine for all tasks in your project:
docs/snippets/node-versions.mdx
+4
View File
@@ -0,0 +1,4 @@
Trigger.dev runs your tasks on specific Node.js versions:
- **v3**: Uses Node.js `21.7.3`
- **v4**: Uses Node.js `21.7.3`
docs/upgrade-to-v4.mdx
+6
View File
@@ -3,10 +3,16 @@ title: "Upgrading to v4"
description: "What's new in v4, how to upgrade, and breaking changes."
---
import NodeVersions from "/snippets/node-versions.mdx";
## What's new in v4?
[Read our blog post](https://trigger.dev/blog/v4-beta-launch) for an overview of the new features.
### Node.js support
<NodeVersions />
### Wait tokens
In addition to waiting for a specific duration, or waiting for a child task to complete, you can now create and wait for a token to be completed, giving you more flexibility and the ability to wait for arbitrary conditions. For example, you can send the token to a Slack channel, and only complete the token when the user has clicked an "Approve" button.
hosting/k8s/helm/Chart.yaml
+1 -1
View File
@@ -2,7 +2,7 @@ apiVersion: v2
name: trigger
description: The official Trigger.dev Helm chart
type: application
version: 4.0.0-beta.11
version: 4.0.0-beta.14
appVersion: trigger-helm-rc.1
home: https://trigger.dev
sources:
hosting/k8s/helm/templates/webapp.yaml
+9 -2
View File
@@ -48,10 +48,11 @@ spec:
{{- include "trigger-v4.componentSelectorLabels" (dict "Chart" .Chart "Release" .Release "Values" .Values "component" $component) | nindent 6 }}
template:
metadata:
{{- with .Values.webapp.podAnnotations }}
annotations:
kubectl.kubernetes.io/default-container: webapp
{{- with .Values.webapp.podAnnotations }}
{{- toYaml . | nindent 8 }}
{{- end }}
{{- end }}
labels:
{{- include "trigger-v4.componentSelectorLabels" (dict "Chart" .Chart "Release" .Release "Values" .Values "component" $component) | nindent 8 }}
spec:
@@ -323,6 +324,9 @@ spec:
volumeMounts:
- name: shared
mountPath: /home/node/shared
{{- with .Values.webapp.extraVolumeMounts }}
{{- toYaml . | nindent 12 }}
{{- end }}
volumes:
- name: shared
{{- if .Values.persistence.shared.enabled }}
@@ -331,6 +335,9 @@ spec:
{{- else }}
emptyDir: {}
{{- end }}
{{- with .Values.webapp.extraVolumes }}
{{- toYaml . | nindent 8 }}
{{- end }}
{{- with .Values.webapp.nodeSelector }}
nodeSelector:
{{- toYaml . | nindent 8 }}
hosting/k8s/helm/values.yaml
+21 -1
View File
@@ -115,6 +115,26 @@ webapp:
# name: my-secret
# key: secret-key
# Extra volumes for the webapp pod
extraVolumes:
[]
# - name: config-volume
# configMap:
# name: my-config
# - name: secret-volume
# secret:
# secretName: my-secret
# Extra volume mounts for the webapp container
extraVolumeMounts:
[]
# - name: config-volume
# mountPath: /etc/config
# readOnly: true
# - name: secret-volume
# mountPath: /etc/secrets
# readOnly: true
# ServiceMonitor for Prometheus monitoring
serviceMonitor:
enabled: false
@@ -335,6 +355,7 @@ postgres:
persistence:
enabled: true
size: 10Gi
resourcesPreset: "small"
resources: {}
configuration: |
listen_addresses = '*'
@@ -657,7 +678,6 @@ persistence:
storageClass: ""
retain: true # Prevents deletion on uninstall
# Telemetry configuration
telemetry:
enabled: true
internal-packages/database/prisma/migrations/20250620144525_add_built_in_overrides_to_runtime_environment/migration.sql
+2
View File
@@ -0,0 +1,2 @@
-- AlterTable
ALTER TABLE "RuntimeEnvironment" ADD COLUMN "builtInEnvironmentVariableOverrides" JSONB;
internal-packages/database/prisma/schema.prisma
+3
View File
@@ -241,6 +241,9 @@ model RuntimeEnvironment {
createdAt DateTime @default(now())
updatedAt DateTime @updatedAt
/// Allows us to customize the built-in environment variables for a specific environment, like TRIGGER_OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT
builtInEnvironmentVariableOverrides Json?
tunnelId String?
backgroundWorkers BackgroundWorker[]
internal-packages/run-engine/src/engine/systems/runAttemptSystem.ts
+55 -9
View File
@@ -2,6 +2,7 @@ import { startSpan } from "@internal/tracing";
import {
CompleteRunAttemptResult,
ExecutionResult,
FlushedRunMetadata,
GitMeta,
StartRunAttemptResult,
TaskRunError,
@@ -35,6 +36,7 @@ import {
import { ReleaseConcurrencySystem } from "./releaseConcurrencySystem.js";
import { SystemResources } from "./systems.js";
import { WaitpointSystem } from "./waitpointSystem.js";
import { tryCatch } from "@trigger.dev/core/utils";
export type RunAttemptSystemOptions = {
resources: SystemResources;
@@ -386,15 +388,7 @@ export class RunAttemptSystem {
workerId?: string;
runnerId?: string;
}): Promise<CompleteRunAttemptResult> {
if (completion.metadata) {
this.$.eventBus.emit("runMetadataUpdated", {
time: new Date(),
run: {
id: runId,
metadata: completion.metadata,
},
});
}
await this.#notifyMetadataUpdated(runId, completion);
switch (completion.ok) {
case true: {
@@ -1314,4 +1308,56 @@ export class RunAttemptSystem {
return taskRun?.runtimeEnvironment;
}
async #notifyMetadataUpdated(runId: string, completion: TaskRunExecutionResult) {
if (completion.metadata) {
this.$.eventBus.emit("runMetadataUpdated", {
time: new Date(),
run: {
id: runId,
metadata: completion.metadata,
},
});
return;
}
if (completion.flushedMetadata) {
const [packetError, packet] = await tryCatch(parsePacket(completion.flushedMetadata));
if (!packet) {
return;
}
if (packetError) {
this.$.logger.error("RunEngine.completeRunAttempt(): failed to parse flushed metadata", {
runId,
flushedMetadata: completion.flushedMetadata,
error: packetError,
});
return;
}
const metadata = FlushedRunMetadata.safeParse(packet);
if (!metadata.success) {
this.$.logger.error("RunEngine.completeRunAttempt(): failed to parse flushed metadata", {
runId,
flushedMetadata: completion.flushedMetadata,
error: metadata.error,
});
return;
}
this.$.eventBus.emit("runMetadataUpdated", {
time: new Date(),
run: {
id: runId,
metadata: metadata.data,
},
});
}
}
}
internal-packages/run-engine/vitest.config.ts
+1 -1
View File
@@ -11,7 +11,7 @@ export default defineConfig({
singleThread: true,
},
},
testTimeout: 60_000,
testTimeout: 120_000,
coverage: {
provider: "v8",
},
packages/cli-v3/src/dev/devSupervisor.ts
+63
View File
@@ -25,6 +25,8 @@ import {
import pLimit from "p-limit";
import { resolveLocalEnvVars } from "../utilities/localEnvVars.js";
import type { Metafile } from "esbuild";
import { TaskRunProcessPool } from "./taskRunProcessPool.js";
import { tryCatch } from "@trigger.dev/core/utils";
export type WorkerRuntimeOptions = {
name: string | undefined;
@@ -67,6 +69,7 @@ class DevSupervisor implements WorkerRuntime {
private socketConnections = new Set<string>();
private runLimiter?: ReturnType<typeof pLimit>;
private taskRunProcessPool?: TaskRunProcessPool;
constructor(public readonly options: WorkerRuntimeOptions) {}
@@ -95,6 +98,42 @@ class DevSupervisor implements WorkerRuntime {
this.runLimiter = pLimit(maxConcurrentRuns);
// Initialize the task run process pool
const env = await this.#getEnvVars();
const enableProcessReuse =
typeof this.options.config.experimental_processKeepAlive === "boolean"
? this.options.config.experimental_processKeepAlive
: typeof this.options.config.experimental_processKeepAlive === "object"
? this.options.config.experimental_processKeepAlive.enabled
: false;
const maxPoolSize =
typeof this.options.config.experimental_processKeepAlive === "object"
? this.options.config.experimental_processKeepAlive.devMaxPoolSize ?? 25
: 25;
const maxExecutionsPerProcess =
typeof this.options.config.experimental_processKeepAlive === "object"
? this.options.config.experimental_processKeepAlive.maxExecutionsPerProcess ?? 50
: 50;
if (enableProcessReuse) {
logger.debug("[DevSupervisor] Enabling process reuse", {
enableProcessReuse,
maxPoolSize,
maxExecutionsPerProcess,
});
}
this.taskRunProcessPool = new TaskRunProcessPool({
env,
cwd: this.options.config.workingDir,
enableProcessReuse,
maxPoolSize,
maxExecutionsPerProcess,
});
this.socket = this.#createSocket();
//start an SSE connection for presence
@@ -111,6 +150,17 @@ class DevSupervisor implements WorkerRuntime {
} catch (error) {
logger.debug("[DevSupervisor] shutdown, socket failed to close", { error });
}
// Shutdown the task run process pool
if (this.taskRunProcessPool) {
const [shutdownError] = await tryCatch(this.taskRunProcessPool.shutdown());
if (shutdownError) {
logger.debug("[DevSupervisor] shutdown, task run process pool failed to shutdown", {
error: shutdownError,
});
}
}
}
async initializeWorker(
@@ -293,12 +343,21 @@ class DevSupervisor implements WorkerRuntime {
continue;
}
if (!this.taskRunProcessPool) {
logger.debug(`[DevSupervisor] dequeueRuns. No task run process pool`, {
run: message.run.friendlyId,
worker,
});
continue;
}
//new run
runController = new DevRunController({
runFriendlyId: message.run.friendlyId,
worker: worker,
httpClient: this.options.client,
logLevel: this.options.args.logLevel,
taskRunProcessPool: this.taskRunProcessPool,
onFinished: () => {
logger.debug("[DevSupervisor] Run finished", { runId: message.run.friendlyId });
@@ -574,6 +633,10 @@ class DevSupervisor implements WorkerRuntime {
return;
}
if (worker.serverWorker?.version) {
this.taskRunProcessPool?.deprecateVersion(worker.serverWorker?.version);
}
if (this.#workerHasInProgressRuns(friendlyId)) {
return;
}
packages/cli-v3/src/dev/taskRunProcessPool.ts
+284
View File
@@ -0,0 +1,284 @@
import {
MachinePresetResources,
ServerBackgroundWorker,
WorkerManifest,
} from "@trigger.dev/core/v3";
import { TaskRunProcess } from "../executions/taskRunProcess.js";
import { logger } from "../utilities/logger.js";
export type TaskRunProcessPoolOptions = {
env: Record<string, string>;
cwd: string;
enableProcessReuse: boolean;
maxPoolSize?: number;
maxExecutionsPerProcess?: number;
};
export class TaskRunProcessPool {
// Group processes by worker version
private availableProcessesByVersion: Map<string, TaskRunProcess[]> = new Map();
private busyProcessesByVersion: Map<string, Set<TaskRunProcess>> = new Map();
private readonly options: TaskRunProcessPoolOptions;
private readonly maxPoolSize: number;
private readonly maxExecutionsPerProcess: number;
private readonly executionCountsPerProcess: Map<number, number> = new Map();
private readonly deprecatedVersions: Set<string> = new Set();
constructor(options: TaskRunProcessPoolOptions) {
this.options = options;
this.maxPoolSize = options.maxPoolSize ?? 3;
this.maxExecutionsPerProcess = options.maxExecutionsPerProcess ?? 50;
}
deprecateVersion(version: string) {
this.deprecatedVersions.add(version);
logger.debug("[TaskRunProcessPool] Deprecating version", { version });
const versionProcesses = this.availableProcessesByVersion.get(version) || [];
const processesToKill = versionProcesses.filter((process) => !process.isExecuting());
Promise.all(processesToKill.map((process) => this.killProcess(process))).then(() => {
this.availableProcessesByVersion.delete(version);
});
}
async getProcess(
workerManifest: WorkerManifest,
serverWorker: ServerBackgroundWorker,
machineResources: MachinePresetResources,
env?: Record<string, string>
): Promise<{ taskRunProcess: TaskRunProcess; isReused: boolean }> {
const version = serverWorker.version || "unknown";
// Try to reuse an existing process if enabled
if (this.options.enableProcessReuse) {
const reusableProcess = this.findReusableProcess(version);
if (reusableProcess) {
const availableCount = this.availableProcessesByVersion.get(version)?.length || 0;
const busyCount = this.busyProcessesByVersion.get(version)?.size || 0;
logger.debug("[TaskRunProcessPool] Reusing existing process", {
version,
availableCount,
busyCount,
});
// Remove from available and add to busy for this version
const availableProcesses = this.availableProcessesByVersion.get(version) || [];
this.availableProcessesByVersion.set(
version,
availableProcesses.filter((p) => p !== reusableProcess)
);
if (!this.busyProcessesByVersion.has(version)) {
this.busyProcessesByVersion.set(version, new Set());
}
this.busyProcessesByVersion.get(version)!.add(reusableProcess);
return { taskRunProcess: reusableProcess, isReused: true };
} else {
const availableCount = this.availableProcessesByVersion.get(version)?.length || 0;
const busyCount = this.busyProcessesByVersion.get(version)?.size || 0;
logger.debug("[TaskRunProcessPool] No reusable process found", {
version,
availableCount,
busyCount,
});
}
}
// Create new process
const availableCount = this.availableProcessesByVersion.get(version)?.length || 0;
const busyCount = this.busyProcessesByVersion.get(version)?.size || 0;
logger.debug("[TaskRunProcessPool] Creating new process", {
version,
availableCount,
busyCount,
});
const newProcess = new TaskRunProcess({
workerManifest,
env: {
...this.options.env,
...env,
},
serverWorker,
machineResources,
cwd: this.options.cwd,
}).initialize();
// Add to busy processes for this version
if (!this.busyProcessesByVersion.has(version)) {
this.busyProcessesByVersion.set(version, new Set());
}
this.busyProcessesByVersion.get(version)!.add(newProcess);
return { taskRunProcess: newProcess, isReused: false };
}
async returnProcess(process: TaskRunProcess, version: string): Promise<void> {
// Remove from busy processes for this version
const busyProcesses = this.busyProcessesByVersion.get(version);
if (busyProcesses) {
busyProcesses.delete(process);
}
if (process.pid) {
this.executionCountsPerProcess.set(
process.pid,
(this.executionCountsPerProcess.get(process.pid) ?? 0) + 1
);
}
if (this.shouldReuseProcess(process, version)) {
const availableCount = this.availableProcessesByVersion.get(version)?.length || 0;
const busyCount = this.busyProcessesByVersion.get(version)?.size || 0;
logger.debug("[TaskRunProcessPool] Returning process to pool", {
version,
availableCount,
busyCount,
});
// Clean up but don't kill the process
try {
await process.cleanup(false);
// Add to available processes for this version
if (!this.availableProcessesByVersion.has(version)) {
this.availableProcessesByVersion.set(version, []);
}
this.availableProcessesByVersion.get(version)!.push(process);
} catch (error) {
logger.debug("[TaskRunProcessPool] Failed to cleanup process for reuse, killing it", {
error,
});
await this.killProcess(process);
}
} else {
const availableCount = this.availableProcessesByVersion.get(version)?.length || 0;
const busyCount = this.busyProcessesByVersion.get(version)?.size || 0;
logger.debug("[TaskRunProcessPool] Killing process", {
version,
availableCount,
busyCount,
});
await this.killProcess(process);
}
}
private findReusableProcess(version: string): TaskRunProcess | undefined {
const availableProcesses = this.availableProcessesByVersion.get(version) || [];
return availableProcesses.find((process) => this.isProcessHealthy(process));
}
private shouldReuseProcess(process: TaskRunProcess, version: string): boolean {
const isHealthy = this.isProcessHealthy(process);
const isBeingKilled = process.isBeingKilled;
const pid = process.pid;
const executionCount = this.executionCountsPerProcess.get(pid ?? 0) ?? 0;
const availableCount = this.availableProcessesByVersion.get(version)?.length || 0;
const busyCount = this.busyProcessesByVersion.get(version)?.size || 0;
const isDeprecated = this.deprecatedVersions.has(version);
logger.debug("[TaskRunProcessPool] Checking if process should be reused", {
version,
isHealthy,
isBeingKilled,
pid,
availableCount,
busyCount,
maxPoolSize: this.maxPoolSize,
executionCount,
isDeprecated,
});
return (
this.options.enableProcessReuse &&
this.isProcessHealthy(process) &&
availableCount < this.maxPoolSize &&
executionCount < this.maxExecutionsPerProcess &&
!isDeprecated
);
}
private isProcessHealthy(process: TaskRunProcess): boolean {
// Basic health checks - we can expand this later
return !process.isBeingKilled && process.pid !== undefined;
}
private async killProcess(process: TaskRunProcess): Promise<void> {
try {
await process.cleanup(true);
} catch (error) {
logger.debug("[TaskRunProcessPool] Error killing process", { error });
}
}
async shutdown(): Promise<void> {
const totalAvailable = Array.from(this.availableProcessesByVersion.values()).reduce(
(sum, processes) => sum + processes.length,
0
);
const totalBusy = Array.from(this.busyProcessesByVersion.values()).reduce(
(sum, processes) => sum + processes.size,
0
);
logger.debug("[TaskRunProcessPool] Shutting down pool", {
availableCount: totalAvailable,
busyCount: totalBusy,
versions: Array.from(this.availableProcessesByVersion.keys()),
});
// Kill all available processes across all versions
const allAvailableProcesses = Array.from(this.availableProcessesByVersion.values()).flat();
await Promise.all(allAvailableProcesses.map((process) => this.killProcess(process)));
this.availableProcessesByVersion.clear();
// Kill all busy processes across all versions
const allBusyProcesses = Array.from(this.busyProcessesByVersion.values())
.map((processSet) => Array.from(processSet))
.flat();
await Promise.all(allBusyProcesses.map((process) => this.killProcess(process)));
this.busyProcessesByVersion.clear();
}
getStats() {
const totalAvailable = Array.from(this.availableProcessesByVersion.values()).reduce(
(sum, processes) => sum + processes.length,
0
);
const totalBusy = Array.from(this.busyProcessesByVersion.values()).reduce(
(sum, processes) => sum + processes.size,
0
);
const statsByVersion: Record<string, { available: number; busy: number }> = {};
for (const [version, processes] of this.availableProcessesByVersion.entries()) {
statsByVersion[version] = {
available: processes.length,
busy: this.busyProcessesByVersion.get(version)?.size || 0,
};
}
for (const [version, processes] of this.busyProcessesByVersion.entries()) {
if (!statsByVersion[version]) {
statsByVersion[version] = {
available: 0,
busy: processes.size,
};
}
}
return {
availableCount: totalAvailable,
busyCount: totalBusy,
totalCount: totalAvailable + totalBusy,
byVersion: statsByVersion,
};
}
}
packages/cli-v3/src/entryPoints/dev-run-controller.ts
+50 -29
View File
@@ -19,6 +19,7 @@ import { sanitizeEnvVars } from "../utilities/sanitizeEnvVars.js";
import { join } from "node:path";
import { BackgroundWorker } from "../dev/backgroundWorker.js";
import { eventBus } from "../utilities/eventBus.js";
import { TaskRunProcessPool } from "../dev/taskRunProcessPool.js";
type DevRunControllerOptions = {
runFriendlyId: string;
@@ -26,6 +27,7 @@ type DevRunControllerOptions = {
httpClient: CliApiClient;
logLevel: LogLevel;
heartbeatIntervalSeconds?: number;
taskRunProcessPool: TaskRunProcessPool;
onSubscribeToRunNotifications: (run: Run, snapshot: Snapshot) => void;
onUnsubscribeFromRunNotifications: (run: Run, snapshot: Snapshot) => void;
onFinished: () => void;
@@ -605,45 +607,58 @@ export class DevRunController {
this.snapshotPoller.start();
this.taskRunProcess = new TaskRunProcess({
workerManifest: this.opts.worker.manifest,
env: {
...sanitizeEnvVars(envVars ?? {}),
...sanitizeEnvVars(this.opts.worker.params.env),
TRIGGER_WORKER_MANIFEST_PATH: join(this.opts.worker.build.outputPath, "index.json"),
RUN_WORKER_SHOW_LOGS: this.opts.logLevel === "debug" ? "true" : "false",
TRIGGER_PROJECT_REF: execution.project.ref,
},
serverWorker: {
// Get process from pool instead of creating new one
const { taskRunProcess, isReused } = await this.opts.taskRunProcessPool.getProcess(
this.opts.worker.manifest,
{
id: "unmanaged",
contentHash: this.opts.worker.build.contentHash,
version: this.opts.worker.serverWorker?.version,
engine: "V2",
},
machineResources: execution.machine,
}).initialize();
execution.machine,
{
TRIGGER_WORKER_MANIFEST_PATH: join(this.opts.worker.build.outputPath, "index.json"),
RUN_WORKER_SHOW_LOGS: this.opts.logLevel === "debug" ? "true" : "false",
TRIGGER_WORKER_VERSION: this.opts.worker.serverWorker?.version,
}
);
logger.debug("executing task run process", {
this.taskRunProcess = taskRunProcess;
// Update the process environment for this specific run
// Note: We may need to enhance TaskRunProcess to support updating env vars
logger.debug("executing task run process from pool", {
attemptNumber: execution.attempt.number,
runId: execution.run.id,
});
const completion = await this.taskRunProcess.execute({
payload: {
execution,
traceContext: execution.run.traceContext ?? {},
metrics,
const completion = await this.taskRunProcess.execute(
{
payload: {
execution,
traceContext: execution.run.traceContext ?? {},
metrics,
},
messageId: run.friendlyId,
env: {
...sanitizeEnvVars(envVars ?? {}),
...sanitizeEnvVars(this.opts.worker.params.env),
TRIGGER_PROJECT_REF: execution.project.ref,
},
},
messageId: run.friendlyId,
});
isReused
);
logger.debug("Completed run", completion);
// Return process to pool instead of killing it
try {
await this.taskRunProcess.cleanup(true);
const version = this.opts.worker.serverWorker?.version || "unknown";
await this.opts.taskRunProcessPool.returnProcess(this.taskRunProcess, version);
this.taskRunProcess = undefined;
} catch (error) {
logger.debug("Failed to cleanup task run process, submitting completion anyway", {
logger.debug("Failed to return task run process to pool, submitting completion anyway", {
error,
});
}
@@ -758,11 +773,15 @@ export class DevRunController {
}
private async runFinished() {
// Kill the run process
try {
await this.taskRunProcess?.kill("SIGKILL");
} catch (error) {
logger.debug("Failed to kill task run process", { error });
// Return the process to the pool instead of killing it directly
if (this.taskRunProcess) {
try {
const version = this.opts.worker.serverWorker?.version || "unknown";
await this.opts.taskRunProcessPool.returnProcess(this.taskRunProcess, version);
this.taskRunProcess = undefined;
} catch (error) {
logger.debug("Failed to return task run process to pool during runFinished", { error });
}
}
this.runHeartbeat.stop();
@@ -794,9 +813,11 @@ export class DevRunController {
if (this.taskRunProcess && !this.taskRunProcess.isBeingKilled) {
try {
await this.taskRunProcess.cleanup(true);
const version = this.opts.worker.serverWorker?.version || "unknown";
await this.opts.taskRunProcessPool.returnProcess(this.taskRunProcess, version);
this.taskRunProcess = undefined;
} catch (error) {
logger.debug("Failed to cleanup task run process", { error });
logger.debug("Failed to return task run process to pool during stop", { error });
}
}
packages/cli-v3/src/entryPoints/dev-run-worker.ts
+242 -145
View File
@@ -19,6 +19,7 @@ import {
runMetadata,
runtime,
runTimelineMetrics,
taskContext,
TaskRunErrorCodes,
TaskRunExecution,
timeout,
@@ -58,6 +59,7 @@ import sourceMapSupport from "source-map-support";
import { env } from "std-env";
import { normalizeImportPath } from "../utilities/normalizeImportPath.js";
import { VERSION } from "../version.js";
import { promiseWithResolvers } from "@trigger.dev/core/utils";
sourceMapSupport.install({
handleUncaughtExceptions: false,
@@ -99,6 +101,10 @@ process.on("uncaughtException", function (error, origin) {
}
});
process.title = `trigger-dev-run-worker (${
getEnvVar("TRIGGER_WORKER_VERSION") ?? "unknown version"
})`;
const heartbeatIntervalMs = getEnvVar("HEARTBEAT_INTERVAL_MS");
const standardLocalsManager = new StandardLocalsManager();
@@ -112,8 +118,12 @@ runTimelineMetrics.setGlobalManager(standardRunTimelineMetricsManager);
const devUsageManager = new DevUsageManager();
usage.setGlobalUsageManager(devUsageManager);
timeout.setGlobalManager(new UsageTimeoutManager(devUsageManager));
resourceCatalog.setGlobalResourceCatalog(new StandardResourceCatalog());
const usageTimeoutManager = new UsageTimeoutManager(devUsageManager);
timeout.setGlobalManager(usageTimeoutManager);
const standardResourceCatalog = new StandardResourceCatalog();
resourceCatalog.setGlobalResourceCatalog(standardResourceCatalog);
const durableClock = new DurableClock();
clock.setGlobalClock(durableClock);
@@ -153,84 +163,110 @@ async function loadWorkerManifest() {
return WorkerManifest.parse(raw);
}
async function doBootstrap() {
return await runTimelineMetrics.measureMetric("trigger.dev/start", "bootstrap", {}, async () => {
log("Bootstrapping worker");
const workerManifest = await loadWorkerManifest();
resourceCatalog.registerWorkerManifest(workerManifest);
const { config, handleError } = await importConfig(workerManifest.configPath);
const tracingSDK = new TracingSDK({
url: env.OTEL_EXPORTER_OTLP_ENDPOINT ?? "http://0.0.0.0:4318",
instrumentations: config.telemetry?.instrumentations ?? config.instrumentations ?? [],
exporters: config.telemetry?.exporters ?? [],
logExporters: config.telemetry?.logExporters ?? [],
diagLogLevel: (env.OTEL_LOG_LEVEL as TracingDiagnosticLogLevel) ?? "none",
forceFlushTimeoutMillis: 30_000,
});
const otelTracer: Tracer = tracingSDK.getTracer("trigger-dev-worker", VERSION);
const otelLogger: Logger = tracingSDK.getLogger("trigger-dev-worker", VERSION);
const tracer = new TriggerTracer({ tracer: otelTracer, logger: otelLogger });
const consoleInterceptor = new ConsoleInterceptor(
otelLogger,
typeof config.enableConsoleLogging === "boolean" ? config.enableConsoleLogging : true,
typeof config.disableConsoleInterceptor === "boolean"
? config.disableConsoleInterceptor
: false
);
const configLogLevel = triggerLogLevel ?? config.logLevel ?? "info";
const otelTaskLogger = new OtelTaskLogger({
logger: otelLogger,
tracer: tracer,
level: logLevels.includes(configLogLevel as any) ? (configLogLevel as LogLevel) : "info",
});
logger.setGlobalTaskLogger(otelTaskLogger);
if (config.init) {
lifecycleHooks.registerGlobalInitHook({
id: "config",
fn: config.init as AnyOnInitHookFunction,
});
}
if (config.onStart) {
lifecycleHooks.registerGlobalStartHook({
id: "config",
fn: config.onStart as AnyOnStartHookFunction,
});
}
if (config.onSuccess) {
lifecycleHooks.registerGlobalSuccessHook({
id: "config",
fn: config.onSuccess as AnyOnSuccessHookFunction,
});
}
if (config.onFailure) {
lifecycleHooks.registerGlobalFailureHook({
id: "config",
fn: config.onFailure as AnyOnFailureHookFunction,
});
}
if (handleError) {
lifecycleHooks.registerGlobalCatchErrorHook({
id: "config",
fn: handleError as AnyOnCatchErrorHookFunction,
});
}
log("Bootstrapped worker");
return {
tracer,
tracingSDK,
consoleInterceptor,
config,
workerManifest,
};
});
}
let bootstrapCache:
| {
tracer: TriggerTracer;
tracingSDK: TracingSDK;
consoleInterceptor: ConsoleInterceptor;
config: TriggerConfig;
workerManifest: WorkerManifest;
}
| undefined;
async function bootstrap() {
const workerManifest = await loadWorkerManifest();
resourceCatalog.registerWorkerManifest(workerManifest);
const { config, handleError } = await importConfig(workerManifest.configPath);
const tracingSDK = new TracingSDK({
url: env.OTEL_EXPORTER_OTLP_ENDPOINT ?? "http://0.0.0.0:4318",
instrumentations: config.telemetry?.instrumentations ?? config.instrumentations ?? [],
exporters: config.telemetry?.exporters ?? [],
logExporters: config.telemetry?.logExporters ?? [],
diagLogLevel: (env.OTEL_LOG_LEVEL as TracingDiagnosticLogLevel) ?? "none",
forceFlushTimeoutMillis: 30_000,
});
const otelTracer: Tracer = tracingSDK.getTracer("trigger-dev-worker", VERSION);
const otelLogger: Logger = tracingSDK.getLogger("trigger-dev-worker", VERSION);
const tracer = new TriggerTracer({ tracer: otelTracer, logger: otelLogger });
const consoleInterceptor = new ConsoleInterceptor(
otelLogger,
typeof config.enableConsoleLogging === "boolean" ? config.enableConsoleLogging : true,
typeof config.disableConsoleInterceptor === "boolean" ? config.disableConsoleInterceptor : false
);
const configLogLevel = triggerLogLevel ?? config.logLevel ?? "info";
const otelTaskLogger = new OtelTaskLogger({
logger: otelLogger,
tracer: tracer,
level: logLevels.includes(configLogLevel as any) ? (configLogLevel as LogLevel) : "info",
});
logger.setGlobalTaskLogger(otelTaskLogger);
if (config.init) {
lifecycleHooks.registerGlobalInitHook({
id: "config",
fn: config.init as AnyOnInitHookFunction,
});
if (!bootstrapCache) {
bootstrapCache = await doBootstrap();
}
if (config.onStart) {
lifecycleHooks.registerGlobalStartHook({
id: "config",
fn: config.onStart as AnyOnStartHookFunction,
});
}
if (config.onSuccess) {
lifecycleHooks.registerGlobalSuccessHook({
id: "config",
fn: config.onSuccess as AnyOnSuccessHookFunction,
});
}
if (config.onFailure) {
lifecycleHooks.registerGlobalFailureHook({
id: "config",
fn: config.onFailure as AnyOnFailureHookFunction,
});
}
if (handleError) {
lifecycleHooks.registerGlobalCatchErrorHook({
id: "config",
fn: handleError as AnyOnCatchErrorHookFunction,
});
}
return {
tracer,
tracingSDK,
consoleInterceptor,
config,
workerManifest,
};
return bootstrapCache;
}
let _execution: TaskRunExecution | undefined;
@@ -238,23 +274,67 @@ let _isRunning = false;
let _isCancelled = false;
let _tracingSDK: TracingSDK | undefined;
let _executionMeasurement: UsageMeasurement | undefined;
const cancelController = new AbortController();
let _cancelController = new AbortController();
let _lastFlushPromise: Promise<void> | undefined;
let _sharedWorkerRuntime: SharedRuntimeManager | undefined;
let _lastEnv: Record<string, string> | undefined;
let _executionCount = 0;
function resetExecutionEnvironment() {
_execution = undefined;
_isRunning = false;
_isCancelled = false;
_executionMeasurement = undefined;
_cancelController = new AbortController();
standardLocalsManager.reset();
standardLifecycleHooksManager.reset();
standardRunTimelineMetricsManager.reset();
devUsageManager.reset();
usageTimeoutManager.reset();
runMetadataManager.reset();
waitUntilManager.reset();
_sharedWorkerRuntime?.reset();
durableClock.reset();
taskContext.disable();
log(`[${new Date().toISOString()}] Reset execution environment`);
}
const zodIpc = new ZodIpcConnection({
listenSchema: WorkerToExecutorMessageCatalog,
emitSchema: ExecutorToWorkerMessageCatalog,
process,
handlers: {
EXECUTE_TASK_RUN: async ({ execution, traceContext, metadata, metrics, env }, sender) => {
EXECUTE_TASK_RUN: async (
{ execution, traceContext, metadata, metrics, env, isWarmStart },
sender
) => {
if (env) {
populateEnv(env, {
override: true,
previousEnv: _lastEnv,
});
_lastEnv = env;
}
log(`[${new Date().toISOString()}] Received EXECUTE_TASK_RUN`, execution);
standardRunTimelineMetricsManager.registerMetricsFromExecution(metrics);
if (_lastFlushPromise) {
const now = performance.now();
await _lastFlushPromise;
const duration = performance.now() - now;
log(`[${new Date().toISOString()}] Awaited last flush in ${duration}ms`);
}
resetExecutionEnvironment();
standardRunTimelineMetricsManager.registerMetricsFromExecution(metrics, isWarmStart);
if (_isRunning) {
logError("Worker is already running a task");
@@ -271,7 +351,7 @@ const zodIpc = new ZodIpcConnection({
usage: {
durationMs: 0,
},
metadata: runMetadataManager.stopAndReturnLastFlush(),
flushedMetadata: await runMetadataManager.stopAndReturnLastFlush(),
},
});
@@ -302,74 +382,82 @@ const zodIpc = new ZodIpcConnection({
usage: {
durationMs: 0,
},
metadata: runMetadataManager.stopAndReturnLastFlush(),
flushedMetadata: await runMetadataManager.stopAndReturnLastFlush(),
},
});
return;
}
try {
await runTimelineMetrics.measureMetric(
"trigger.dev/start",
"import",
{
entryPoint: taskManifest.entryPoint,
file: taskManifest.filePath,
},
async () => {
const beforeImport = performance.now();
resourceCatalog.setCurrentFileContext(taskManifest.entryPoint, taskManifest.filePath);
// First attempt to get the task from the resource catalog
let task = resourceCatalog.getTask(execution.task.id);
// Load init file if it exists
if (workerManifest.initEntryPoint) {
try {
await import(normalizeImportPath(workerManifest.initEntryPoint));
log(`Loaded init file from ${workerManifest.initEntryPoint}`);
} catch (err) {
logError(`Failed to load init file`, err);
throw err;
if (!task) {
log(`Could not find task ${execution.task.id} in resource catalog, importing...`);
try {
await runTimelineMetrics.measureMetric(
"trigger.dev/start",
"import",
{
entryPoint: taskManifest.entryPoint,
file: taskManifest.filePath,
},
async () => {
const beforeImport = performance.now();
resourceCatalog.setCurrentFileContext(
taskManifest.entryPoint,
taskManifest.filePath
);
// Load init file if it exists
if (workerManifest.initEntryPoint) {
try {
await import(normalizeImportPath(workerManifest.initEntryPoint));
log(`Loaded init file from ${workerManifest.initEntryPoint}`);
} catch (err) {
logError(`Failed to load init file`, err);
throw err;
}
}
await import(normalizeImportPath(taskManifest.entryPoint));
resourceCatalog.clearCurrentFileContext();
const durationMs = performance.now() - beforeImport;
log(
`Imported task ${execution.task.id} [${taskManifest.entryPoint}] in ${durationMs}ms`
);
}
);
} catch (err) {
logError(`Failed to import task ${execution.task.id}`, err);
await import(normalizeImportPath(taskManifest.entryPoint));
resourceCatalog.clearCurrentFileContext();
const durationMs = performance.now() - beforeImport;
log(
`Imported task ${execution.task.id} [${taskManifest.entryPoint}] in ${durationMs}ms`
);
}
);
} catch (err) {
logError(`Failed to import task ${execution.task.id}`, err);
await sender.send("TASK_RUN_COMPLETED", {
execution,
result: {
ok: false,
id: execution.run.id,
error: {
type: "INTERNAL_ERROR",
code: TaskRunErrorCodes.COULD_NOT_IMPORT_TASK,
message: err instanceof Error ? err.message : String(err),
stackTrace: err instanceof Error ? err.stack : undefined,
await sender.send("TASK_RUN_COMPLETED", {
execution,
result: {
ok: false,
id: execution.run.id,
error: {
type: "INTERNAL_ERROR",
code: TaskRunErrorCodes.COULD_NOT_IMPORT_TASK,
message: err instanceof Error ? err.message : String(err),
stackTrace: err instanceof Error ? err.stack : undefined,
},
usage: {
durationMs: 0,
},
flushedMetadata: await runMetadataManager.stopAndReturnLastFlush(),
},
usage: {
durationMs: 0,
},
metadata: runMetadataManager.stopAndReturnLastFlush(),
},
});
});
return;
return;
}
// Now try and get the task again
task = resourceCatalog.getTask(execution.task.id);
}
process.title = `trigger-dev-worker: ${execution.task.id} ${execution.run.id}`;
// Import the task module
const task = resourceCatalog.getTask(execution.task.id);
if (!task) {
logError(`Could not find task ${execution.task.id}`);
@@ -385,7 +473,7 @@ const zodIpc = new ZodIpcConnection({
usage: {
durationMs: 0,
},
metadata: runMetadataManager.stopAndReturnLastFlush(),
flushedMetadata: await runMetadataManager.stopAndReturnLastFlush(),
},
});
@@ -393,12 +481,15 @@ const zodIpc = new ZodIpcConnection({
}
runMetadataManager.runId = execution.run.id;
_executionCount++;
const executor = new TaskExecutor(task, {
tracer,
tracingSDK,
consoleInterceptor,
retries: config.retries,
isWarmStart,
executionCount: _executionCount,
});
try {
@@ -413,7 +504,7 @@ const zodIpc = new ZodIpcConnection({
const timeoutController = timeout.abortAfterTimeout(execution.run.maxDuration);
const signal = AbortSignal.any([cancelController.signal, timeoutController.signal]);
const signal = AbortSignal.any([_cancelController.signal, timeoutController.signal]);
const { result } = await executor.execute(execution, metadata, traceContext, signal);
@@ -427,13 +518,14 @@ const zodIpc = new ZodIpcConnection({
usage: {
durationMs: usageSample.cpuTime,
},
metadata: runMetadataManager.stopAndReturnLastFlush(),
flushedMetadata: await runMetadataManager.stopAndReturnLastFlush(),
},
});
}
} finally {
_execution = undefined;
_isRunning = false;
log(`[${new Date().toISOString()}] Task run completed`);
}
} catch (err) {
logError("Failed to execute task", err);
@@ -452,14 +544,14 @@ const zodIpc = new ZodIpcConnection({
usage: {
durationMs: 0,
},
metadata: runMetadataManager.stopAndReturnLastFlush(),
flushedMetadata: await runMetadataManager.stopAndReturnLastFlush(),
},
});
}
},
CANCEL: async ({ timeoutInMs }) => {
_isCancelled = true;
cancelController.abort("run cancelled");
_cancelController.abort("run cancelled");
await callCancelHooks(timeoutInMs);
if (_executionMeasurement) {
usage.stop(_executionMeasurement);
@@ -470,7 +562,7 @@ const zodIpc = new ZodIpcConnection({
await flushAll(timeoutInMs);
},
RESOLVE_WAITPOINT: async ({ waitpoint }) => {
sharedWorkerRuntime.resolveWaitpoints([waitpoint]);
_sharedWorkerRuntime?.resolveWaitpoints([waitpoint]);
},
},
});
@@ -490,6 +582,10 @@ async function callCancelHooks(timeoutInMs: number = 10_000) {
async function flushAll(timeoutInMs: number = 10_000) {
const now = performance.now();
const { promise, resolve } = promiseWithResolvers<void>();
_lastFlushPromise = promise;
const results = await Promise.allSettled([
flushTracingSDK(timeoutInMs),
flushMetadata(timeoutInMs),
@@ -522,6 +618,9 @@ async function flushAll(timeoutInMs: number = 10_000) {
const duration = performance.now() - now;
log(`Flushed all in ${duration}ms`);
// Resolve the last flush promise
resolve();
}
async function flushTracingSDK(timeoutInMs: number = 10_000) {
@@ -554,10 +653,8 @@ async function flushMetadata(timeoutInMs: number = 10_000) {
};
}
const sharedWorkerRuntime = new SharedRuntimeManager(zodIpc, showInternalLogs);
runtime.setGlobalRuntimeManager(sharedWorkerRuntime);
process.title = "trigger-managed-worker";
_sharedWorkerRuntime = new SharedRuntimeManager(zodIpc, showInternalLogs);
runtime.setGlobalRuntimeManager(_sharedWorkerRuntime);
const heartbeatInterval = parseInt(heartbeatIntervalMs ?? "30000", 10);
packages/cli-v3/src/entryPoints/managed-index-worker.ts
+6
View File
@@ -159,6 +159,12 @@ await sendMessageInCatalog(
loaderEntryPoint: buildManifest.loaderEntryPoint,
customConditions: buildManifest.customConditions,
initEntryPoint: buildManifest.initEntryPoint,
processKeepAlive:
typeof config.experimental_processKeepAlive === "object"
? config.experimental_processKeepAlive
: typeof config.experimental_processKeepAlive === "boolean"
? { enabled: config.experimental_processKeepAlive }
: undefined,
timings,
},
importErrors,
packages/cli-v3/src/entryPoints/managed-run-worker.ts
+240 -150
View File
@@ -18,6 +18,7 @@ import {
runMetadata,
runtime,
runTimelineMetrics,
taskContext,
TaskRunErrorCodes,
TaskRunExecution,
timeout,
@@ -58,6 +59,7 @@ import sourceMapSupport from "source-map-support";
import { env } from "std-env";
import { normalizeImportPath } from "../utilities/normalizeImportPath.js";
import { VERSION } from "../version.js";
import { promiseWithResolvers } from "@trigger.dev/core/utils";
sourceMapSupport.install({
handleUncaughtExceptions: false,
@@ -110,15 +112,18 @@ lifecycleHooks.setGlobalLifecycleHooksManager(standardLifecycleHooksManager);
const standardRunTimelineMetricsManager = new StandardRunTimelineMetricsManager();
runTimelineMetrics.setGlobalManager(standardRunTimelineMetricsManager);
resourceCatalog.setGlobalResourceCatalog(new StandardResourceCatalog());
const standardResourceCatalog = new StandardResourceCatalog();
resourceCatalog.setGlobalResourceCatalog(standardResourceCatalog);
const durableClock = new DurableClock();
clock.setGlobalClock(durableClock);
const runMetadataManager = new StandardMetadataManager(
apiClientManager.clientOrThrow(),
getEnvVar("TRIGGER_STREAM_URL", getEnvVar("TRIGGER_API_URL")) ?? "https://api.trigger.dev"
);
runMetadata.setGlobalManager(runMetadataManager);
const waitUntilManager = new StandardWaitUntilManager();
waitUntil.setGlobalManager(waitUntilManager);
// Wait for all streams to finish before completing the run
@@ -149,86 +154,108 @@ async function loadWorkerManifest() {
return WorkerManifest.parse(raw);
}
async function doBootstrap() {
return await runTimelineMetrics.measureMetric("trigger.dev/start", "bootstrap", {}, async () => {
const workerManifest = await loadWorkerManifest();
resourceCatalog.registerWorkerManifest(workerManifest);
const { config, handleError } = await importConfig(
normalizeImportPath(workerManifest.configPath)
);
const tracingSDK = new TracingSDK({
url: env.OTEL_EXPORTER_OTLP_ENDPOINT ?? "http://0.0.0.0:4318",
instrumentations: config.instrumentations ?? [],
diagLogLevel: (env.OTEL_LOG_LEVEL as TracingDiagnosticLogLevel) ?? "none",
forceFlushTimeoutMillis: 30_000,
exporters: config.telemetry?.exporters ?? [],
logExporters: config.telemetry?.logExporters ?? [],
});
const otelTracer: Tracer = tracingSDK.getTracer("trigger-dev-worker", VERSION);
const otelLogger: Logger = tracingSDK.getLogger("trigger-dev-worker", VERSION);
const tracer = new TriggerTracer({ tracer: otelTracer, logger: otelLogger });
const consoleInterceptor = new ConsoleInterceptor(
otelLogger,
typeof config.enableConsoleLogging === "boolean" ? config.enableConsoleLogging : true,
typeof config.disableConsoleInterceptor === "boolean"
? config.disableConsoleInterceptor
: false
);
const configLogLevel = triggerLogLevel ?? config.logLevel ?? "info";
const otelTaskLogger = new OtelTaskLogger({
logger: otelLogger,
tracer: tracer,
level: logLevels.includes(configLogLevel as any) ? (configLogLevel as LogLevel) : "info",
});
logger.setGlobalTaskLogger(otelTaskLogger);
if (config.init) {
lifecycleHooks.registerGlobalInitHook({
id: "config",
fn: config.init as AnyOnInitHookFunction,
});
}
if (config.onStart) {
lifecycleHooks.registerGlobalStartHook({
id: "config",
fn: config.onStart as AnyOnStartHookFunction,
});
}
if (config.onSuccess) {
lifecycleHooks.registerGlobalSuccessHook({
id: "config",
fn: config.onSuccess as AnyOnSuccessHookFunction,
});
}
if (config.onFailure) {
lifecycleHooks.registerGlobalFailureHook({
id: "config",
fn: config.onFailure as AnyOnFailureHookFunction,
});
}
if (handleError) {
lifecycleHooks.registerGlobalCatchErrorHook({
id: "config",
fn: handleError as AnyOnCatchErrorHookFunction,
});
}
return {
tracer,
tracingSDK,
consoleInterceptor,
config,
workerManifest,
};
});
}
let bootstrapCache:
| {
tracer: TriggerTracer;
tracingSDK: TracingSDK;
consoleInterceptor: ConsoleInterceptor;
config: TriggerConfig;
workerManifest: WorkerManifest;
}
| undefined;
async function bootstrap() {
const workerManifest = await loadWorkerManifest();
resourceCatalog.registerWorkerManifest(workerManifest);
const { config, handleError } = await importConfig(
normalizeImportPath(workerManifest.configPath)
);
const tracingSDK = new TracingSDK({
url: env.OTEL_EXPORTER_OTLP_ENDPOINT ?? "http://0.0.0.0:4318",
instrumentations: config.instrumentations ?? [],
diagLogLevel: (env.OTEL_LOG_LEVEL as TracingDiagnosticLogLevel) ?? "none",
forceFlushTimeoutMillis: 30_000,
exporters: config.telemetry?.exporters ?? [],
logExporters: config.telemetry?.logExporters ?? [],
});
const otelTracer: Tracer = tracingSDK.getTracer("trigger-dev-worker", VERSION);
const otelLogger: Logger = tracingSDK.getLogger("trigger-dev-worker", VERSION);
const tracer = new TriggerTracer({ tracer: otelTracer, logger: otelLogger });
const consoleInterceptor = new ConsoleInterceptor(
otelLogger,
typeof config.enableConsoleLogging === "boolean" ? config.enableConsoleLogging : true,
typeof config.disableConsoleInterceptor === "boolean" ? config.disableConsoleInterceptor : false
);
const configLogLevel = triggerLogLevel ?? config.logLevel ?? "info";
const otelTaskLogger = new OtelTaskLogger({
logger: otelLogger,
tracer: tracer,
level: logLevels.includes(configLogLevel as any) ? (configLogLevel as LogLevel) : "info",
});
logger.setGlobalTaskLogger(otelTaskLogger);
if (config.init) {
lifecycleHooks.registerGlobalInitHook({
id: "config",
fn: config.init as AnyOnInitHookFunction,
});
if (!bootstrapCache) {
bootstrapCache = await doBootstrap();
}
if (config.onStart) {
lifecycleHooks.registerGlobalStartHook({
id: "config",
fn: config.onStart as AnyOnStartHookFunction,
});
}
if (config.onSuccess) {
lifecycleHooks.registerGlobalSuccessHook({
id: "config",
fn: config.onSuccess as AnyOnSuccessHookFunction,
});
}
if (config.onFailure) {
lifecycleHooks.registerGlobalFailureHook({
id: "config",
fn: config.onFailure as AnyOnFailureHookFunction,
});
}
if (handleError) {
lifecycleHooks.registerGlobalCatchErrorHook({
id: "config",
fn: handleError as AnyOnCatchErrorHookFunction,
});
}
return {
tracer,
tracingSDK,
consoleInterceptor,
config,
workerManifest,
};
return bootstrapCache;
}
let _execution: TaskRunExecution | undefined;
@@ -236,7 +263,33 @@ let _isRunning = false;
let _isCancelled = false;
let _tracingSDK: TracingSDK | undefined;
let _executionMeasurement: UsageMeasurement | undefined;
const cancelController = new AbortController();
let _cancelController = new AbortController();
let _lastFlushPromise: Promise<void> | undefined;
let _sharedWorkerRuntime: SharedRuntimeManager | undefined;
function resetExecutionEnvironment() {
_execution = undefined;
_isRunning = false;
_isCancelled = false;
_executionMeasurement = undefined;
_cancelController = new AbortController();
standardLocalsManager.reset();
standardLifecycleHooksManager.reset();
standardRunTimelineMetricsManager.reset();
usage.reset();
timeout.reset();
runMetadataManager.reset();
waitUntilManager.reset();
_sharedWorkerRuntime?.reset();
durableClock.reset();
taskContext.disable();
console.log(`[${new Date().toISOString()}] Reset execution environment`);
}
let _lastEnv: Record<string, string> | undefined;
let _executionCount = 0;
const zodIpc = new ZodIpcConnection({
listenSchema: WorkerToExecutorMessageCatalog,
@@ -250,16 +303,35 @@ const zodIpc = new ZodIpcConnection({
if (env) {
populateEnv(env, {
override: true,
previousEnv: _lastEnv,
});
_lastEnv = env;
}
console.log(
`[${new Date().toISOString()}] Received EXECUTE_TASK_RUN isWarmStart ${String(isWarmStart)}`
);
if (_lastFlushPromise) {
const now = performance.now();
await _lastFlushPromise;
const duration = performance.now() - now;
console.log(`[${new Date().toISOString()}] Awaited last flush in ${duration}ms`);
}
resetExecutionEnvironment();
initializeUsageManager({
usageIntervalMs: getEnvVar("USAGE_HEARTBEAT_INTERVAL_MS"),
usageEventUrl: getEnvVar("USAGE_EVENT_URL"),
triggerJWT: getEnvVar("TRIGGER_JWT"),
});
standardRunTimelineMetricsManager.registerMetricsFromExecution(metrics);
standardRunTimelineMetricsManager.registerMetricsFromExecution(metrics, isWarmStart);
console.log(`[${new Date().toISOString()}] Received EXECUTE_TASK_RUN`, execution);
@@ -278,7 +350,7 @@ const zodIpc = new ZodIpcConnection({
usage: {
durationMs: 0,
},
metadata: runMetadataManager.stopAndReturnLastFlush(),
flushedMetadata: await runMetadataManager.stopAndReturnLastFlush(),
},
});
@@ -309,73 +381,81 @@ const zodIpc = new ZodIpcConnection({
usage: {
durationMs: 0,
},
metadata: runMetadataManager.stopAndReturnLastFlush(),
flushedMetadata: await runMetadataManager.stopAndReturnLastFlush(),
},
});
return;
}
try {
await runTimelineMetrics.measureMetric(
"trigger.dev/start",
"import",
{
entryPoint: taskManifest.entryPoint,
file: taskManifest.filePath,
},
async () => {
const beforeImport = performance.now();
resourceCatalog.setCurrentFileContext(taskManifest.entryPoint, taskManifest.filePath);
// Load init file if it exists
if (workerManifest.initEntryPoint) {
try {
await import(normalizeImportPath(workerManifest.initEntryPoint));
console.log(`Loaded init file from ${workerManifest.initEntryPoint}`);
} catch (err) {
console.error(`Failed to load init file`, err);
throw err;
}
}
await import(normalizeImportPath(taskManifest.entryPoint));
resourceCatalog.clearCurrentFileContext();
const durationMs = performance.now() - beforeImport;
console.log(
`Imported task ${execution.task.id} [${taskManifest.entryPoint}] in ${durationMs}ms`
);
}
);
} catch (err) {
console.error(`Failed to import task ${execution.task.id}`, err);
await sender.send("TASK_RUN_COMPLETED", {
execution,
result: {
ok: false,
id: execution.run.id,
error: {
type: "INTERNAL_ERROR",
code: TaskRunErrorCodes.COULD_NOT_IMPORT_TASK,
message: err instanceof Error ? err.message : String(err),
stackTrace: err instanceof Error ? err.stack : undefined,
},
usage: {
durationMs: 0,
},
metadata: runMetadataManager.stopAndReturnLastFlush(),
},
});
return;
}
process.title = `trigger-dev-worker: ${execution.task.id} ${execution.run.id}`;
// Import the task module
const task = resourceCatalog.getTask(execution.task.id);
let task = resourceCatalog.getTask(execution.task.id);
if (!task) {
try {
await runTimelineMetrics.measureMetric(
"trigger.dev/start",
"import",
{
entryPoint: taskManifest.entryPoint,
file: taskManifest.filePath,
},
async () => {
const beforeImport = performance.now();
resourceCatalog.setCurrentFileContext(
taskManifest.entryPoint,
taskManifest.filePath
);
// Load init file if it exists
if (workerManifest.initEntryPoint) {
try {
await import(normalizeImportPath(workerManifest.initEntryPoint));
console.log(`Loaded init file from ${workerManifest.initEntryPoint}`);
} catch (err) {
console.error(`Failed to load init file`, err);
throw err;
}
}
await import(normalizeImportPath(taskManifest.entryPoint));
resourceCatalog.clearCurrentFileContext();
const durationMs = performance.now() - beforeImport;
console.log(
`Imported task ${execution.task.id} [${taskManifest.entryPoint}] in ${durationMs}ms`
);
}
);
} catch (err) {
console.error(`Failed to import task ${execution.task.id}`, err);
await sender.send("TASK_RUN_COMPLETED", {
execution,
result: {
ok: false,
id: execution.run.id,
error: {
type: "INTERNAL_ERROR",
code: TaskRunErrorCodes.COULD_NOT_IMPORT_TASK,
message: err instanceof Error ? err.message : String(err),
stackTrace: err instanceof Error ? err.stack : undefined,
},
usage: {
durationMs: 0,
},
flushedMetadata: await runMetadataManager.stopAndReturnLastFlush(),
},
});
return;
}
process.title = `trigger-dev-worker: ${execution.task.id} ${execution.run.id}`;
// Now try and get the task again
task = resourceCatalog.getTask(execution.task.id);
}
if (!task) {
console.error(`Could not find task ${execution.task.id}`);
@@ -392,7 +472,7 @@ const zodIpc = new ZodIpcConnection({
usage: {
durationMs: 0,
},
metadata: runMetadataManager.stopAndReturnLastFlush(),
flushedMetadata: await runMetadataManager.stopAndReturnLastFlush(),
},
});
@@ -400,6 +480,7 @@ const zodIpc = new ZodIpcConnection({
}
runMetadataManager.runId = execution.run.id;
_executionCount++;
const executor = new TaskExecutor(task, {
tracer,
@@ -407,6 +488,7 @@ const zodIpc = new ZodIpcConnection({
consoleInterceptor,
retries: config.retries,
isWarmStart,
executionCount: _executionCount,
});
try {
@@ -421,7 +503,7 @@ const zodIpc = new ZodIpcConnection({
const timeoutController = timeout.abortAfterTimeout(execution.run.maxDuration);
const signal = AbortSignal.any([cancelController.signal, timeoutController.signal]);
const signal = AbortSignal.any([_cancelController.signal, timeoutController.signal]);
const { result } = await executor.execute(execution, metadata, traceContext, signal);
@@ -435,13 +517,15 @@ const zodIpc = new ZodIpcConnection({
usage: {
durationMs: usageSample.cpuTime,
},
metadata: runMetadataManager.stopAndReturnLastFlush(),
flushedMetadata: await runMetadataManager.stopAndReturnLastFlush(),
},
});
}
} finally {
_execution = undefined;
_isRunning = false;
console.log(`[${new Date().toISOString()}] Task run completed`);
}
} catch (err) {
console.error("Failed to execute task", err);
@@ -460,14 +544,14 @@ const zodIpc = new ZodIpcConnection({
usage: {
durationMs: 0,
},
metadata: runMetadataManager.stopAndReturnLastFlush(),
flushedMetadata: await runMetadataManager.stopAndReturnLastFlush(),
},
});
}
},
CANCEL: async ({ timeoutInMs }) => {
_isCancelled = true;
cancelController.abort("run cancelled");
_cancelController.abort("run cancelled");
await callCancelHooks(timeoutInMs);
if (_executionMeasurement) {
usage.stop(_executionMeasurement);
@@ -478,7 +562,7 @@ const zodIpc = new ZodIpcConnection({
await flushAll(timeoutInMs);
},
RESOLVE_WAITPOINT: async ({ waitpoint }) => {
sharedWorkerRuntime.resolveWaitpoints([waitpoint]);
_sharedWorkerRuntime?.resolveWaitpoints([waitpoint]);
},
},
});
@@ -498,6 +582,10 @@ async function callCancelHooks(timeoutInMs: number = 10_000) {
async function flushAll(timeoutInMs: number = 10_000) {
const now = performance.now();
const { promise, resolve } = promiseWithResolvers<void>();
_lastFlushPromise = promise;
const results = await Promise.allSettled([
flushUsage(timeoutInMs),
flushTracingSDK(timeoutInMs),
@@ -530,6 +618,9 @@ async function flushAll(timeoutInMs: number = 10_000) {
const duration = performance.now() - now;
console.log(`Flushed all in ${duration}ms`);
// Resolve the last flush promise
resolve();
}
async function flushUsage(timeoutInMs: number = 10_000) {
@@ -597,9 +688,8 @@ function initializeUsageManager({
timeout.setGlobalManager(new UsageTimeoutManager(devUsageManager));
}
const sharedWorkerRuntime = new SharedRuntimeManager(zodIpc, true);
runtime.setGlobalRuntimeManager(sharedWorkerRuntime);
_sharedWorkerRuntime = new SharedRuntimeManager(zodIpc, true);
runtime.setGlobalRuntimeManager(_sharedWorkerRuntime);
process.title = "trigger-managed-worker";
packages/cli-v3/src/entryPoints/managed/controller.ts
+44 -2
View File
@@ -11,6 +11,7 @@ import { RunnerEnv } from "./env.js";
import { ManagedRunLogger, RunLogger, SendDebugLogOptions } from "./logger.js";
import { EnvObject } from "std-env";
import { RunExecution } from "./execution.js";
import { TaskRunProcessProvider } from "./taskRunProcessProvider.js";
import { tryCatch } from "@trigger.dev/core/utils";
type ManagedRunControllerOptions = {
@@ -27,6 +28,7 @@ export class ManagedRunController {
private readonly warmStartClient: WarmStartClient | undefined;
private socket: SupervisorSocket;
private readonly logger: RunLogger;
private readonly taskRunProcessProvider: TaskRunProcessProvider;
private warmStartCount = 0;
private restoreCount = 0;
@@ -36,11 +38,17 @@ export class ManagedRunController {
private currentExecution: RunExecution | null = null;
private processKeepAliveEnabled: boolean;
private processKeepAliveMaxExecutionCount: number;
constructor(opts: ManagedRunControllerOptions) {
const env = new RunnerEnv(opts.env);
this.env = env;
this.workerManifest = opts.workerManifest;
this.processKeepAliveEnabled = opts.workerManifest.processKeepAlive?.enabled ?? false;
this.processKeepAliveMaxExecutionCount =
opts.workerManifest.processKeepAlive?.maxExecutionsPerProcess ?? 100;
this.httpClient = new WorkloadHttpClient({
workerApiUrl: this.workerApiUrl,
@@ -55,6 +63,15 @@ export class ManagedRunController {
env,
});
// Create the TaskRunProcessProvider
this.taskRunProcessProvider = new TaskRunProcessProvider({
workerManifest: this.workerManifest,
env: this.env,
logger: this.logger,
processKeepAliveEnabled: this.processKeepAliveEnabled,
processKeepAliveMaxExecutionCount: this.processKeepAliveMaxExecutionCount,
});
const properties = {
...env.raw,
TRIGGER_POD_SCHEDULED_AT_MS: env.TRIGGER_POD_SCHEDULED_AT_MS.toISOString(),
@@ -96,6 +113,7 @@ export class ManagedRunController {
restoreCount: this.restoreCount,
notificationCount: this.notificationCount,
lastNotificationAt: this.lastNotificationAt,
...this.taskRunProcessProvider.metrics,
};
}
@@ -189,7 +207,15 @@ export class ManagedRunController {
runId: runFriendlyId,
message: "killing existing execution before starting new run",
});
await this.currentExecution.kill().catch(() => {});
await this.currentExecution.shutdown().catch((error) => {
this.sendDebugLog({
runId: runFriendlyId,
message: "Error during execution shutdown",
properties: { error: error instanceof Error ? error.message : String(error) },
});
});
this.currentExecution = null;
}
@@ -203,6 +229,7 @@ export class ManagedRunController {
httpClient: this.httpClient,
logger: this.logger,
supervisorSocket: this.socket,
taskRunProcessProvider: this.taskRunProcessProvider,
});
}
@@ -298,7 +325,10 @@ export class ManagedRunController {
httpClient: this.httpClient,
logger: this.logger,
supervisorSocket: this.socket,
}).prepareForExecution({
taskRunProcessProvider: this.taskRunProcessProvider,
});
await this.currentExecution.prepareForExecution({
taskRunEnv: previousTaskRunEnv,
});
}
@@ -395,6 +425,7 @@ export class ManagedRunController {
});
this.currentExecution?.kill().catch(() => {});
this.taskRunProcessProvider.cleanup().catch(() => {});
process.exit(code);
}
@@ -534,6 +565,17 @@ export class ManagedRunController {
});
}
// Cleanup the task run process provider
const [cleanupError] = await tryCatch(this.taskRunProcessProvider.cleanup());
if (cleanupError) {
this.sendDebugLog({
runId: this.runFriendlyId,
message: "Error during task run process provider cleanup",
properties: { error: String(cleanupError) },
});
}
// Close the socket
this.socket.close();
}
packages/cli-v3/src/entryPoints/managed/execution.ts
+95 -80
View File
@@ -22,6 +22,7 @@ import { randomBytes } from "node:crypto";
import { SnapshotManager, SnapshotState } from "./snapshot.js";
import type { SupervisorSocket } from "./controller.js";
import { RunNotifier } from "./notifier.js";
import { TaskRunProcessProvider } from "./taskRunProcessProvider.js";
class ExecutionAbortError extends Error {
constructor(message: string) {
@@ -36,6 +37,7 @@ type RunExecutionOptions = {
httpClient: WorkloadHttpClient;
logger: RunLogger;
supervisorSocket: SupervisorSocket;
taskRunProcessProvider: TaskRunProcessProvider;
};
type RunExecutionPrepareOptions = {
@@ -77,6 +79,7 @@ export class RunExecution {
private supervisorSocket: SupervisorSocket;
private notifier?: RunNotifier;
private metadataClient?: MetadataClient;
private taskRunProcessProvider: TaskRunProcessProvider;
constructor(opts: RunExecutionOptions) {
this.id = randomBytes(4).toString("hex");
@@ -85,6 +88,7 @@ export class RunExecution {
this.httpClient = opts.httpClient;
this.logger = opts.logger;
this.supervisorSocket = opts.supervisorSocket;
this.taskRunProcessProvider = opts.taskRunProcessProvider;
this.restoreCount = 0;
this.executionAbortController = new AbortController();
@@ -111,18 +115,28 @@ export class RunExecution {
* Kills the current execution.
*/
public async kill({ exitExecution = true }: { exitExecution?: boolean } = {}) {
await this.taskRunProcess?.kill("SIGKILL");
if (this.taskRunProcess) {
await this.taskRunProcessProvider.handleProcessAbort(this.taskRunProcess);
}
if (exitExecution) {
this.shutdown("kill");
this.shutdownExecution("kill");
}
}
public async shutdown() {
if (this.taskRunProcess) {
await this.taskRunProcessProvider.handleProcessAbort(this.taskRunProcess);
}
this.shutdownExecution("shutdown");
}
/**
* Prepares the execution with task run environment variables.
* This should be called before executing, typically after a successful run to prepare for the next one.
*/
public prepareForExecution(opts: RunExecutionPrepareOptions): this {
public async prepareForExecution(opts: RunExecutionPrepareOptions) {
if (this.isShuttingDown) {
throw new Error("prepareForExecution called after execution shut down");
}
@@ -131,40 +145,14 @@ export class RunExecution {
throw new Error("prepareForExecution called after process was already created");
}
this.taskRunProcess = this.createTaskRunProcess({
envVars: opts.taskRunEnv,
this.taskRunProcess = await this.taskRunProcessProvider.getProcess({
taskRunEnv: opts.taskRunEnv,
isWarmStart: true,
});
return this;
}
private createTaskRunProcess({
envVars,
isWarmStart,
}: {
envVars: Record<string, string>;
isWarmStart?: boolean;
}) {
const taskRunProcess = new TaskRunProcess({
workerManifest: this.workerManifest,
env: {
...envVars,
...this.env.gatherProcessEnv(),
HEARTBEAT_INTERVAL_MS: String(this.env.TRIGGER_HEARTBEAT_INTERVAL_SECONDS * 1000),
},
serverWorker: {
id: "managed",
contentHash: this.env.TRIGGER_CONTENT_HASH,
version: this.env.TRIGGER_DEPLOYMENT_VERSION,
engine: "V2",
},
machineResources: {
cpu: Number(this.env.TRIGGER_MACHINE_CPU),
memory: Number(this.env.TRIGGER_MACHINE_MEMORY),
},
isWarmStart,
}).initialize();
private attachTaskRunProcessHandlers(taskRunProcess: TaskRunProcess): void {
taskRunProcess.unsafeDetachEvtHandlers();
taskRunProcess.onTaskRunHeartbeat.attach(async (runId) => {
if (!this.runFriendlyId) {
@@ -194,20 +182,23 @@ export class RunExecution {
taskRunProcess.onSetSuspendable.attach(async ({ suspendable }) => {
this.suspendable = suspendable;
});
return taskRunProcess;
}
/**
* Returns true if no run has been started yet and the process is prepared for the next run.
* Returns true if no run has been started yet and we're prepared for the next run.
*/
get canExecute(): boolean {
if (this.taskRunProcessProvider.hasPersistentProcess) {
return true;
}
// If we've ever had a run ID, this execution can't be reused
if (this._runFriendlyId) {
return false;
}
return !!this.taskRunProcess?.isPreparedForNextRun;
// We can execute if we have the task run environment ready
return !!this.currentTaskRunEnv;
}
/**
@@ -249,7 +240,10 @@ export class RunExecution {
if (this.currentAttemptNumber && this.currentAttemptNumber !== run.attemptNumber) {
this.sendDebugLog("error: attempt number mismatch", snapshotMetadata);
// This is a rogue execution, a new one will already have been created elsewhere
await this.exitTaskRunProcessWithoutFailingRun({ flush: false });
await this.exitTaskRunProcessWithoutFailingRun({
flush: false,
reason: "attempt number mismatch",
});
return;
}
@@ -263,7 +257,10 @@ export class RunExecution {
if (deprecated) {
this.sendDebugLog("run execution is deprecated", { incomingSnapshot: snapshot });
await this.exitTaskRunProcessWithoutFailingRun({ flush: false });
await this.exitTaskRunProcessWithoutFailingRun({
flush: false,
reason: "deprecated execution",
});
return;
}
@@ -286,13 +283,13 @@ export class RunExecution {
case "QUEUED": {
this.sendDebugLog("run was re-queued", snapshotMetadata);
await this.exitTaskRunProcessWithoutFailingRun({ flush: true });
await this.exitTaskRunProcessWithoutFailingRun({ flush: true, reason: "re-queued" });
return;
}
case "FINISHED": {
this.sendDebugLog("run is finished", snapshotMetadata);
await this.exitTaskRunProcessWithoutFailingRun({ flush: true });
// This can sometimes be called before the handleCompletionResult, so we don't need to do anything here
return;
}
case "QUEUED_EXECUTING":
@@ -307,7 +304,7 @@ export class RunExecution {
// This will kill the process and fail the execution with a SuspendedProcessError
// We don't flush because we already did before suspending
await this.exitTaskRunProcessWithoutFailingRun({ flush: false });
await this.exitTaskRunProcessWithoutFailingRun({ flush: false, reason: "suspended" });
return;
}
case "PENDING_EXECUTING": {
@@ -377,7 +374,7 @@ export class RunExecution {
throw new Error("Cannot start attempt: missing run or snapshot manager");
}
this.sendDebugLog("starting attempt");
this.sendDebugLog("starting attempt", { isWarmStart: String(isWarmStart) });
const attemptStartedAt = Date.now();
@@ -422,7 +419,7 @@ export class RunExecution {
podScheduledAt: this.podScheduledAt?.getTime(),
});
this.sendDebugLog("started attempt");
this.sendDebugLog("started attempt", { start: start.data });
return { ...start.data, metrics };
}
@@ -478,21 +475,23 @@ export class RunExecution {
if (startError) {
this.sendDebugLog("failed to start attempt", { error: startError.message });
this.shutdown("failed to start attempt");
this.shutdownExecution("failed to start attempt");
return;
}
const [executeError] = await tryCatch(this.executeRunWrapper(start));
const [executeError] = await tryCatch(
this.executeRunWrapper({ ...start, isWarmStart: runOpts.isWarmStart })
);
if (executeError) {
this.sendDebugLog("failed to execute run", { error: executeError.message });
this.shutdown("failed to execute run");
this.shutdownExecution("failed to execute run");
return;
}
// This is here for safety, but it
this.shutdown("execute call finished");
this.shutdownExecution("execute call finished");
}
private async executeRunWrapper({
@@ -502,9 +501,11 @@ export class RunExecution {
execution,
metrics,
isWarmStart,
isImmediateRetry,
}: WorkloadRunAttemptStartResponseBody & {
metrics: TaskRunExecutionMetrics;
isWarmStart?: boolean;
isImmediateRetry?: boolean;
}) {
this.currentTaskRunEnv = envVars;
@@ -516,6 +517,7 @@ export class RunExecution {
execution,
metrics,
isWarmStart,
isImmediateRetry,
})
);
@@ -570,38 +572,39 @@ export class RunExecution {
execution,
metrics,
isWarmStart,
isImmediateRetry,
}: WorkloadRunAttemptStartResponseBody & {
metrics: TaskRunExecutionMetrics;
isWarmStart?: boolean;
isImmediateRetry?: boolean;
}) {
// For immediate retries, we need to ensure the task run process is prepared for the next attempt
if (
this.runFriendlyId &&
this.taskRunProcess &&
!this.taskRunProcess.isPreparedForNextAttempt
) {
this.sendDebugLog("killing existing task run process before executing next attempt");
await this.kill({ exitExecution: false }).catch(() => {});
if (isImmediateRetry) {
await this.taskRunProcessProvider.handleImmediateRetry();
}
// To skip this step and eagerly create the task run process, run prepareForExecution first
if (!this.taskRunProcess || !this.taskRunProcess.isPreparedForNextRun) {
this.taskRunProcess = this.createTaskRunProcess({
envVars: { ...envVars, TRIGGER_PROJECT_REF: execution.project.ref },
isWarmStart,
});
}
const taskRunEnv = this.currentTaskRunEnv ?? envVars;
this.taskRunProcess = await this.taskRunProcessProvider.getProcess({
taskRunEnv: { ...taskRunEnv, TRIGGER_PROJECT_REF: execution.project.ref },
isWarmStart,
});
this.attachTaskRunProcessHandlers(this.taskRunProcess);
this.sendDebugLog("executing task run process", { runId: execution.run.id });
// Set up an abort handler that will cleanup the task run process
this.executionAbortController.signal.addEventListener("abort", async () => {
const abortHandler = async () => {
this.sendDebugLog("execution aborted during task run, cleaning up process", {
runId: execution.run.id,
});
await this.taskRunProcess?.cleanup(true);
});
if (this.taskRunProcess) {
await this.taskRunProcessProvider.handleProcessAbort(this.taskRunProcess);
}
};
// Set up an abort handler that will cleanup the task run process
this.executionAbortController.signal.addEventListener("abort", abortHandler);
const completion = await this.taskRunProcess.execute(
{
@@ -616,15 +619,19 @@ export class RunExecution {
isWarmStart
);
this.executionAbortController.signal.removeEventListener("abort", abortHandler);
// If we get here, the task completed normally
this.sendDebugLog("completed run attempt", { attemptSuccess: completion.ok });
// The execution has finished, so we can cleanup the task run process. Killing it should be safe.
const [error] = await tryCatch(this.taskRunProcess.cleanup(true));
// Return the process to the provider - this handles all cleanup logic
const [returnError] = await tryCatch(
this.taskRunProcessProvider.returnProcess(this.taskRunProcess)
);
if (error) {
this.sendDebugLog("failed to cleanup task run process, submitting completion anyway", {
error: error.message,
if (returnError) {
this.sendDebugLog("failed to return task run process, submitting completion anyway", {
error: returnError.message,
});
}
@@ -785,16 +792,18 @@ export class RunExecution {
if (startError) {
this.sendDebugLog("failed to start attempt for retry", { error: startError.message });
this.shutdown("retryImmediately: failed to start attempt");
this.shutdownExecution("retryImmediately: failed to start attempt");
return;
}
const [executeError] = await tryCatch(this.executeRunWrapper({ ...start, isWarmStart: true }));
const [executeError] = await tryCatch(
this.executeRunWrapper({ ...start, isWarmStart: true, isImmediateRetry: true })
);
if (executeError) {
this.sendDebugLog("failed to execute run for retry", { error: executeError.message });
this.shutdown("retryImmediately: failed to execute run");
this.shutdownExecution("retryImmediately: failed to execute run");
return;
}
}
@@ -828,11 +837,17 @@ export class RunExecution {
this.restoreCount++;
}
private async exitTaskRunProcessWithoutFailingRun({ flush }: { flush: boolean }) {
await this.taskRunProcess?.suspend({ flush });
private async exitTaskRunProcessWithoutFailingRun({
flush,
reason,
}: {
flush: boolean;
reason: string;
}) {
await this.taskRunProcessProvider.suspendProcess(flush, this.taskRunProcess);
// No services should be left running after this line - let's make sure of it
this.shutdown("exitTaskRunProcessWithoutFailingRun");
this.shutdownExecution(`exitTaskRunProcessWithoutFailingRun: ${reason}`);
}
/**
@@ -1003,10 +1018,10 @@ export class RunExecution {
}
this.executionAbortController.abort();
this.shutdown("abortExecution");
this.shutdownExecution("abortExecution");
}
private shutdown(reason: string) {
private shutdownExecution(reason: string) {
if (this.isShuttingDown) {
this.sendDebugLog(`[shutdown] ${reason} (already shutting down)`, {
firstShutdownReason: this.shutdownReason,
packages/cli-v3/src/entryPoints/managed/taskRunProcessProvider.ts
+317
View File
@@ -0,0 +1,317 @@
import { WorkerManifest } from "@trigger.dev/core/v3";
import { TaskRunProcess } from "../../executions/taskRunProcess.js";
import { RunnerEnv } from "./env.js";
import { RunLogger, SendDebugLogOptions } from "./logger.js";
export interface TaskRunProcessProviderOptions {
workerManifest: WorkerManifest;
env: RunnerEnv;
logger: RunLogger;
processKeepAliveEnabled: boolean;
processKeepAliveMaxExecutionCount: number;
}
export interface GetProcessOptions {
taskRunEnv: Record<string, string>;
isWarmStart?: boolean;
}
export class TaskRunProcessProvider {
private readonly workerManifest: WorkerManifest;
private readonly env: RunnerEnv;
private readonly logger: RunLogger;
private readonly processKeepAliveEnabled: boolean;
private readonly processKeepAliveMaxExecutionCount: number;
// Process keep-alive state
private persistentProcess: TaskRunProcess | null = null;
private executionCount = 0;
constructor(opts: TaskRunProcessProviderOptions) {
this.workerManifest = opts.workerManifest;
this.env = opts.env;
this.logger = opts.logger;
this.processKeepAliveEnabled = opts.processKeepAliveEnabled;
this.processKeepAliveMaxExecutionCount = opts.processKeepAliveMaxExecutionCount;
}
get hasPersistentProcess(): boolean {
return !!this.persistentProcess;
}
async handleImmediateRetry(): Promise<void> {
if (!this.processKeepAliveEnabled) {
// For immediate retries, we need to ensure we have a clean process
if (this.persistentProcess) {
// If the process is not prepared for the next attempt, we need to get a fresh one
if (!this.persistentProcess.isPreparedForNextAttempt) {
this.sendDebugLog(
"existing task run process not prepared for retry, will get fresh process"
);
await this.persistentProcess.kill("SIGKILL");
this.persistentProcess = null;
}
}
}
}
/**
* Gets a TaskRunProcess, either by reusing an existing one or creating a new one
*/
async getProcess(opts: GetProcessOptions): Promise<TaskRunProcess> {
this.sendDebugLog("Getting TaskRunProcess", {
processKeepAliveEnabled: this.processKeepAliveEnabled,
hasPersistentProcess: !!this.persistentProcess,
executionCount: this.executionCount,
maxExecutionCount: this.processKeepAliveMaxExecutionCount,
isWarmStart: opts.isWarmStart,
});
// If process keep-alive is disabled, always create a new process
if (!this.processKeepAliveEnabled) {
this.sendDebugLog("Creating new TaskRunProcess (keep-alive disabled)");
return this.createTaskRunProcess(opts);
}
// If process keep-alive is enabled and we have a healthy persistent process, reuse it
if (this.shouldReusePersistentProcess()) {
this.sendDebugLog("Reusing persistent TaskRunProcess", {
executionCount: this.executionCount,
});
return this.persistentProcess!;
}
// Create new process (keep-alive enabled but no reusable process available)
this.sendDebugLog("Creating new TaskRunProcess", {
hadPersistentProcess: !!this.persistentProcess,
reason: this.processKeepAliveEnabled
? "execution limit reached or unhealthy"
: "keep-alive disabled",
});
const existingPersistentProcess = this.persistentProcess;
// Clean up old persistent process if it exists
if (existingPersistentProcess) {
await this.cleanupProcess(existingPersistentProcess);
}
this.persistentProcess = this.createTaskRunProcess(opts);
this.executionCount = 0;
return this.persistentProcess;
}
/**
* Returns a process after execution, handling keep-alive logic and cleanup
*/
async returnProcess(process: TaskRunProcess): Promise<void> {
this.sendDebugLog("Returning TaskRunProcess", {
processKeepAliveEnabled: this.processKeepAliveEnabled,
executionCount: this.executionCount,
maxExecutionCount: this.processKeepAliveMaxExecutionCount,
});
if (!this.processKeepAliveEnabled) {
// Keep-alive disabled - immediately cleanup the process
this.sendDebugLog("Keep-alive disabled, cleaning up process immediately");
await process.cleanup(true);
return;
}
// Keep-alive enabled - check if we should keep the process alive
if (this.shouldKeepProcessAlive(process)) {
this.sendDebugLog("Keeping TaskRunProcess alive for next run", {
executionCount: this.executionCount,
maxExecutionCount: this.processKeepAliveMaxExecutionCount,
});
// Call cleanup(false) to prepare for next run but keep process alive
await process.cleanup(false);
this.persistentProcess = process;
this.executionCount++;
} else {
this.sendDebugLog("Not keeping TaskRunProcess alive, cleaning up", {
executionCount: this.executionCount,
maxExecutionCount: this.processKeepAliveMaxExecutionCount,
isHealthy: this.isProcessHealthy(process),
});
// Cleanup the process completely
await process.cleanup(true);
}
}
async suspendProcess(flush: boolean, process?: TaskRunProcess): Promise<void> {
if (this.persistentProcess) {
if (process) {
if (this.persistentProcess.pid === process.pid) {
this.sendDebugLog("Suspending matching persistent TaskRunProcess (process provided)", {
pid: process.pid,
flush,
});
this.persistentProcess = null;
this.executionCount = 0;
await process.suspend({ flush });
} else {
this.sendDebugLog("Suspending TaskRunProcess (does not match persistent process)", {
pid: process.pid,
flush,
});
await process.suspend({ flush });
}
} else {
this.sendDebugLog("Suspending persistent TaskRunProcess (no process provided)", {
pid: this.persistentProcess.pid,
flush,
});
this.persistentProcess = null;
this.executionCount = 0;
}
} else {
if (process) {
this.sendDebugLog("Suspending non-persistent TaskRunProcess (process provided)", {
pid: process.pid,
flush,
});
await process.suspend({ flush });
} else {
this.sendDebugLog("Suspending non-persistent TaskRunProcess (no process provided)", {
flush,
});
}
}
}
/**
* Handles process abort/kill scenarios
*/
async handleProcessAbort(process: TaskRunProcess): Promise<void> {
this.sendDebugLog("Handling process abort");
// If this was our persistent process, clear it
if (this.persistentProcess?.pid === process.pid) {
this.persistentProcess = null;
this.executionCount = 0;
}
// Kill the process
await process.cleanup(true);
}
async killProcess(process: TaskRunProcess): Promise<void> {
this.sendDebugLog("Killing process");
// If this was our persistent process, clear it
if (this.persistentProcess?.pid === process.pid) {
this.persistentProcess = null;
this.executionCount = 0;
}
// Kill the process
await this.cleanupProcess(process);
}
/**
* Forces cleanup of any persistent process
*/
async cleanup() {
if (this.persistentProcess) {
this.sendDebugLog("cleanup() called");
await this.cleanupProcess(this.persistentProcess);
}
}
/**
* Gets metrics about the provider state
*/
get metrics() {
return {
processKeepAlive: {
enabled: this.processKeepAliveEnabled,
executionCount: this.executionCount,
maxExecutionCount: this.processKeepAliveMaxExecutionCount,
hasPersistentProcess: !!this.persistentProcess,
},
};
}
private createTaskRunProcess({ taskRunEnv, isWarmStart }: GetProcessOptions): TaskRunProcess {
const processEnv = this.buildProcessEnvironment(taskRunEnv);
const taskRunProcess = new TaskRunProcess({
workerManifest: this.workerManifest,
env: processEnv,
serverWorker: {
id: "managed",
contentHash: this.env.TRIGGER_CONTENT_HASH,
version: this.env.TRIGGER_DEPLOYMENT_VERSION,
engine: "V2",
},
machineResources: {
cpu: Number(this.env.TRIGGER_MACHINE_CPU),
memory: Number(this.env.TRIGGER_MACHINE_MEMORY),
},
isWarmStart,
}).initialize();
return taskRunProcess;
}
private buildProcessEnvironment(taskRunEnv: Record<string, string>): Record<string, string> {
return {
...taskRunEnv,
...this.env.gatherProcessEnv(),
HEARTBEAT_INTERVAL_MS: String(this.env.TRIGGER_HEARTBEAT_INTERVAL_SECONDS * 1000),
};
}
private shouldReusePersistentProcess(): boolean {
this.sendDebugLog("Checking if persistent process should be reused", {
executionCount: this.executionCount,
maxExecutionCount: this.processKeepAliveMaxExecutionCount,
pid: this.persistentProcess?.pid ?? "unknown",
isBeingKilled: this.persistentProcess?.isBeingKilled ?? "unknown",
});
return (
!!this.persistentProcess &&
this.executionCount < this.processKeepAliveMaxExecutionCount &&
this.isProcessHealthy(this.persistentProcess)
);
}
private shouldKeepProcessAlive(process: TaskRunProcess): boolean {
return (
this.executionCount < this.processKeepAliveMaxExecutionCount && this.isProcessHealthy(process)
);
}
private isProcessHealthy(process: TaskRunProcess): boolean {
// Basic health check - TaskRunProcess will handle more detailed internal health checks
return !process.isBeingKilled && process.pid !== undefined;
}
private async cleanupProcess(taskRunProcess: TaskRunProcess): Promise<void> {
if (taskRunProcess && taskRunProcess.pid !== undefined) {
this.sendDebugLog("Cleaning up TaskRunProcess", { pid: taskRunProcess.pid });
await taskRunProcess.kill("SIGKILL").catch(() => {});
}
}
private sendDebugLog(message: string, properties?: SendDebugLogOptions["properties"]): void {
this.logger.sendDebugLog({
runId: undefined, // Provider doesn't have access to current run ID
message: `[taskRunProcessProvider] ${message}`,
properties,
});
}
}
packages/cli-v3/src/executions/taskRunProcess.ts
+12 -5
View File
@@ -78,7 +78,6 @@ export class TaskRunProcess {
public onTaskRunHeartbeat: Evt<string> = new Evt();
public onExit: Evt<{ code: number | null; signal: NodeJS.Signals | null; pid?: number }> =
new Evt();
public onIsBeingKilled: Evt<TaskRunProcess> = new Evt();
public onSendDebugLog: Evt<OnSendDebugLogMessage> = new Evt();
public onSetSuspendable: Evt<OnSetSuspendableMessage> = new Evt();
@@ -100,7 +99,6 @@ export class TaskRunProcess {
unsafeDetachEvtHandlers() {
this.onExit.detach();
this.onIsBeingKilled.detach();
this.onSendDebugLog.detach();
this.onSetSuspendable.detach();
this.onTaskRunHeartbeat.detach();
@@ -150,6 +148,7 @@ export class TaskRunProcess {
PATH: process.env.PATH,
TRIGGER_PROCESS_FORK_START_TIME: String(Date.now()),
TRIGGER_WARM_START: this.options.isWarmStart ? "true" : "false",
TRIGGERDOTDEV: "1",
};
logger.debug(`initializing task run process`, {
@@ -169,6 +168,12 @@ export class TaskRunProcess {
this._childPid = this._child?.pid;
logger.debug("initialized task run process", {
path: workerManifest.workerEntryPoint,
cwd,
pid: this._childPid,
});
this._ipc = new ZodIpcConnection({
listenSchema: ExecutorToWorkerMessageCatalog,
emitSchema: WorkerToExecutorMessageCatalog,
@@ -286,6 +291,10 @@ export class TaskRunProcess {
return result;
}
isExecuting() {
return this._currentExecution !== undefined;
}
waitpointCompleted(waitpoint: CompletedWaitpoint) {
if (!this._child?.connected || this._isBeingKilled || this._child.killed) {
console.error(
@@ -298,7 +307,7 @@ export class TaskRunProcess {
}
async #handleExit(code: number | null, signal: NodeJS.Signals | null) {
logger.debug("handling child exit", { code, signal });
logger.debug("handling child exit", { code, signal, pid: this.pid });
// Go through all the attempts currently pending and reject them
for (const [id, status] of this._attemptStatuses.entries()) {
@@ -398,8 +407,6 @@ export class TaskRunProcess {
const killTimeout = this.onExit.waitFor(timeoutInMs);
this.onIsBeingKilled.post(this);
try {
this._child?.kill(signal);
} catch (error) {
packages/core/src/v3/config.ts
+25
View File
@@ -234,6 +234,31 @@ export type TriggerConfig = {
env?: Record<string, string>;
};
/**
* @default false
* @description Keep the process alive after the task has finished running so the next task doesn't have to wait for the process to start up again.
*
* Note that the process could be killed at any time, and we don't make any guarantees about the process being alive for a certain amount of time
*/
experimental_processKeepAlive?:
| boolean
| {
enabled: boolean;
/**
* The maximum number of executions per process. If the process has run more than this number of times, it will be killed.
*
* @default 50
*/
maxExecutionsPerProcess?: number;
/**
* The maximum number of processes to keep alive in dev.
*
* @default 25
*/
devMaxPoolSize?: number;
};
/**
* @deprecated Use `dirs` instead
*/
packages/core/src/v3/lifecycleHooks/manager.ts
+6
View File
@@ -68,6 +68,12 @@ export class StandardLifecycleHooksManager implements LifecycleHooksManager {
private taskCancelHooks: Map<string, RegisteredHookFunction<AnyOnCancelHookFunction>> = new Map();
private onCancelHookListeners: (() => Promise<void>)[] = [];
reset(): void {
this.onCancelHookListeners.length = 0;
this.onWaitHookListeners.length = 0;
this.onResumeHookListeners.length = 0;
}
registerOnCancelHookListener(listener: () => Promise<void>): void {
this.onCancelHookListeners.push(listener);
}
packages/core/src/v3/limits.ts
+41 -8
View File
@@ -1,13 +1,46 @@
import { AttributeValue, Attributes } from "@opentelemetry/api";
import { getEnvVar } from "./utils/getEnv.js";
function getOtelEnvVarLimit(key: string, defaultValue: number) {
const value = getEnvVar(key);
if (!value) {
return defaultValue;
}
return parseInt(value, 10);
}
export const OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT = getOtelEnvVarLimit(
"TRIGGER_OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT",
256
);
export const OTEL_LOG_ATTRIBUTE_COUNT_LIMIT = getOtelEnvVarLimit(
"TRIGGER_OTEL_LOG_ATTRIBUTE_COUNT_LIMIT",
256
);
export const OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT = getOtelEnvVarLimit(
"TRIGGER_OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT",
131072
);
export const OTEL_LOG_ATTRIBUTE_VALUE_LENGTH_LIMIT = getOtelEnvVarLimit(
"TRIGGER_OTEL_LOG_ATTRIBUTE_VALUE_LENGTH_LIMIT",
131072
);
export const OTEL_SPAN_EVENT_COUNT_LIMIT = getOtelEnvVarLimit(
"TRIGGER_OTEL_SPAN_EVENT_COUNT_LIMIT",
10
);
export const OTEL_LINK_COUNT_LIMIT = getOtelEnvVarLimit("TRIGGER_OTEL_LINK_COUNT_LIMIT", 2);
export const OTEL_ATTRIBUTE_PER_LINK_COUNT_LIMIT = getOtelEnvVarLimit(
"TRIGGER_OTEL_ATTRIBUTE_PER_LINK_COUNT_LIMIT",
10
);
export const OTEL_ATTRIBUTE_PER_EVENT_COUNT_LIMIT = getOtelEnvVarLimit(
"TRIGGER_OTEL_ATTRIBUTE_PER_EVENT_COUNT_LIMIT",
10
);
export const OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT = 256;
export const OTEL_LOG_ATTRIBUTE_COUNT_LIMIT = 256;
export const OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT = 1028;
export const OTEL_LOG_ATTRIBUTE_VALUE_LENGTH_LIMIT = 1028;
export const OTEL_SPAN_EVENT_COUNT_LIMIT = 10;
export const OTEL_LINK_COUNT_LIMIT = 2;
export const OTEL_ATTRIBUTE_PER_LINK_COUNT_LIMIT = 10;
export const OTEL_ATTRIBUTE_PER_EVENT_COUNT_LIMIT = 10;
export const OFFLOAD_IO_PACKET_LENGTH_LIMIT = 128 * 1024;
export function imposeAttributeLimits(attributes: Attributes): Attributes {
packages/core/src/v3/locals/manager.ts
+4 -1
View File
@@ -33,5 +33,8 @@ export class StandardLocalsManager implements LocalsManager {
setLocal<T>(key: LocalsKey<T>, value: T): void {
this.store.set(key.__type, value);
}
reset(): void {
this.store.clear();
}
}
0;
packages/core/src/v3/otel/tracingSDK.ts
+6
View File
@@ -57,6 +57,10 @@ class AsyncResourceDetector implements DetectorSync {
});
}
get isResolved() {
return this._resolved;
}
detect(_config?: ResourceDetectionConfig): Resource {
return new Resource({}, this._promise);
}
@@ -123,6 +127,8 @@ export class TracingSDK {
getEnvVar("OTEL_SERVICE_NAME") ?? "trigger.dev",
[SemanticInternalAttributes.TRIGGER]: true,
[SemanticInternalAttributes.CLI_VERSION]: VERSION,
[SemanticInternalAttributes.SDK_VERSION]: VERSION,
[SemanticInternalAttributes.SDK_LANGUAGE]: "typescript",
})
)
.merge(config.resource ?? new Resource({}))
packages/core/src/v3/runMetadata/manager.ts
+24 -3
View File
@@ -7,6 +7,7 @@ import { MetadataStream } from "./metadataStream.js";
import { applyMetadataOperations, collapseOperations } from "./operations.js";
import { RunMetadataManager, RunMetadataUpdater } from "./types.js";
import { AsyncIterableStream } from "../streams/asyncIterableStream.js";
import { IOPacket, stringifyIO } from "../utils/ioSerialization.js";
const MAXIMUM_ACTIVE_STREAMS = 5;
const MAXIMUM_TOTAL_STREAMS = 10;
@@ -30,6 +31,22 @@ export class StandardMetadataManager implements RunMetadataManager {
private streamsVersion: "v1" | "v2" = "v1"
) {}
reset(): void {
this.queuedOperations.clear();
this.queuedParentOperations.clear();
this.queuedRootOperations.clear();
this.activeStreams.clear();
this.store = undefined;
this.runId = undefined;
if (this.flushTimeoutId) {
clearTimeout(this.flushTimeoutId);
this.flushTimeoutId = null;
}
this.isFlushing = false;
}
get parent(): RunMetadataUpdater {
// Store a reference to 'this' to ensure proper context
const self = this;
@@ -406,23 +423,27 @@ export class StandardMetadataManager implements RunMetadataManager {
}
}
stopAndReturnLastFlush(): FlushedRunMetadata | undefined {
async stopAndReturnLastFlush(): Promise<IOPacket> {
this.stopPeriodicFlush();
this.isFlushing = true;
if (!this.#needsFlush()) {
return;
return { dataType: "application/json" };
}
const operations = Array.from(this.queuedOperations);
const parentOperations = Array.from(this.queuedParentOperations);
const rootOperations = Array.from(this.queuedRootOperations);
return {
const data = {
operations: collapseOperations(operations),
parentOperations: collapseOperations(parentOperations),
rootOperations: collapseOperations(rootOperations),
};
const packet = await stringifyIO(data);
return packet;
}
#needsFlush(): boolean {
packages/core/src/v3/runTimelineMetrics/runTimelineMetricsManager.ts
+13 -4
View File
@@ -13,8 +13,11 @@ export class StandardRunTimelineMetricsManager implements RunTimelineMetricsMana
return this._metrics;
}
registerMetricsFromExecution(metrics?: TaskRunExecutionMetrics): void {
this.#seedMetricsFromEnvironment();
registerMetricsFromExecution(
metrics?: TaskRunExecutionMetrics,
isWarmStartOverride?: boolean
): void {
this.#seedMetricsFromEnvironment(isWarmStartOverride);
if (metrics) {
metrics.forEach((metric) => {
@@ -30,10 +33,16 @@ export class StandardRunTimelineMetricsManager implements RunTimelineMetricsMana
}
}
#seedMetricsFromEnvironment() {
reset(): void {
this._metrics = [];
}
// TODO: handle this when processKeepAlive is enabled
#seedMetricsFromEnvironment(isWarmStartOverride?: boolean) {
const forkStartTime = getEnvVar("TRIGGER_PROCESS_FORK_START_TIME");
const warmStart = getEnvVar("TRIGGER_WARM_START");
const isWarmStart = warmStart === "true";
const isWarmStart =
typeof isWarmStartOverride === "boolean" ? isWarmStartOverride : warmStart === "true";
if (typeof forkStartTime === "string" && !isWarmStart) {
const forkStartTimeMs = parseInt(forkStartTime, 10);
packages/core/src/v3/runtime/sharedRuntimeManager.ts
+5
View File
@@ -41,6 +41,11 @@ export class SharedRuntimeManager implements RuntimeManager {
}, 300_000);
}
reset(): void {
this.resolversById.clear();
this.waitpointsByResolverId.clear();
}
disable(): void {
// do nothing
}
packages/core/src/v3/schemas/build.ts
+7
View File
@@ -92,6 +92,13 @@ export const WorkerManifest = z.object({
runtime: BuildRuntime,
customConditions: z.array(z.string()).optional(),
timings: z.record(z.number()).optional(),
processKeepAlive: z
.object({
enabled: z.boolean(),
maxExecutionsPerProcess: z.number().int().positive().optional(),
})
.optional(),
otelImportHook: z
.object({
include: z.array(z.string()).optional(),
packages/core/src/v3/schemas/common.ts
+16
View File
@@ -376,7 +376,15 @@ export const TaskRunFailedExecutionResult = z.object({
usage: TaskRunExecutionUsage.optional(),
// Optional for now for backwards compatibility
taskIdentifier: z.string().optional(),
// This is deprecated, use flushedMetadata instead
metadata: FlushedRunMetadata.optional(),
// This is the new way to flush metadata
flushedMetadata: z
.object({
data: z.string().optional(),
dataType: z.string(),
})
.optional(),
});
export type TaskRunFailedExecutionResult = z.infer<typeof TaskRunFailedExecutionResult>;
@@ -389,7 +397,15 @@ export const TaskRunSuccessfulExecutionResult = z.object({
usage: TaskRunExecutionUsage.optional(),
// Optional for now for backwards compatibility
taskIdentifier: z.string().optional(),
// This is deprecated, use flushedMetadata instead
metadata: FlushedRunMetadata.optional(),
// This is the new way to flush metadata
flushedMetadata: z
.object({
data: z.string().optional(),
dataType: z.string(),
})
.optional(),
});
export type TaskRunSuccessfulExecutionResult = z.infer<typeof TaskRunSuccessfulExecutionResult>;
packages/core/src/v3/schemas/schemas.ts
+1 -1
View File
@@ -163,7 +163,7 @@ export const QueueManifest = z.object({
/** An optional property that specifies the maximum number of concurrent run executions.
*
* If this property is omitted, the task can potentially use up the full concurrency of an environment */
concurrencyLimit: z.number().int().min(0).max(1000).optional().nullable(),
concurrencyLimit: z.number().int().min(0).max(100000).optional().nullable(),
/** An optional property that specifies whether to release concurrency on waitpoint.
*
* If this property is omitted, the task will not release concurrency on waitpoint.
packages/core/src/v3/semanticInternalAttributes.ts
+1
View File
@@ -60,4 +60,5 @@ export const SemanticInternalAttributes = {
METRIC_EVENTS: "$metrics.events",
EXECUTION_ENVIRONMENT: "exec_env",
WARM_START: "warm_start",
ATTEMPT_EXECUTION_COUNT: "$trigger.executionCount",
};
packages/core/src/v3/taskContext/index.ts
+21 -12
View File
@@ -47,6 +47,27 @@ export class TaskContextAPI {
return {};
}
get resourceAttributes(): Attributes {
if (this.ctx) {
return {
[SemanticInternalAttributes.ENVIRONMENT_ID]: this.ctx.environment.id,
[SemanticInternalAttributes.ENVIRONMENT_TYPE]: this.ctx.environment.type,
[SemanticInternalAttributes.ORGANIZATION_ID]: this.ctx.organization.id,
[SemanticInternalAttributes.PROJECT_ID]: this.ctx.project.id,
[SemanticInternalAttributes.PROJECT_REF]: this.ctx.project.ref,
[SemanticInternalAttributes.PROJECT_NAME]: this.ctx.project.name,
[SemanticInternalAttributes.ORGANIZATION_SLUG]: this.ctx.organization.slug,
[SemanticInternalAttributes.ORGANIZATION_NAME]: this.ctx.organization.name,
[SemanticInternalAttributes.MACHINE_PRESET_NAME]: this.ctx.machine?.name,
[SemanticInternalAttributes.MACHINE_PRESET_CPU]: this.ctx.machine?.cpu,
[SemanticInternalAttributes.MACHINE_PRESET_MEMORY]: this.ctx.machine?.memory,
[SemanticInternalAttributes.MACHINE_PRESET_CENTS_PER_MS]: this.ctx.machine?.centsPerMs,
};
}
return {};
}
get workerAttributes(): Attributes {
if (this.worker) {
return {
@@ -68,22 +89,10 @@ export class TaskContextAPI {
[SemanticInternalAttributes.TASK_EXPORT_NAME]: this.ctx.task.exportName,
[SemanticInternalAttributes.QUEUE_NAME]: this.ctx.queue.name,
[SemanticInternalAttributes.QUEUE_ID]: this.ctx.queue.id,
[SemanticInternalAttributes.ENVIRONMENT_ID]: this.ctx.environment.id,
[SemanticInternalAttributes.ENVIRONMENT_TYPE]: this.ctx.environment.type,
[SemanticInternalAttributes.ORGANIZATION_ID]: this.ctx.organization.id,
[SemanticInternalAttributes.PROJECT_ID]: this.ctx.project.id,
[SemanticInternalAttributes.PROJECT_REF]: this.ctx.project.ref,
[SemanticInternalAttributes.PROJECT_NAME]: this.ctx.project.name,
[SemanticInternalAttributes.RUN_ID]: this.ctx.run.id,
[SemanticInternalAttributes.RUN_IS_TEST]: this.ctx.run.isTest,
[SemanticInternalAttributes.ORGANIZATION_SLUG]: this.ctx.organization.slug,
[SemanticInternalAttributes.ORGANIZATION_NAME]: this.ctx.organization.name,
[SemanticInternalAttributes.BATCH_ID]: this.ctx.batch?.id,
[SemanticInternalAttributes.IDEMPOTENCY_KEY]: this.ctx.run.idempotencyKey,
[SemanticInternalAttributes.MACHINE_PRESET_NAME]: this.ctx.machine?.name,
[SemanticInternalAttributes.MACHINE_PRESET_CPU]: this.ctx.machine?.cpu,
[SemanticInternalAttributes.MACHINE_PRESET_MEMORY]: this.ctx.machine?.memory,
[SemanticInternalAttributes.MACHINE_PRESET_CENTS_PER_MS]: this.ctx.machine?.centsPerMs,
};
}
packages/core/src/v3/taskContext/otelProcessors.ts
+3 -21
View File
@@ -19,13 +19,7 @@ export class TaskContextSpanProcessor implements SpanProcessor {
onStart(span: Span, parentContext: Context): void {
if (taskContext.ctx) {
span.setAttributes(
flattenAttributes(
{
[SemanticInternalAttributes.ATTEMPT_ID]: taskContext.ctx.attempt.id,
[SemanticInternalAttributes.ATTEMPT_NUMBER]: taskContext.ctx.attempt.number,
},
SemanticInternalAttributes.METADATA
)
flattenAttributes(taskContext.attributes, SemanticInternalAttributes.METADATA)
);
}
@@ -75,13 +69,7 @@ function createPartialSpan(tracer: Tracer, span: Span, parentContext: Context) {
if (taskContext.ctx) {
partialSpan.setAttributes(
flattenAttributes(
{
[SemanticInternalAttributes.ATTEMPT_ID]: taskContext.ctx.attempt.id,
[SemanticInternalAttributes.ATTEMPT_NUMBER]: taskContext.ctx.attempt.number,
},
SemanticInternalAttributes.METADATA
)
flattenAttributes(taskContext.attributes, SemanticInternalAttributes.METADATA)
);
}
@@ -107,13 +95,7 @@ export class TaskContextLogProcessor implements LogRecordProcessor {
// Adds in the context attributes to the log record
if (taskContext.ctx) {
logRecord.setAttributes(
flattenAttributes(
{
[SemanticInternalAttributes.ATTEMPT_ID]: taskContext.ctx.attempt.id,
[SemanticInternalAttributes.ATTEMPT_NUMBER]: taskContext.ctx.attempt.number,
},
SemanticInternalAttributes.METADATA
)
flattenAttributes(taskContext.attributes, SemanticInternalAttributes.METADATA)
);
}
packages/core/src/v3/timeout/api.ts
+7
View File
@@ -7,6 +7,8 @@ class NoopTimeoutManager implements TimeoutManager {
abortAfterTimeout(timeoutInSeconds?: number): AbortController {
return new AbortController();
}
reset() {}
}
const NOOP_TIMEOUT_MANAGER = new NoopTimeoutManager();
@@ -40,6 +42,11 @@ export class TimeoutAPI implements TimeoutManager {
unregisterGlobal(API_NAME);
}
public reset() {
this.#getManager().reset();
this.disable();
}
#getManager(): TimeoutManager {
return getGlobal(API_NAME) ?? NOOP_TIMEOUT_MANAGER;
}
packages/core/src/v3/timeout/types.ts
+1
View File
@@ -1,6 +1,7 @@
export interface TimeoutManager {
abortAfterTimeout: (timeoutInSeconds?: number) => AbortController;
signal?: AbortSignal;
reset: () => void;
}
export class TaskRunExceededMaxDuration extends Error {
packages/core/src/v3/timeout/usageTimeoutManager.ts
+10
View File
@@ -14,6 +14,16 @@ export class UsageTimeoutManager implements TimeoutManager {
return this._abortSignal;
}
reset(): void {
this._abortController = new AbortController();
this._abortSignal = undefined;
if (this._intervalId) {
clearInterval(this._intervalId);
this._intervalId = undefined;
}
}
abortAfterTimeout(timeoutInSeconds?: number): AbortController {
this._abortSignal = this._abortController.signal;
packages/core/src/v3/usage/api.ts
+5
View File
@@ -48,6 +48,11 @@ export class UsageAPI implements UsageManager {
return this.#getUsageManager().flush();
}
public reset() {
this.#getUsageManager().reset();
this.disable();
}
#getUsageManager(): UsageManager {
return getGlobal(API_NAME) ?? NOOP_USAGE_MANAGER;
}
packages/core/src/v3/usage/devUsageManager.ts
+6
View File
@@ -50,6 +50,12 @@ export class DevUsageManager implements UsageManager {
async flush(): Promise<void> {}
reset(): void {
this._firstMeasurement = undefined;
this._currentMeasurements.clear();
this._pauses.clear();
}
sample(): UsageSample | undefined {
return this._firstMeasurement?.sample();
}
packages/core/src/v3/usage/noopUsageManager.ts
+4
View File
@@ -26,4 +26,8 @@ export class NoopUsageManager implements UsageManager {
sample(): UsageSample | undefined {
return undefined;
}
reset(): void {
// Noop
}
}
packages/core/src/v3/usage/prodUsageManager.ts
+19 -4
View File
@@ -27,6 +27,15 @@ export class ProdUsageManager implements UsageManager {
return typeof this._usageClient !== "undefined";
}
reset(): void {
this.delegageUsageManager.reset();
this._abortController?.abort();
this._abortController = new AbortController();
this._usageClient = undefined;
this._measurement = undefined;
this._lastSample = undefined;
}
disable(): void {
this.delegageUsageManager.disable();
this._abortController?.abort();
@@ -67,12 +76,18 @@ export class ProdUsageManager implements UsageManager {
this._abortController = new AbortController();
for await (const _ of setInterval(this.options.heartbeatIntervalMs)) {
if (this._abortController.signal.aborted) {
break;
try {
for await (const _ of setInterval(this.options.heartbeatIntervalMs, undefined, {
signal: this._abortController.signal,
})) {
await this.#reportUsage();
}
} catch (error) {
if (error instanceof Error && error.name === "AbortError") {
return;
}
await this.#reportUsage();
throw error;
}
}
packages/core/src/v3/usage/types.ts
+1
View File
@@ -14,4 +14,5 @@ export interface UsageManager {
sample(): UsageSample | undefined;
pauseAsync<T>(cb: () => Promise<T>): Promise<T>;
flush(): Promise<void>;
reset(): void;
}
packages/core/src/v3/waitUntil/manager.ts
+4
View File
@@ -3,6 +3,10 @@ import { MaybeDeferredPromise, WaitUntilManager } from "./types.js";
export class StandardWaitUntilManager implements WaitUntilManager {
private maybeDeferredPromises: Set<MaybeDeferredPromise> = new Set();
reset(): void {
this.maybeDeferredPromises.clear();
}
register(promise: MaybeDeferredPromise): void {
this.maybeDeferredPromises.add(promise);
}
packages/core/src/v3/workers/populateEnv.ts
+16 -1
View File
@@ -13,6 +13,12 @@ interface PopulateEnvOptions {
* @default false
*/
debug?: boolean;
/**
* The previous environment variables
* @default undefined
*/
previousEnv?: Record<string, string>;
}
/**
@@ -25,7 +31,7 @@ export function populateEnv(
envObject: Record<string, string>,
options: PopulateEnvOptions = {}
): void {
const { override = false, debug = false } = options;
const { override = false, debug = false, previousEnv } = options;
if (!envObject || typeof envObject !== "object") {
return;
@@ -47,4 +53,13 @@ export function populateEnv(
process.env[key] = envObject[key];
}
}
if (previousEnv) {
// if there are any keys in previousEnv that are not in envObject, remove them from process.env
for (const key of Object.keys(previousEnv)) {
if (!Object.prototype.hasOwnProperty.call(envObject, key)) {
delete process.env[key];
}
}
}
}
packages/core/src/v3/workers/taskExecutor.ts
+13 -9
View File
@@ -62,6 +62,7 @@ export type TaskExecutorOptions = {
default?: RetryOptions;
};
isWarmStart?: boolean;
executionCount?: number;
};
export class TaskExecutor {
@@ -75,6 +76,7 @@ export class TaskExecutor {
}
| undefined;
private _isWarmStart: boolean | undefined;
private _executionCount: number | undefined;
constructor(
public task: TaskMetadataWithFunctions,
@@ -85,6 +87,7 @@ export class TaskExecutor {
this._consoleInterceptor = options.consoleInterceptor;
this._retries = options.retries;
this._isWarmStart = options.isWarmStart;
this._executionCount = options.executionCount;
}
async execute(
@@ -112,11 +115,11 @@ export class TaskExecutor {
runMetadata.enterWithMetadata(execution.run.metadata);
}
this._tracingSDK.asyncResourceDetector.resolveWithAttributes({
...taskContext.attributes,
[SemanticInternalAttributes.SDK_VERSION]: VERSION,
[SemanticInternalAttributes.SDK_LANGUAGE]: "typescript",
});
if (!this._tracingSDK.asyncResourceDetector.isResolved) {
this._tracingSDK.asyncResourceDetector.resolveWithAttributes({
...taskContext.resourceAttributes,
});
}
const result = await this._tracer.startActiveSpan(
attemptMessage,
@@ -351,11 +354,12 @@ export class TaskExecutor {
...(execution.attempt.number === 1
? runTimelineMetrics.convertMetricsToSpanAttributes()
: {}),
...(execution.environment.type !== "DEVELOPMENT"
[SemanticInternalAttributes.STYLE_VARIANT]: this._isWarmStart
? WARM_VARIANT
: COLD_VARIANT,
...(typeof this._executionCount === "number"
? {
[SemanticInternalAttributes.STYLE_VARIANT]: this._isWarmStart
? WARM_VARIANT
: COLD_VARIANT,
[SemanticInternalAttributes.ATTEMPT_EXECUTION_COUNT]: this._executionCount,
}
: {}),
},
references/hello-world/src/trigger/example.ts
+18
View File
@@ -4,8 +4,20 @@ import { ResourceMonitor } from "../resourceMonitor.js";
export const helloWorldTask = task({
id: "hello-world",
retry: {
maxAttempts: 3,
minTimeoutInMs: 500,
maxTimeoutInMs: 1000,
factor: 1.5,
},
onStart: async ({ payload, ctx, init }) => {
logger.info("Hello, world from the onStart hook", { payload, init });
},
run: async (payload: any, { ctx }) => {
logger.info("Hello, world from the init", { ctx, payload });
logger.info("env vars", {
env: process.env,
});
logger.debug("debug: Hello, world!", { payload });
logger.info("info: Hello, world!", { payload });
@@ -17,6 +29,12 @@ export const helloWorldTask = task({
logger.debug("some log", { span });
});
await setTimeout(payload.sleepFor ?? 180_000);
if (payload.throwError) {
throw new Error("Forced error to cause a retry");
}
logger.trace(
"my trace",
async (span) => {
references/hello-world/src/trigger/metadata.ts
+16
View File
@@ -0,0 +1,16 @@
import { metadata, task } from "@trigger.dev/sdk";
export const metadataTestTask = task({
id: "metadata-tester",
retry: {
maxAttempts: 3,
minTimeoutInMs: 500,
maxTimeoutInMs: 1000,
factor: 1.5,
},
run: async (payload: any, { ctx }) => {
metadata.set("test-key", "test-value");
metadata.append("test-keys", "test-value");
metadata.increment("test-counter", 1);
},
});
references/hello-world/src/trigger/schedule.ts
+1 -2
View File
@@ -2,8 +2,7 @@ import { schedules } from "@trigger.dev/sdk/v3";
export const simpleSchedule = schedules.task({
id: "simple-schedule",
// Every other minute
cron: "*/2 * * * *",
cron: "0 0 * * *",
run: async (payload, { ctx }) => {
return {
message: "Hello, world!",
references/hello-world/trigger.config.ts
+4
View File
@@ -4,6 +4,10 @@ import { syncEnvVars } from "@trigger.dev/build/extensions/core";
export default defineConfig({
compatibilityFlags: ["run_engine_v2"],
project: "proj_rrkpdguyagvsoktglnod",
experimental_processKeepAlive: {
enabled: true,
maxExecutionsPerProcess: 20,
},
logLevel: "log",
maxDuration: 3600,
retries: {