A Task job is a user-controlled durable state machine. Unlike Continuous jobs (which run on a fixed schedule) or Debounce jobs (which flush after a delay), Task jobs give you complete control: you decide when to schedule execution, how to handle each event, and when the job is done.
Each Task instance runs in its own Durable Object. Events update state and optionally schedule an alarm. When the alarm fires, execute runs. You control what happens next — reschedule, terminate, or wait for another event.
Use Task jobs for multi-step workflows: order processing, document review pipelines, or any pattern where the next step depends on external events.
Defining a task job
import { Effect, Schema, Duration } from "effect";
import { Task } from "@durable-effect/jobs";
const orderProcessor = Task.make({
stateSchema: Schema.Struct({
orderId: Schema.String,
status: Schema.Literal("pending", "processing", "shipped", "delivered"),
attempts: Schema.Number,
}),
eventSchema: Schema.Union(
Schema.Struct({ _tag: Schema.Literal("OrderPlaced"), orderId: Schema.String }),
Schema.Struct({ _tag: Schema.Literal("PaymentReceived") }),
Schema.Struct({ _tag: Schema.Literal("Shipped"), trackingNumber: Schema.String }),
),
onEvent: (event, ctx) =>
Effect.gen(function* () {
switch (event._tag) {
case "OrderPlaced":
yield* ctx.setState({
orderId: event.orderId,
status: "pending",
attempts: 0,
});
yield* ctx.schedule(Duration.minutes(5));
break;
case "PaymentReceived":
yield* ctx.updateState((s) => ({ ...s, status: "processing" }));
break;
case "Shipped":
yield* ctx.updateState((s) => ({ ...s, status: "shipped" }));
yield* ctx.schedule(Duration.hours(24));
break;
}
}),
execute: (ctx) =>
Effect.gen(function* () {
const state = yield* ctx.state;
if (!state) return;
if (state.status === "pending") {
yield* sendPaymentReminder(state.orderId);
yield* ctx.schedule(Duration.minutes(30));
}
if (state.status === "shipped") {
const delivered = yield* checkDelivery(state.orderId);
if (delivered) {
yield* ctx.updateState((s) => ({ ...s, status: "delivered" }));
yield* ctx.terminate();
} else {
yield* ctx.schedule(Duration.hours(24));
}
}
}),
onIdle: (ctx) =>
Effect.gen(function* () {
yield* ctx.schedule(Duration.hours(1));
}),
onError: (error, ctx) =>
Effect.gen(function* () {
yield* Effect.logError("Task failed", error);
yield* ctx.updateState((s) => ({ ...s, attempts: s.attempts + 1 }));
yield* ctx.schedule(Duration.seconds(30));
}),
});
Configuration
Effect Schema for validating and serializing the task’s persistent state. State is null until first set via ctx.setState.
Effect Schema for validating incoming events. Each event sent via client.task(...).send() is validated against this schema before being passed to onEvent.
onEvent
(event: E, ctx: TaskEventContext<S>) => Effect<void, Err, never>
required
Handler called for each incoming event. The event is passed as the first argument (not on ctx) — it is a direct value, not an Effect. Update state and optionally schedule execution.
execute
(ctx: TaskExecuteContext<S>) => Effect<void, Err, never>
required
Handler called when the scheduled alarm fires. Process state and schedule the next execution or call ctx.terminate() when done.
onIdle
(ctx: TaskIdleContext<S>) => Effect<void, never, never>
Optional handler called after onEvent or execute completes with no alarm scheduled. Use it to schedule delayed cleanup or trigger maintenance.
onError
(error: Err, ctx: TaskErrorContext<S>) => Effect<void, never, never>
Optional error handler for onEvent and execute failures. If not provided, errors are logged and the task continues. Use to track errors in state and schedule retries.
Control logging verbosity. false (default) logs only errors. true enables debug-level logging.
onEvent context
The event value is passed as the first argument to onEvent, separate from ctx. This makes it clear the event is a plain value, not an Effect.
| Property | Type | Description |
|---|
state | Effect<S | null> | Current state. null if no state has been set yet. |
setState(s) | Effect<void> | Replace the entire state. |
updateState(fn) | Effect<void> | Transform state. No-op if state is null. |
schedule(when) | Effect<void> | Schedule the next execution. |
cancelSchedule() | Effect<void> | Cancel any scheduled execution. |
getScheduledTime() | Effect<number | null> | Get the currently scheduled time (ms), or null if none. |
terminate() | Effect<never> | Cancel alarms and delete all state. Short-circuits execution. |
instanceId | string | Unique Durable Object instance ID. |
jobName | string | Job name as registered. |
isFirstEvent | boolean | true if this is the first event (state was null). |
eventCount | Effect<number> | Total events received. |
createdAt | Effect<number> | When this task instance was created (ms). |
Execute context
| Property | Type | Description |
|---|
state | Effect<S | null> | Current state. |
setState(s) | Effect<void> | Replace the entire state. |
updateState(fn) | Effect<void> | Transform state. |
schedule(when) | Effect<void> | Schedule the next execution. |
cancelSchedule() | Effect<void> | Cancel any scheduled execution. |
getScheduledTime() | Effect<number | null> | Get the currently scheduled time (ms). |
terminate() | Effect<never> | Cancel alarms and delete all state. Short-circuits execution. |
instanceId | string | Unique Durable Object instance ID. |
jobName | string | Job name as registered. |
executeCount | Effect<number> | Number of times execute has been called (1-indexed). |
eventCount | Effect<number> | Total events received. |
createdAt | Effect<number> | When this task instance was created (ms). |
Scheduling execution
ctx.schedule(when) accepts flexible time inputs:
// Duration (from now)
yield* ctx.schedule(Duration.seconds(30));
yield* ctx.schedule("5 minutes");
// Absolute timestamp (ms since epoch)
yield* ctx.schedule(Date.now() + 60000);
// Date object
yield* ctx.schedule(new Date("2024-12-31"));
Canceling and inspecting a schedule
// Cancel the current alarm
yield* ctx.cancelSchedule();
// Read the scheduled time
const scheduledAt = yield* ctx.getScheduledTime();
if (scheduledAt !== null) {
console.log("Next run at:", new Date(scheduledAt));
}
Terminating a task
ctx.terminate() cancels any scheduled alarm, deletes all state from storage, and short-circuits the current handler. The instance ID can be reused to start a new task from scratch.
execute: (ctx) =>
Effect.gen(function* () {
const state = yield* ctx.state;
if (!state) return;
if (state.status === "delivered") {
// All done — clean up
yield* ctx.terminate();
}
// Code here never runs after terminate()
}),
Complete example
import { Effect, Schema, Duration } from "effect";
import { Task, createDurableJobs } from "@durable-effect/jobs";
const orderProcessor = Task.make({
stateSchema: Schema.Struct({
orderId: Schema.String,
status: Schema.Literal("pending", "processing", "shipped", "delivered"),
attempts: Schema.Number,
}),
eventSchema: Schema.Union(
Schema.Struct({ _tag: Schema.Literal("OrderPlaced"), orderId: Schema.String }),
Schema.Struct({ _tag: Schema.Literal("PaymentReceived") }),
Schema.Struct({ _tag: Schema.Literal("Shipped"), trackingNumber: Schema.String }),
),
onEvent: (event, ctx) =>
Effect.gen(function* () {
switch (event._tag) {
case "OrderPlaced":
yield* ctx.setState({
orderId: event.orderId,
status: "pending",
attempts: 0,
});
yield* ctx.schedule(Duration.minutes(5));
break;
case "PaymentReceived":
yield* ctx.updateState((s) => ({ ...s, status: "processing" }));
break;
case "Shipped":
yield* ctx.updateState((s) => ({ ...s, status: "shipped" }));
yield* ctx.schedule(Duration.hours(24));
break;
}
}),
execute: (ctx) =>
Effect.gen(function* () {
const state = yield* ctx.state;
if (!state) return;
if (state.status === "pending") {
yield* sendPaymentReminder(state.orderId);
yield* ctx.schedule(Duration.minutes(30));
}
if (state.status === "shipped") {
const delivered = yield* checkDelivery(state.orderId);
if (delivered) {
yield* ctx.updateState((s) => ({ ...s, status: "delivered" }));
yield* ctx.terminate();
} else {
yield* ctx.schedule(Duration.hours(24));
}
}
}),
onIdle: (ctx) =>
Effect.gen(function* () {
yield* ctx.schedule(Duration.hours(1));
}),
onError: (error, ctx) =>
Effect.gen(function* () {
yield* Effect.logError("Task failed", error);
yield* ctx.updateState((s) => ({ ...s, attempts: s.attempts + 1 }));
yield* ctx.schedule(Duration.seconds(30));
}),
});
const { Jobs, JobsClient } = createDurableJobs({ orderProcessor });
export { Jobs };
