Tasks

Tasks are the boundary for side effects and nondeterministic work in a workflow. A task handler is where network requests, database writes, reading the clock, and generating random values belong.

When your workflow calls a task, the handler runs then and there, on the worker executing the run. Nothing is queued and nothing is handed to another process - a task is an ordinary function call, with its result recorded. Child workflows are what gets dispatched.

Once a task completes, its result is recorded and reused on replay instead of running the handler again. A task can still execute more than once - a worker may fail after the side effect lands but before the result is durably recorded - so task handlers should be idempotent.

Defining a Task

import { task } from "@aikirun/workflow";

const sendEmail = task({
	name: "send-email",
	handler(input: { email: string; message: string }) {
		// Your business logic
		return sendEmailToUser(input.email, input.message);
	},
});

Task Properties

name

A unique identifier for the task. Use descriptive names like "send-email" or "process-payment".

handler Function

The function that performs the actual work. It receives:

• input - Input data for the task

const processPayment = task({
	name: "process-payment",
	handler(input: { paymentId: string; amount: number }) {
		console.log(`Processing payment for ${input.paymentId}`);

		return processPaymentWithId(input.paymentId, input.amount);
	},
});

Executing Tasks

Tasks are executed within workflows using .start(), which calls the handler and returns its result:

const orderWorkflowV1 = orderWorkflow.v("1.0.0", {
	async handler(run, input: { orderData: any }) {
		const validation = await validateOrder.start(run, {
			orderData: input.orderData,
		});

		const payment = await processPayment.start(run, {
			paymentId: validation.paymentId,
			amount: validation.amount,
		});

		return { success: true };
	},
});

Starting several tasks with Promise.all works, and interleaves them the way any async code interleaves in one process - useful for overlapping I/O, but it does not spread the work across workers. Reach for child workflows when you want that.

Defining a Task Inside a Workflow

A task can be defined where it is used, inside the workflow handler:

const orderWorkflowV1 = orderWorkflow.v("1.0.0", {
	async handler(run, input: { orderId: string; email: string }) {
		const receipt = await task({
			name: "charge-order",
			handler: ({ orderId }: { orderId: string }) => chargeOrder(orderId),
		}).start(run, { orderId: input.orderId });

		await task({
			name: "email-receipt",
			handler: ({ email, receiptId }: { email: string; receiptId: string }) => emailReceipt(email, receiptId),
		}).start(run, { email: input.email, receiptId: receipt.id });
	},
});

It is a task like any other. Aiki identifies it by its name and input, and a replay that reaches it with a different input gets a non-determinism error.

A handler can also take no input and use values from the workflow handler directly:

const receipt = await task({
	name: "charge-order",
	handler: () => chargeOrder(input.orderId),
}).start(run);

Warning: Prefer passing input. Without it, the name is all that identifies the task, and Aiki does not see the values the handler uses:

• A replay returns the recorded result even when the values the task takes from the workflow handler have changed. No non-determinism error is raised.
• Tasks that share a name get their recorded results in the order they are called, so reordering them swaps their results.
• Tasks that share a name and are started together, as with Promise.all, can get each other's results on replay.

If you do use it, give each task its own name, which avoids the last two.

A task handler cannot use run. If it starts a task or a child workflow, sleeps, or waits through run, the task and the run fail, and the task is not retried.

Task Retry

Configure automatic retries for failed tasks using the retry property:

const processPayment = task({
	name: "process-payment",
	handler(input: { paymentId: string; amount: number }) {
		return paymentService.charge(input.paymentId, input.amount);
	},
	retry: {
		type: "exponential",
		maxAttempts: 3,
		baseDelayMs: 1000,
	},
});

For available strategies and best practices, see the Retry Strategies Guide.

Handling a Failed Task

When a task runs out of attempts, the task's start throws a TaskFailedError in the workflow handler. Left uncaught, the error fails the workflow attempt, as described in Task States. Catch it to react to the failure, for example to undo work done earlier in the run:

import { TaskFailedError } from "@aikirun/workflow";

const orderWorkflowV1 = orderWorkflow.v("1.0.0", {
	async handler(run, input: { orderId: string; paymentId: string; amount: number }) {
		await reserveInventory.start(run, { orderId: input.orderId });

		try {
			await processPayment.start(run, { paymentId: input.paymentId, amount: input.amount });
		} catch (err) {
			if (err instanceof TaskFailedError) {
				await releaseInventory.start(run, { orderId: input.orderId });
			}
			throw err;
		}
	},
});

Rethrowing, as here, still fails the attempt once the inventory is released. Leave out the throw and the workflow carries on.

Task States

A task moves through these states:

• running - Executing now
• awaiting_retry - The attempt failed and the task is backing off before the next one; carries the error and the time of the next attempt
• completed - Succeeded; its output is recorded and returned on replay
• failed - The retry strategy has no attempts left, or the task's output was rejected
• discarded - An unfinished task the run left behind; it no longer takes part in replay

A task's retry delay is durable unless it is short. When the delay is longer than the worker's workflowRun.maxInlineWaitMs setting (10ms by default), it is recorded: the task moves to awaiting_retry, its run is suspended in awaiting_task_retry and releases its worker, and the run is re-queued when the task is due. A shorter delay is not durable: the worker waits it out in memory, the task stays running, the run is not suspended, and the failed attempt is not recorded.

A task's attempts are its own, separate from the run's, so a task backing off does not move the run's attempt count. When the task runs out of attempts it goes failed, and the workflow attempt fails too, unless the workflow handler catches the error the task's start throws. The run then moves to awaiting_retry if the workflow has attempts left, or to failed if it does not, with cause task either way.

A task whose output fails its output schema, or cannot be stored as JSON, is not retried: it goes failed at once, and the run goes failed with cause task without a workflow retry.

Only unfinished tasks are ever discarded. Cancelling or stalling a run discards the tasks it left running or awaiting_retry, since nothing will finish them. Retrying a workflow attempt discards those too, along with any failed task, so the new attempt runs them again from their first attempt. A completed task is never discarded - its output is kept and replayed, so a retry resumes rather than repeating work that already succeeded.

Schema Validation

Define schemas to validate task input and output at runtime:

import { z } from "zod";

const processPayment = task({
	name: "process-payment",
	schema: {
		input: z.object({
			paymentId: z.string(),
			amount: z.number().positive(),
		}),
		output: z.object({
			transactionId: z.string(),
			status: z.enum(["success", "failed"]),
		}),
	},
	handler(input) {
		return paymentService.charge(input);
	},
});

Schemas work with any validation library that implements Standard Schema (Zod, Valibot, ArkType, etc.).

Why use output schemas? The output schema checks what the handler returns, at the moment it returns it, so a task cannot record a result that does not match its declared shape. It runs when the task executes, not when a recorded result is replayed - a result recorded before you changed the shape is replayed as it was written. See Refactoring Workflows.

Task Input and Output

A task receives its input directly and returns its output. Both must be plain JSON data; see Inputs and Outputs.

const exampleTask = task({
	name: "example",
	handler(input: { data: string }) {
		// input: Input data for this task
		console.log("Task input:", input);

		return { processed: true };
	},
});

Task Best Practices

1. Keep tasks focused - One responsibility per task
2. Make tasks idempotent - A task may re-execute after a crash; running it twice with the same input should be safe
3. Put side effects in tasks - API calls, database writes, and anything non-deterministic belong in tasks, keeping the workflow handler deterministic
4. Use meaningful names - Clear, descriptive task names

Next Steps

• Workflows - Learn about workflow orchestration
• Determinism - Workflow determinism and task idempotency
• Reference IDs - Custom identifiers for workflows and events
• Dependency Injection - Inject dependencies into tasks