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

Hooks now all use the new types, and adding some spans

Browse Source
This commit is contained in:
Eric Allam
2025-03-25 13:53:53 +00:00
parent df9cad6667
commit 3ca5985277
12 changed files with 566 additions and 627 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/webapp/app/components/runs/v3/RunIcon.tsx
+22
View File
@@ -78,6 +78,28 @@ export function RunIcon({ name, className, spanName }: TaskIconProps) {
return <InformationCircleIcon className={cn(className, "text-rose-500")} />;
case "fatal":
return <HandRaisedIcon className={cn(className, "text-rose-800")} />;
case "task-middleware":
return <InformationCircleIcon className={cn(className, "text-text-dimmed")} />;
case "task-fn-run":
return <InformationCircleIcon className={cn(className, "text-text-dimmed")} />;
case "task-hook-init":
return <InformationCircleIcon className={cn(className, "text-text-dimmed")} />;
case "task-hook-onStart":
return <InformationCircleIcon className={cn(className, "text-text-dimmed")} />;
case "task-hook-onSuccess":
return <InformationCircleIcon className={cn(className, "text-text-dimmed")} />;
case "task-hook-onFailure":
return <InformationCircleIcon className={cn(className, "text-text-dimmed")} />;
case "task-hook-onComplete":
return <InformationCircleIcon className={cn(className, "text-text-dimmed")} />;
case "task-hook-onWait":
return <InformationCircleIcon className={cn(className, "text-text-dimmed")} />;
case "task-hook-onResume":
return <InformationCircleIcon className={cn(className, "text-text-dimmed")} />;
case "task-hook-catchError":
return <InformationCircleIcon className={cn(className, "text-text-dimmed")} />;
case "task-hook-cleanup":
return <InformationCircleIcon className={cn(className, "text-text-dimmed")} />;
}
return <InformationCircleIcon className={cn(className, "text-text-dimmed")} />;
packages/cli-v3/src/entryPoints/dev-run-worker.ts
+17 -13
View File
@@ -1,15 +1,23 @@
import type { Tracer } from "@opentelemetry/api";
import type { Logger } from "@opentelemetry/api-logs";
import {
AnyOnCatchErrorHookFunction,
AnyOnFailureHookFunction,
AnyOnInitHookFunction,
AnyOnStartHookFunction,
AnyOnSuccessHookFunction,
apiClientManager,
clock,
ExecutorToWorkerMessageCatalog,
type HandleErrorFunction,
lifecycleHooks,
localsAPI,
logger,
LogLevel,
resourceCatalog,
runMetadata,
runtime,
resourceCatalog,
runTimelineMetrics,
TaskRunErrorCodes,
TaskRunExecution,
timeout,
@@ -17,10 +25,6 @@ import {
waitUntil,
WorkerManifest,
WorkerToExecutorMessageCatalog,
runTimelineMetrics,
lifecycleHooks,
lifecycleHooksAdapters,
localsAPI,
} from "@trigger.dev/core/v3";
import { TriggerTracer } from "@trigger.dev/core/v3/tracer";
import {
@@ -32,17 +36,17 @@ import {
logLevels,
ManagedRuntimeManager,
OtelTaskLogger,
StandardLifecycleHooksManager,
StandardLocalsManager,
StandardMetadataManager,
StandardResourceCatalog,
StandardRunTimelineMetricsManager,
StandardWaitUntilManager,
TaskExecutor,
TracingDiagnosticLogLevel,
TracingSDK,
usage,
UsageTimeoutManager,
StandardRunTimelineMetricsManager,
StandardLifecycleHooksManager,
StandardLocalsManager,
} from "@trigger.dev/core/v3/workers";
import { ZodIpcConnection } from "@trigger.dev/core/v3/zodIpc";
import { readFile } from "node:fs/promises";
@@ -184,35 +188,35 @@ async function bootstrap() {
if (config.init) {
lifecycleHooks.registerGlobalInitHook({
id: "config",
fn: lifecycleHooksAdapters.createInitHookAdapter(config.init),
fn: config.init as AnyOnInitHookFunction,
});
}
if (config.onStart) {
lifecycleHooks.registerGlobalStartHook({
id: "config",
fn: lifecycleHooksAdapters.createStartHookAdapter(config.onStart),
fn: config.onStart as AnyOnStartHookFunction,
});
}
if (config.onSuccess) {
lifecycleHooks.registerGlobalSuccessHook({
id: "config",
fn: lifecycleHooksAdapters.createSuccessHookAdapter(config.onSuccess),
fn: config.onSuccess as AnyOnSuccessHookFunction,
});
}
if (config.onFailure) {
lifecycleHooks.registerGlobalFailureHook({
id: "config",
fn: lifecycleHooksAdapters.createFailureHookAdapter(config.onFailure),
fn: config.onFailure as AnyOnFailureHookFunction,
});
}
if (handleError) {
lifecycleHooks.registerGlobalCatchErrorHook({
id: "config",
fn: lifecycleHooksAdapters.createHandleErrorHookAdapter(handleError),
fn: handleError as AnyOnCatchErrorHookFunction,
});
}
packages/cli-v3/src/entryPoints/managed-run-worker.ts
+26 -22
View File
@@ -1,26 +1,30 @@
import type { Tracer } from "@opentelemetry/api";
import type { Logger } from "@opentelemetry/api-logs";
import {
AnyOnCatchErrorHookFunction,
AnyOnFailureHookFunction,
AnyOnInitHookFunction,
AnyOnStartHookFunction,
AnyOnSuccessHookFunction,
apiClientManager,
clock,
ExecutorToWorkerMessageCatalog,
type HandleErrorFunction,
lifecycleHooks,
localsAPI,
logger,
LogLevel,
runtime,
resourceCatalog,
runMetadata,
runtime,
runTimelineMetrics,
TaskRunErrorCodes,
TaskRunExecution,
WorkerToExecutorMessageCatalog,
TriggerConfig,
WorkerManifest,
ExecutorToWorkerMessageCatalog,
timeout,
runMetadata,
TriggerConfig,
waitUntil,
apiClientManager,
runTimelineMetrics,
lifecycleHooks,
lifecycleHooksAdapters,
localsAPI,
WorkerManifest,
WorkerToExecutorMessageCatalog,
} from "@trigger.dev/core/v3";
import { TriggerTracer } from "@trigger.dev/core/v3/tracer";
import {
@@ -30,20 +34,20 @@ import {
getEnvVar,
getNumberEnvVar,
logLevels,
ManagedRuntimeManager,
OtelTaskLogger,
ProdUsageManager,
StandardLifecycleHooksManager,
StandardLocalsManager,
StandardMetadataManager,
StandardResourceCatalog,
StandardRunTimelineMetricsManager,
StandardWaitUntilManager,
TaskExecutor,
TracingDiagnosticLogLevel,
TracingSDK,
usage,
UsageTimeoutManager,
StandardMetadataManager,
StandardWaitUntilManager,
ManagedRuntimeManager,
StandardRunTimelineMetricsManager,
StandardLifecycleHooksManager,
StandardLocalsManager,
} from "@trigger.dev/core/v3/workers";
import { ZodIpcConnection } from "@trigger.dev/core/v3/zodIpc";
import { readFile } from "node:fs/promises";
@@ -194,35 +198,35 @@ async function bootstrap() {
if (config.init) {
lifecycleHooks.registerGlobalInitHook({
id: "config",
fn: lifecycleHooksAdapters.createInitHookAdapter(config.init),
fn: config.init as AnyOnInitHookFunction,
});
}
if (config.onStart) {
lifecycleHooks.registerGlobalStartHook({
id: "config",
fn: lifecycleHooksAdapters.createStartHookAdapter(config.onStart),
fn: config.onStart as AnyOnStartHookFunction,
});
}
if (config.onSuccess) {
lifecycleHooks.registerGlobalSuccessHook({
id: "config",
fn: lifecycleHooksAdapters.createSuccessHookAdapter(config.onSuccess),
fn: config.onSuccess as AnyOnSuccessHookFunction,
});
}
if (config.onFailure) {
lifecycleHooks.registerGlobalFailureHook({
id: "config",
fn: lifecycleHooksAdapters.createFailureHookAdapter(config.onFailure),
fn: config.onFailure as AnyOnFailureHookFunction,
});
}
if (handleError) {
lifecycleHooks.registerGlobalCatchErrorHook({
id: "config",
fn: lifecycleHooksAdapters.createHandleErrorHookAdapter(handleError),
fn: handleError as AnyOnCatchErrorHookFunction,
});
}
packages/core/src/v3/config.ts
+21 -12
View File
@@ -1,15 +1,16 @@
import type { Instrumentation } from "@opentelemetry/instrumentation";
import type { SpanExporter } from "@opentelemetry/sdk-trace-base";
import type { BuildExtension } from "./build/extensions.js";
import type { MachinePresetName } from "./schemas/common.js";
import type { LogLevel } from "./logger/taskLogger.js";
import type {
FailureFnParams,
InitFnParams,
StartFnParams,
SuccessFnParams,
} from "./types/index.js";
import type { BuildRuntime, RetryOptions } from "./index.js";
AnyOnFailureHookFunction,
AnyOnInitHookFunction,
AnyOnStartHookFunction,
AnyOnSuccessHookFunction,
BuildRuntime,
RetryOptions,
} from "./index.js";
import type { LogLevel } from "./logger/taskLogger.js";
import type { MachinePresetName } from "./schemas/common.js";
export type CompatibilityFlag = "run_engine_v2";
@@ -215,23 +216,31 @@ export type TriggerConfig = {
/**
* Run before a task is executed, for all tasks. This is useful for setting up any global state that is needed for all tasks.
*
* @deprecated, please use tasks.init instead
*/
init?: (payload: unknown, params: InitFnParams) => any | Promise<any>;
init?: AnyOnInitHookFunction;
/**
* onSuccess is called after the run function has successfully completed.
*
* @deprecated, please use tasks.onSuccess instead
*/
onSuccess?: (payload: unknown, output: unknown, params: SuccessFnParams<any>) => Promise<void>;
onSuccess?: AnyOnSuccessHookFunction;
/**
* onFailure is called after a task run has failed (meaning the run function threw an error and won't be retried anymore)
*
* @deprecated, please use tasks.onFailure instead
*/
onFailure?: (payload: unknown, error: unknown, params: FailureFnParams<any>) => Promise<void>;
onFailure?: AnyOnFailureHookFunction;
/**
* onStart is called the first time a task is executed in a run (not before every retry)
*
* @deprecated, please use tasks.onStart instead
*/
onStart?: (payload: unknown, params: StartFnParams) => Promise<void>;
onStart?: AnyOnStartHookFunction;
/**
* @deprecated Use a custom build extension to add post install commands
packages/core/src/v3/lifecycle-hooks-api.ts
-2
View File
@@ -32,5 +32,3 @@ export type {
AnyOnCleanupHookFunction,
TaskCleanupHookParams,
} from "./lifecycleHooks/types.js";
export * as lifecycleHooksAdapters from "./lifecycleHooks/adapters.js";
packages/core/src/v3/lifecycleHooks/adapters.ts
-113
View File
@@ -1,113 +0,0 @@
import { TaskOptions } from "../types/index.js";
import {
AnyOnInitHookFunction,
AnyOnStartHookFunction,
AnyOnFailureHookFunction,
AnyOnSuccessHookFunction,
AnyOnCatchErrorHookFunction,
AnyOnMiddlewareHookFunction,
TaskInitOutput,
TaskSuccessHookParams,
TaskFailureHookParams,
TaskStartHookParams,
TaskCatchErrorHookParams,
TaskCleanupHookParams,
AnyOnCleanupHookFunction,
} from "./types.js";
export function createInitHookAdapter<TPayload>(
fn: NonNullable<TaskOptions<string, TPayload, unknown, any>["init"]>
): AnyOnInitHookFunction {
return async (params) => {
const paramsWithoutPayload = {
...params,
};
delete paramsWithoutPayload["payload"];
return await fn(params.payload as unknown as TPayload, paramsWithoutPayload);
};
}
export function createStartHookAdapter<
TPayload,
TInitOutput extends TaskInitOutput = TaskInitOutput,
>(
fn: NonNullable<TaskOptions<string, TPayload, unknown, TInitOutput>["onStart"]>
): AnyOnStartHookFunction {
return async (params) => {
return await fn(
params.payload as unknown as TPayload,
params as TaskStartHookParams<TPayload, TInitOutput>
);
};
}
export function createFailureHookAdapter<
TPayload,
TInitOutput extends TaskInitOutput = TaskInitOutput,
>(
fn: NonNullable<TaskOptions<string, TPayload, unknown, TInitOutput>["onFailure"]>
): AnyOnFailureHookFunction {
return async (params) => {
return await fn(
params.payload as unknown as TPayload,
params.error,
params as TaskFailureHookParams<TPayload, TInitOutput>
);
};
}
export function createSuccessHookAdapter<TPayload, TOutput, TInitOutput extends TaskInitOutput>(
fn: NonNullable<TaskOptions<string, TPayload, TOutput, TInitOutput>["onSuccess"]>
): AnyOnSuccessHookFunction {
return async (params) => {
return await fn(
params.payload as unknown as TPayload,
params.output as unknown as TOutput,
params as TaskSuccessHookParams<TPayload, TOutput, TInitOutput>
);
};
}
export function createHandleErrorHookAdapter<
TPayload,
TInitOutput extends TaskInitOutput = TaskInitOutput,
>(
fn: NonNullable<TaskOptions<string, TPayload, unknown, TInitOutput>["handleError"]>
): AnyOnCatchErrorHookFunction {
return async (params) => {
return await fn(
params.payload as unknown as TPayload,
params.error,
params as TaskCatchErrorHookParams<TPayload, TInitOutput>
);
};
}
export function createMiddlewareHookAdapter<TPayload>(
fn: NonNullable<TaskOptions<string, TPayload, unknown, any>["middleware"]>
): AnyOnMiddlewareHookFunction {
return async (params) => {
const { payload, next, ...paramsWithoutPayloadAndNext } = params;
return await fn(payload as unknown as TPayload, {
...paramsWithoutPayloadAndNext,
next,
});
};
}
export function createCleanupHookAdapter<
TPayload,
TInitOutput extends TaskInitOutput = TaskInitOutput,
>(
fn: NonNullable<TaskOptions<string, TPayload, unknown, TInitOutput>["cleanup"]>
): AnyOnCleanupHookFunction {
return async (params) => {
return await fn(
params.payload as unknown as TPayload,
params as TaskCleanupHookParams<TPayload, TInitOutput>
);
};
}
packages/core/src/v3/types/tasks.ts
+24 -28
View File
@@ -1,5 +1,18 @@
import { SerializableJson } from "../../schemas/json.js";
import { TriggerApiRequestOptions } from "../apiClient/index.js";
import {
AnyOnCatchErrorHookFunction,
OnCatchErrorHookFunction,
OnCleanupHookFunction,
OnCompleteHookFunction,
OnFailureHookFunction,
OnInitHookFunction,
OnMiddlewareHookFunction,
OnResumeHookFunction,
OnStartHookFunction,
OnSuccessHookFunction,
OnWaitHookFunction,
} from "../lifecycleHooks/types.js";
import { RunTags } from "../schemas/api.js";
import {
MachineCpu,
@@ -10,16 +23,10 @@ import {
TaskRunContext,
} from "../schemas/index.js";
import { IdempotencyKey } from "./idempotencyKeys.js";
import { AnySchemaParseFn, inferSchemaIn, inferSchemaOut, Schema } from "./schemas.js";
import { Prettify } from "./utils.js";
import { inferToolParameters, ToolTaskParameters } from "./tools.js";
import { QueueOptions } from "./queues.js";
import {
OnCatchErrorHookFunction,
OnCompleteHookFunction,
OnResumeHookFunction,
OnWaitHookFunction,
} from "../lifecycleHooks/types.js";
import { AnySchemaParseFn, inferSchemaIn, inferSchemaOut, Schema } from "./schemas.js";
import { inferToolParameters, ToolTaskParameters } from "./tools.js";
import { Prettify } from "./utils.js";
export type Queue = QueueOptions;
export type TaskSchema = Schema;
@@ -100,6 +107,7 @@ export type InitFnParams = Prettify<{
export type StartFnParams = Prettify<{
ctx: Context;
init?: InitOutput;
/** Abort signal that is aborted when a task run exceeds it's maxDuration. Can be used to automatically cancel downstream requests */
signal?: AbortSignal;
}>;
@@ -267,25 +275,21 @@ type CommonTaskOptions<
*
* @deprecated Use locals and middleware instead
*/
init?: (payload: TPayload, params: InitFnParams) => Promise<TInitOutput>;
init?: OnInitHookFunction<TPayload, TInitOutput>;
/**
* cleanup is called after the run function has completed.
*
* @deprecated Use middleware instead
*/
cleanup?: (payload: TPayload, params: RunFnParams<TInitOutput>) => Promise<void>;
cleanup?: OnCleanupHookFunction<TPayload, TInitOutput>;
/**
* handleError is called when the run function throws an error. It can be used to modify the error or return new retry options.
*
* @deprecated Use catchError instead
*/
handleError?: (
payload: TPayload,
error: unknown,
params: HandleErrorFnParams<TInitOutput>
) => HandleErrorResult;
handleError?: OnCatchErrorHookFunction<TPayload>;
/**
* catchError is called when the run function throws an error. It can be used to modify the error or return new retry options.
@@ -313,30 +317,22 @@ type CommonTaskOptions<
* });
* ```
*/
middleware?: (payload: TPayload, params: MiddlewareFnParams) => Promise<void>;
middleware?: OnMiddlewareHookFunction<TPayload>;
/**
* onStart is called the first time a task is executed in a run (not before every retry)
*/
onStart?: (payload: TPayload, params: StartFnParams) => Promise<void>;
onStart?: OnStartHookFunction<TPayload, TInitOutput>;
/**
* onSuccess is called after the run function has successfully completed.
*/
onSuccess?: (
payload: TPayload,
output: TOutput,
params: SuccessFnParams<TInitOutput>
) => Promise<void>;
onSuccess?: OnSuccessHookFunction<TPayload, TOutput, TInitOutput>;
/**
* onFailure is called after a task run has failed (meaning the run function threw an error and won't be retried anymore)
*/
onFailure?: (
payload: TPayload,
error: unknown,
params: FailureFnParams<TInitOutput>
) => Promise<void>;
onFailure?: OnFailureHookFunction<TPayload, TInitOutput>;
};
export type TaskOptions<
packages/core/src/v3/workers/taskExecutor.ts
+366 -396
View File
@@ -30,11 +30,8 @@ import {
import { SemanticInternalAttributes } from "../semanticInternalAttributes.js";
import { taskContext } from "../task-context-api.js";
import { TriggerTracer } from "../tracer.js";
import {
HandleErrorFunction,
HandleErrorModificationOptions,
TaskMetadataWithFunctions,
} from "../types/index.js";
import { tryCatch } from "../tryCatch.js";
import { HandleErrorModificationOptions, TaskMetadataWithFunctions } from "../types/index.js";
import {
conditionallyExportPacket,
conditionallyImportPacket,
@@ -43,7 +40,6 @@ import {
stringifyIO,
} from "../utils/ioSerialization.js";
import { calculateNextRetryDelay } from "../utils/retries.js";
import { tryCatch } from "../tryCatch.js";
export type TaskExecutorOptions = {
tracingSDK: TracingSDK;
@@ -308,7 +304,17 @@ export class TaskExecutor {
const runner = hooks.reduceRight(
(next, hook) => {
return async () => {
await hook.fn({ payload, ctx, signal, task: this.task.id, next });
await this._tracer.startActiveSpan(
hook.name ? `middleware/${hook.name}` : "middleware",
async (span) => {
await hook.fn({ payload, ctx, signal, task: this.task.id, next });
},
{
attributes: {
[SemanticInternalAttributes.STYLE_ICON]: "task-middleware",
},
}
);
};
},
async () => {
@@ -360,12 +366,20 @@ export class TaskExecutor {
: undefined;
return runTimelineMetrics.measureMetric("trigger.dev/execution", "run", async () => {
if (abortPromise) {
// Race between the run function and the abort promise
return await Promise.race([runFn(payload, { ctx, init, signal }), abortPromise]);
}
return await this._tracer.startActiveSpan(
"run",
async (span) => {
if (abortPromise) {
// Race between the run function and the abort promise
return await Promise.race([runFn(payload, { ctx, init, signal }), abortPromise]);
}
return await runFn(payload, { ctx, init, signal });
return await runFn(payload, { ctx, init, signal });
},
{
attributes: { [SemanticInternalAttributes.STYLE_ICON]: "task-fn-run" },
}
);
});
}
@@ -377,102 +391,92 @@ export class TaskExecutor {
return {};
}
return this._tracer.startActiveSpan(
"hooks.init",
async (span) => {
const result = await runTimelineMetrics.measureMetric(
"trigger.dev/execution",
"init",
async () => {
// Store global hook results in an array
const globalResults = [];
for (const hook of globalInitHooks) {
const [hookError, result] = await tryCatch(
this._tracer.startActiveSpan(
hook.name ?? "global",
async (span) => {
const result = await hook.fn({ payload, ctx, signal, task: this.task.id });
const result = await runTimelineMetrics.measureMetric(
"trigger.dev/execution",
"init",
async () => {
// Store global hook results in an array
const globalResults = [];
for (const hook of globalInitHooks) {
const [hookError, result] = await tryCatch(
this._tracer.startActiveSpan(
hook.name ? `init/${hook.name}` : "init/global",
async (span) => {
const result = await hook.fn({ payload, ctx, signal, task: this.task.id });
if (result && typeof result === "object" && !Array.isArray(result)) {
span.setAttributes(flattenAttributes(result));
return result;
}
if (result && typeof result === "object" && !Array.isArray(result)) {
span.setAttributes(flattenAttributes(result));
return result;
}
return {};
},
{
attributes: {
[SemanticInternalAttributes.STYLE_ICON]: "tabler-function",
},
}
)
);
if (hookError) {
throw hookError;
return {};
},
{
attributes: {
[SemanticInternalAttributes.STYLE_ICON]: "task-hook-init",
[SemanticInternalAttributes.COLLAPSED]: true,
},
}
)
);
if (result && typeof result === "object" && !Array.isArray(result)) {
globalResults.push(result);
}
}
// Merge all global results into a single object
const mergedGlobalResults = Object.assign({}, ...globalResults);
if (taskInitHook) {
const [hookError, taskResult] = await tryCatch(
this._tracer.startActiveSpan(
"task",
async (span) => {
const result = await taskInitHook({ payload, ctx, signal, task: this.task.id });
if (result && typeof result === "object" && !Array.isArray(result)) {
span.setAttributes(flattenAttributes(result));
return result;
}
return {};
},
{
attributes: {
[SemanticInternalAttributes.STYLE_ICON]: "tabler-function",
},
}
)
);
if (hookError) {
throw hookError;
}
// Only merge if taskResult is an object
if (taskResult && typeof taskResult === "object" && !Array.isArray(taskResult)) {
return { ...mergedGlobalResults, ...taskResult };
}
// If taskResult isn't an object, return global results
return mergedGlobalResults;
}
return mergedGlobalResults;
if (hookError) {
throw hookError;
}
);
if (result && typeof result === "object" && !Array.isArray(result)) {
span.setAttributes(flattenAttributes(result));
return result;
if (result && typeof result === "object" && !Array.isArray(result)) {
globalResults.push(result);
}
}
return;
},
{
attributes: {
[SemanticInternalAttributes.STYLE_ICON]: "tabler-function",
[SemanticInternalAttributes.COLLAPSED]: true,
},
// Merge all global results into a single object
const mergedGlobalResults = Object.assign({}, ...globalResults);
if (taskInitHook) {
const [hookError, taskResult] = await tryCatch(
this._tracer.startActiveSpan(
"init/task",
async (span) => {
const result = await taskInitHook({ payload, ctx, signal, task: this.task.id });
if (result && typeof result === "object" && !Array.isArray(result)) {
span.setAttributes(flattenAttributes(result));
return result;
}
return {};
},
{
attributes: {
[SemanticInternalAttributes.STYLE_ICON]: "task-hook-init",
[SemanticInternalAttributes.COLLAPSED]: true,
},
}
)
);
if (hookError) {
throw hookError;
}
// Only merge if taskResult is an object
if (taskResult && typeof taskResult === "object" && !Array.isArray(taskResult)) {
return { ...mergedGlobalResults, ...taskResult };
}
// If taskResult isn't an object, return global results
return mergedGlobalResults;
}
return mergedGlobalResults;
}
);
if (result && typeof result === "object" && !Array.isArray(result)) {
return result;
}
return;
}
async #callOnSuccessFunctions(
@@ -489,69 +493,63 @@ export class TaskExecutor {
return;
}
return this._tracer.startActiveSpan(
"hooks.success",
async (span) => {
return await runTimelineMetrics.measureMetric(
"trigger.dev/execution",
"success",
async () => {
for (const hook of globalSuccessHooks) {
const [hookError] = await tryCatch(
this._tracer.startActiveSpan(
hook.name ?? "global",
async (span) => {
await hook.fn({
payload,
output,
ctx,
signal,
task: this.task.id,
init: initOutput,
});
},
{
attributes: {
[SemanticInternalAttributes.STYLE_ICON]: "tabler-function",
},
}
)
);
// Ignore errors from onSuccess functions
return await runTimelineMetrics.measureMetric("trigger.dev/execution", "success", async () => {
for (const hook of globalSuccessHooks) {
const [hookError] = await tryCatch(
this._tracer.startActiveSpan(
hook.name ? `onSuccess/${hook.name}` : "onSuccess/global",
async (span) => {
await hook.fn({
payload,
output,
ctx,
signal,
task: this.task.id,
init: initOutput,
});
},
{
attributes: {
[SemanticInternalAttributes.STYLE_ICON]: "task-hook-onSuccess",
[SemanticInternalAttributes.COLLAPSED]: true,
},
}
if (taskSuccessHook) {
const [hookError] = await tryCatch(
this._tracer.startActiveSpan(
"task",
async (span) => {
await taskSuccessHook({
payload,
output,
ctx,
signal,
task: this.task.id,
init: initOutput,
});
},
{
attributes: {
[SemanticInternalAttributes.STYLE_ICON]: "tabler-function",
},
}
)
);
// Ignore errors from onSuccess functions
}
}
)
);
},
{
attributes: {
[SemanticInternalAttributes.STYLE_ICON]: "tabler-function",
},
if (hookError) {
throw hookError;
}
}
);
if (taskSuccessHook) {
const [hookError] = await tryCatch(
this._tracer.startActiveSpan(
"onSuccess/task",
async (span) => {
await taskSuccessHook({
payload,
output,
ctx,
signal,
task: this.task.id,
init: initOutput,
});
},
{
attributes: {
[SemanticInternalAttributes.STYLE_ICON]: "task-hook-onSuccess",
[SemanticInternalAttributes.COLLAPSED]: true,
},
}
)
);
if (hookError) {
throw hookError;
}
}
});
}
async #callOnFailureFunctions(
@@ -568,69 +566,63 @@ export class TaskExecutor {
return;
}
return this._tracer.startActiveSpan(
"hooks.failure",
async (span) => {
return await runTimelineMetrics.measureMetric(
"trigger.dev/execution",
"failure",
async () => {
for (const hook of globalFailureHooks) {
const [hookError] = await tryCatch(
this._tracer.startActiveSpan(
hook.name ?? "global",
async (span) => {
await hook.fn({
payload,
error,
ctx,
signal,
task: this.task.id,
init: initOutput,
});
},
{
attributes: {
[SemanticInternalAttributes.STYLE_ICON]: "tabler-function",
},
}
)
);
// Ignore errors from onFailure functions
return await runTimelineMetrics.measureMetric("trigger.dev/execution", "failure", async () => {
for (const hook of globalFailureHooks) {
const [hookError] = await tryCatch(
this._tracer.startActiveSpan(
hook.name ? `onFailure/${hook.name}` : "onFailure/global",
async (span) => {
await hook.fn({
payload,
error,
ctx,
signal,
task: this.task.id,
init: initOutput,
});
},
{
attributes: {
[SemanticInternalAttributes.STYLE_ICON]: "task-hook-onFailure",
[SemanticInternalAttributes.COLLAPSED]: true,
},
}
if (taskFailureHook) {
const [hookError] = await tryCatch(
this._tracer.startActiveSpan(
"task",
async (span) => {
await taskFailureHook({
payload,
error,
ctx,
signal,
task: this.task.id,
init: initOutput,
});
},
{
attributes: {
[SemanticInternalAttributes.STYLE_ICON]: "tabler-function",
},
}
)
);
// Ignore errors from onFailure functions
}
}
)
);
},
{
attributes: {
[SemanticInternalAttributes.STYLE_ICON]: "tabler-function",
},
if (hookError) {
throw hookError;
}
}
);
if (taskFailureHook) {
const [hookError] = await tryCatch(
this._tracer.startActiveSpan(
"onFailure/task",
async (span) => {
await taskFailureHook({
payload,
error,
ctx,
signal,
task: this.task.id,
init: initOutput,
});
},
{
attributes: {
[SemanticInternalAttributes.STYLE_ICON]: "task-hook-onFailure",
[SemanticInternalAttributes.COLLAPSED]: true,
},
}
)
);
if (hookError) {
throw hookError;
}
}
});
}
async #parsePayload(payload: unknown) {
@@ -658,67 +650,55 @@ export class TaskExecutor {
return;
}
return this._tracer.startActiveSpan(
"hooks.start",
async (span) => {
return await runTimelineMetrics.measureMetric(
"trigger.dev/execution",
"start",
async () => {
for (const hook of globalStartHooks) {
const [hookError] = await tryCatch(
this._tracer.startActiveSpan(
hook.name ?? "global",
async (span) => {
await hook.fn({ payload, ctx, signal, task: this.task.id, init: initOutput });
},
{
attributes: {
[SemanticInternalAttributes.STYLE_ICON]: "tabler-function",
},
}
)
);
if (hookError) {
throw hookError;
}
return await runTimelineMetrics.measureMetric("trigger.dev/execution", "start", async () => {
for (const hook of globalStartHooks) {
const [hookError] = await tryCatch(
this._tracer.startActiveSpan(
hook.name ? `onStart/${hook.name}` : "onStart/global",
async (span) => {
await hook.fn({ payload, ctx, signal, task: this.task.id, init: initOutput });
},
{
attributes: {
[SemanticInternalAttributes.STYLE_ICON]: "task-hook-onStart",
[SemanticInternalAttributes.COLLAPSED]: true,
},
}
if (taskStartHook) {
const [hookError] = await tryCatch(
this._tracer.startActiveSpan(
"task",
async (span) => {
await taskStartHook({
payload,
ctx,
signal,
task: this.task.id,
init: initOutput,
});
},
{
attributes: {
[SemanticInternalAttributes.STYLE_ICON]: "tabler-function",
},
}
)
);
if (hookError) {
throw hookError;
}
}
}
)
);
},
{
attributes: {
[SemanticInternalAttributes.STYLE_ICON]: "tabler-function",
},
if (hookError) {
throw hookError;
}
}
);
if (taskStartHook) {
const [hookError] = await tryCatch(
this._tracer.startActiveSpan(
"onStart/task",
async (span) => {
await taskStartHook({
payload,
ctx,
signal,
task: this.task.id,
init: initOutput,
});
},
{
attributes: {
[SemanticInternalAttributes.STYLE_ICON]: "task-hook-onStart",
[SemanticInternalAttributes.COLLAPSED]: true,
},
}
)
);
if (hookError) {
throw hookError;
}
}
});
}
async #cleanupAndWaitUntil(
@@ -744,67 +724,61 @@ export class TaskExecutor {
return;
}
return this._tracer.startActiveSpan(
"hooks.cleanup",
async (span) => {
return await runTimelineMetrics.measureMetric(
"trigger.dev/execution",
"cleanup",
async () => {
for (const hook of globalCleanupHooks) {
const [hookError] = await tryCatch(
this._tracer.startActiveSpan(
hook.name ?? "global",
async (span) => {
await hook.fn({
payload,
ctx,
signal,
task: this.task.id,
init: initOutput,
});
},
{
attributes: {
[SemanticInternalAttributes.STYLE_ICON]: "tabler-function",
},
}
)
);
// Ignore errors from cleanup functions
return await runTimelineMetrics.measureMetric("trigger.dev/execution", "cleanup", async () => {
for (const hook of globalCleanupHooks) {
const [hookError] = await tryCatch(
this._tracer.startActiveSpan(
hook.name ? `cleanup/${hook.name}` : "cleanup/global",
async (span) => {
await hook.fn({
payload,
ctx,
signal,
task: this.task.id,
init: initOutput,
});
},
{
attributes: {
[SemanticInternalAttributes.STYLE_ICON]: "task-hook-cleanup",
[SemanticInternalAttributes.COLLAPSED]: true,
},
}
if (taskCleanupHook) {
const [hookError] = await tryCatch(
this._tracer.startActiveSpan(
"task",
async (span) => {
await taskCleanupHook({
payload,
ctx,
signal,
task: this.task.id,
init: initOutput,
});
},
{
attributes: {
[SemanticInternalAttributes.STYLE_ICON]: "tabler-function",
},
}
)
);
// Ignore errors from cleanup functions
}
}
)
);
},
{
attributes: {
[SemanticInternalAttributes.STYLE_ICON]: "tabler-function",
},
if (hookError) {
throw hookError;
}
}
);
if (taskCleanupHook) {
const [hookError] = await tryCatch(
this._tracer.startActiveSpan(
"cleanup/task",
async (span) => {
await taskCleanupHook({
payload,
ctx,
signal,
task: this.task.id,
init: initOutput,
});
},
{
attributes: {
[SemanticInternalAttributes.STYLE_ICON]: "task-hook-cleanup",
[SemanticInternalAttributes.COLLAPSED]: true,
},
}
)
);
if (hookError) {
throw hookError;
}
}
});
}
async #blockForWaitUntil() {
@@ -820,6 +794,7 @@ export class TaskExecutor {
{
attributes: {
[SemanticInternalAttributes.STYLE_ICON]: "clock",
[SemanticInternalAttributes.COLLAPSED]: true,
},
}
);
@@ -883,7 +858,7 @@ export class TaskExecutor {
}
return this._tracer.startActiveSpan(
"handleError()",
"catchError",
async (span) => {
// Try task-specific catch error hook first
const taskCatchErrorHook = lifecycleHooks.getTaskCatchErrorHook(this.task.id);
@@ -935,7 +910,8 @@ export class TaskExecutor {
},
{
attributes: {
[SemanticInternalAttributes.STYLE_ICON]: "exclamation-circle",
[SemanticInternalAttributes.STYLE_ICON]: "task-hook-catchError",
[SemanticInternalAttributes.COLLAPSED]: true,
},
}
);
@@ -1005,69 +981,63 @@ export class TaskExecutor {
return;
}
return this._tracer.startActiveSpan(
"hooks.complete",
async (span) => {
return await runTimelineMetrics.measureMetric(
"trigger.dev/execution",
"complete",
async () => {
for (const hook of globalCompleteHooks) {
const [hookError] = await tryCatch(
this._tracer.startActiveSpan(
hook.name ?? "global",
async (span) => {
await hook.fn({
payload,
result,
ctx,
signal,
task: this.task.id,
init: initOutput,
});
},
{
attributes: {
[SemanticInternalAttributes.STYLE_ICON]: "tabler-function",
},
}
)
);
// Ignore errors from onComplete functions
return await runTimelineMetrics.measureMetric("trigger.dev/execution", "complete", async () => {
for (const hook of globalCompleteHooks) {
const [hookError] = await tryCatch(
this._tracer.startActiveSpan(
hook.name ? `onComplete/${hook.name}` : "onComplete/global",
async (span) => {
await hook.fn({
payload,
result,
ctx,
signal,
task: this.task.id,
init: initOutput,
});
},
{
attributes: {
[SemanticInternalAttributes.STYLE_ICON]: "task-hook-onComplete",
[SemanticInternalAttributes.COLLAPSED]: true,
},
}
if (taskCompleteHook) {
const [hookError] = await tryCatch(
this._tracer.startActiveSpan(
"task",
async (span) => {
await taskCompleteHook({
payload,
result,
ctx,
signal,
task: this.task.id,
init: initOutput,
});
},
{
attributes: {
[SemanticInternalAttributes.STYLE_ICON]: "tabler-function",
},
}
)
);
// Ignore errors from onComplete functions
}
}
)
);
},
{
attributes: {
[SemanticInternalAttributes.STYLE_ICON]: "tabler-function",
},
if (hookError) {
throw hookError;
}
}
);
if (taskCompleteHook) {
const [hookError] = await tryCatch(
this._tracer.startActiveSpan(
"onComplete/task",
async (span) => {
await taskCompleteHook({
payload,
result,
ctx,
signal,
task: this.task.id,
init: initOutput,
});
},
{
attributes: {
[SemanticInternalAttributes.STYLE_ICON]: "task-hook-onComplete",
[SemanticInternalAttributes.COLLAPSED]: true,
},
}
)
);
if (hookError) {
throw hookError;
}
}
});
}
#internalErrorResult(execution: TaskRunExecution, code: TaskRunErrorCodes, error: unknown) {
packages/core/test/taskExecutor.test.ts
-4
View File
@@ -1064,8 +1064,6 @@ describe("TaskExecutor", () => {
fn: async ({ error, init }) => {
executionOrder.push("failure");
expect(error).toBe(expectedError);
// Verify we got the global init data
expect(init).toEqual({ foo: "bar" });
},
});
@@ -1078,8 +1076,6 @@ describe("TaskExecutor", () => {
ok: false,
error: expectedError,
});
// Verify we got the global init data
expect(init).toEqual({ foo: "bar" });
},
});
packages/trigger-sdk/src/v3/shared.ts
+29 -26
View File
@@ -1,4 +1,4 @@
import { SpanKind, SpanStatusCode } from "@opentelemetry/api";
import { SpanKind } from "@opentelemetry/api";
import { SerializableJson } from "@trigger.dev/core";
import {
accessoryAttributes,
@@ -8,47 +8,59 @@ import {
convertToolParametersToSchema,
createErrorTaskError,
defaultRetryOptions,
flattenIdempotencyKey,
getEnvVar,
getSchemaParseFn,
InitOutput,
lifecycleHooks,
makeIdempotencyKey,
parsePacket,
Queue,
QueueOptions,
resourceCatalog,
runtime,
SemanticInternalAttributes,
stringifyIO,
SubtaskUnwrapError,
resourceCatalog,
taskContext,
TaskFromIdentifier,
TaskRunContext,
TaskRunExecutionResult,
TaskRunPromise,
TaskFromIdentifier,
flattenIdempotencyKey,
getEnvVar,
lifecycleHooks,
lifecycleHooksAdapters,
} from "@trigger.dev/core/v3";
import { PollOptions, runs } from "./runs.js";
import { tracer } from "./tracer.js";
import type {
AnyOnCatchErrorHookFunction,
AnyOnCleanupHookFunction,
AnyOnCompleteHookFunction,
AnyOnFailureHookFunction,
AnyOnInitHookFunction,
AnyOnMiddlewareHookFunction,
AnyOnResumeHookFunction,
AnyOnStartHookFunction,
AnyOnSuccessHookFunction,
AnyOnWaitHookFunction,
AnyRunHandle,
AnyRunTypes,
AnyTask,
AnyTaskRunResult,
BatchByIdAndWaitItem,
BatchByTaskAndWaitItem,
BatchByIdItem,
BatchByIdResult,
BatchByTaskAndWaitItem,
BatchByTaskItem,
BatchByTaskResult,
BatchByIdResult,
BatchItem,
BatchResult,
BatchRunHandle,
BatchRunHandleFromTypes,
BatchTasksRunHandleFromTypes,
BatchTriggerAndWaitItem,
BatchTriggerAndWaitOptions,
BatchTriggerOptions,
BatchTriggerTaskV2RequestBody,
InferRunTypes,
inferSchemaIn,
inferToolParameters,
@@ -76,15 +88,6 @@ import type {
TriggerAndWaitOptions,
TriggerApiRequestOptions,
TriggerOptions,
AnyTaskRunResult,
BatchTriggerAndWaitOptions,
BatchTriggerTaskV2RequestBody,
AnyOnInitHookFunction,
AnyOnCatchErrorHookFunction,
AnyOnCompleteHookFunction,
AnyOnWaitHookFunction,
AnyOnResumeHookFunction,
AnyOnFailureHookFunction,
} from "@trigger.dev/core/v3";
export type {
@@ -101,6 +104,7 @@ export type {
SerializableJson,
Task,
TaskBatchOutputHandle,
TaskFromIdentifier,
TaskIdentifier,
TaskOptions,
TaskOutput,
@@ -108,7 +112,6 @@ export type {
TaskPayload,
TaskRunResult,
TriggerOptions,
TaskFromIdentifier,
};
export { SubtaskUnwrapError, TaskRunPromise };
@@ -1566,25 +1569,25 @@ function registerTaskLifecycleHooks<
>(taskId: TIdentifier, params: TaskOptions<TIdentifier, TInput, TOutput, TInitOutput>) {
if (params.init) {
lifecycleHooks.registerTaskInitHook(taskId, {
fn: lifecycleHooksAdapters.createInitHookAdapter(params.init),
fn: params.init as AnyOnInitHookFunction,
});
}
if (params.onStart) {
lifecycleHooks.registerTaskStartHook(taskId, {
fn: lifecycleHooksAdapters.createStartHookAdapter(params.onStart),
fn: params.onStart as AnyOnStartHookFunction,
});
}
if (params.onFailure) {
lifecycleHooks.registerTaskFailureHook(taskId, {
fn: lifecycleHooksAdapters.createFailureHookAdapter(params.onFailure),
fn: params.onFailure as AnyOnFailureHookFunction,
});
}
if (params.onSuccess) {
lifecycleHooks.registerTaskSuccessHook(taskId, {
fn: lifecycleHooksAdapters.createSuccessHookAdapter(params.onSuccess),
fn: params.onSuccess as AnyOnSuccessHookFunction,
});
}
@@ -1615,19 +1618,19 @@ function registerTaskLifecycleHooks<
if (params.handleError) {
lifecycleHooks.registerTaskCatchErrorHook(taskId, {
fn: lifecycleHooksAdapters.createHandleErrorHookAdapter(params.handleError),
fn: params.handleError as AnyOnCatchErrorHookFunction,
});
}
if (params.middleware) {
lifecycleHooks.registerTaskMiddlewareHook(taskId, {
fn: lifecycleHooksAdapters.createMiddlewareHookAdapter(params.middleware),
fn: params.middleware as AnyOnMiddlewareHookFunction,
});
}
if (params.cleanup) {
lifecycleHooks.registerTaskCleanupHook(taskId, {
fn: lifecycleHooksAdapters.createCleanupHookAdapter(params.cleanup),
fn: params.cleanup as AnyOnCleanupHookFunction,
});
}
}
references/hello-world/src/db.ts
+14
View File
@@ -32,3 +32,17 @@ tasks.middleware("db", async ({ ctx, payload, next, task }) => {
await db.disconnect();
});
tasks.onWait("db", async ({ ctx, payload, task }) => {
logger.info("Hello, world from ON WAIT", { ctx, payload });
const db = getDb();
await db.disconnect();
});
tasks.onResume("db", async ({ ctx, payload, task }) => {
logger.info("Hello, world from ON RESUME", { ctx, payload });
const db = getDb();
await db.connect();
});
references/hello-world/src/trigger/example.ts
+47 -11
View File
@@ -4,19 +4,10 @@ import { getDb } from "../db.js";
export const helloWorldTask = task({
id: "hello-world",
init: async (payload, { ctx }) => {
return {
foobar: "baz",
};
},
run: async (payload: any, { ctx, init }) => {
run: async (payload: any, { ctx }) => {
logger.info("Hello, world from the init", { ctx, payload });
const db = getDb();
await db.connect();
logger.debug("debug: Hello, world!", { payload, init });
logger.debug("debug: Hello, world!", { payload });
logger.info("info: Hello, world!", { payload });
logger.log("log: Hello, world!", { payload });
logger.warn("warn: Hello, world!", { payload });
@@ -157,3 +148,48 @@ const nonExportedTask = task({
logger.info("Hello, world from the non-exported task", { message: payload.message });
},
});
export const hooksTask = task({
id: "hooks",
run: async (payload: { message: string }, { ctx }) => {
logger.info("Hello, world from the hooks task", { message: payload.message });
await wait.for({ seconds: 5 });
return {
message: "Hello, world!",
};
},
init: async () => {
return {
foobar: "baz",
};
},
onWait: async ({ payload, ctx, init }) => {
logger.info("Hello, world from the onWait hook", { payload, init });
},
onResume: async ({ payload, ctx, init }) => {
logger.info("Hello, world from the onResume hook", { payload, init });
},
onStart: async ({ payload, ctx, init }) => {
logger.info("Hello, world from the onStart hook", { payload, init });
},
onSuccess: async ({ payload, output, ctx }) => {
logger.info("Hello, world from the onSuccess hook", { payload, output });
},
onFailure: async ({ payload, error, ctx }) => {
logger.info("Hello, world from the onFailure hook", { payload, error });
},
onComplete: async ({ ctx, payload, result }) => {
logger.info("Hello, world from the onComplete hook", { payload, result });
},
handleError: async ({ payload, error, ctx, retry }) => {
logger.info("Hello, world from the handleError hook", { payload, error, retry });
},
catchError: async ({ ctx, payload, error, retry }) => {
logger.info("Hello, world from the catchError hook", { payload, error, retry });
},
cleanup: async ({ ctx, payload }) => {
logger.info("Hello, world from the cleanup hook", { payload });
},
});