Jobs#

defineJob is the packaged experience on top of tasks. Start a job from a request handler and return immediately; check on it from anywhere in the cluster.

import { defineJob } from '@sigx/actors/job';

export const SecuritySync = defineJob({
    type: 'SecuritySync',
    authorize: requireStaff,
    maxAttempts: 3,          // crash-resume attempts before 'failed'
    retainMs: 86_400_000,    // keep the terminal record a day, then forget
    run: async (job, input: { providerId: string }) => {
        const users = await loadUsers(input.providerId, { signal: job.signal });
        const from = (job.resumedFrom as { cursor: number } | undefined)?.cursor ?? 0;
        for (let i = from; i < users.length; i++) {
            job.signal.throwIfAborted();
            await syncOne(users[i]);
            await job.progress({ done: i + 1, total: users.length });
            if (i % 100 === 0) await job.checkpoint({ cursor: i + 1 });
        }
        return { synced: users.length };
    },
});

// A request handler — returns immediately, the job runs on the cluster:
const runId = crypto.randomUUID();
await actor(SecuritySync, runId).start({ providerId });

// Later, from anywhere:
await actor(SecuritySync, runId).status();   // JobInfo: status/progress/attempts
await actor(SecuritySync, runId).cancel();
await actor(SecuritySync, runId).result();   // the return value, once completed

watch() is a stream of JobInfo, so a progress bar is a component that consumes it:

import { component, onMounted, signal } from 'sigx';

const JobProgress = component<{ runId: string }>(({ props }) => {
    const info = signal<JobInfo | null>(null);

    onMounted(async () => {
        for await (const next of actor(SecuritySync, props.runId).watch()) info.value = next;
    });

    return () => (
        <progress value={info.value?.progress?.done ?? 0} max={info.value?.progress?.total ?? 1} />
    );
});

Throttle the feed for consumers that redraw#

watch({ throttleMs }) coalesces per subscriber: at most one JobInfo per window, leading edge plus a trailing emit taken fresh, so a watcher is never staler than the window. It forwards ctx.changes()'s throttle exactly, validation included.

for await (const info of actor(SecuritySync, runId).watch({ throttleMs: 250 })) render(info);

Set it whenever the consumer redraws rather than accumulates — a progress bar, a status tile. An unthrottled watcher costs one whole-state snapshot per mutating turn, and a job that reports progress per step over state that grows through the run pays that on every step: on such a job the unthrottled feed measured at about 2.5× the per-step cost, and the throttled one at the unwatched floor.

It never loses the end of the run. A window still owing an emit when the actor deactivates is flushed, so a throttled watcher always sees the terminal JobInfo before the feed ends. Omitting the option keeps one JobInfo per mutating turn.

What the layer decides for you#

The state machine. status() and watch() return JobInfo — never the checkpoint, which is private, and never the result, which is fetched once via result().

Reads cost O(1), not O(state). status(), JobControl.info, the onSettled argument and the info returned by start/cancel/resume are built from live state without cloning the whole record. JobInfo.progress on that path is a shallow copy of its declared all-primitive shape; the watch() feed is a full snapshot. A job checkpointing a growing table keeps its status reads flat however large the state gets.

stateDiagram-v2
    [*] --> pending: start(input)
    pending --> running
    running --> paused: job.pause(cp)
    paused --> running: resume(data)
    running --> running: crash-resume
    running --> completed: run() returns
    running --> failed: past maxAttempts
    running --> cancelled: cancel()
    paused --> cancelled: cancel()
    completed --> [*]: retainMs
    failed --> [*]: retainMs
    cancelled --> [*]: retainMs

One actor per run, keyed by your run id. The directory's single-activation guarantee is the "exactly one runner" guarantee — there is no separate lock.

start is idempotent under retry. A non-pending job returns its current info and never restarts. Safe to call from a handler that a client may retry.

A rejected start or resume is retryable. start() resolves once the run is durably recorded and launched, not when it finishes. If the launch itself rejects — task liveness cannot reach storage, the host is mid-shutdown — the transition is taken back, live and on disk: a rejected start() leaves the job pending, and a rejected resume() leaves it paused with that call's resume data dropped. Nothing resumes the run behind your back, and the same call can simply be retried. The caller always sees the original rejection. The revert is best-effort: if its own save fails too, the record stays running and the run resumes on the next activation, at-least-once.

The job's state record is its ledger. A job keeps no separate task-ledger entry: the run to resume, its input and its attempt count come from the job's own status, input and attempts, which it persists anyway, so a crash at any point resumes from that one record. With the default task liveness, a run's whole lifecycle costs one load and four saves, and no clears. The $sigx:tasks ledger record belongs to plain defineActor + tasks: only.

Crash-resume counts; pause-resume is free. A crash-resumed run arrives with job.attempt bumped and job.resumedFrom set to the last checkpoint. Past maxAttempts the job is marked failed. resume(data) on a paused job re-runs with job.resumeData and costs no attempt.

pause parks durably. return job.pause(checkpoint) writes the checkpoint, marks paused and releases the task — the actor idles at zero cost until resume(). For a timeout, arm job.reminders before pausing and handle it in onReminder(control, name); control.resume() and control.cancel() are internal, so there is no self-dispatch deadlock.

Progress rides the change feed, not storage. job.progress() — and job.update() for your own state: extra fields — mutate state in a turn so watch() pushes them live, but nothing is persisted per tick. After a crash, progress honestly regresses to the last checkpoint rather than claiming ground the work did not keep.

A job run's storage footprint is one load and four saves. Crash-resume liveness rides the host's task roster, so a running job costs nothing periodic.

retainMs keeps the terminal record around for late result() readers, then a one-shot reminder clears the state and deactivates. discard() does it on demand.

Checkpoints are the contract#

progress() is for humans; checkpoint() is for the runtime. Only a checkpoint survives a crash, so how often you call it is how much work you are willing to redo.

The resume contract is at-least-once: the runtime resumes the function, and your code resumes the work by reading job.resumedFrom. Make the step after a checkpoint idempotent.

job.resumedFrom is a detached copy: mutating it changes nothing, and nothing changes it.

Append per step, checkpoint every N#

job.checkpoint(cp) re-encodes the whole checkpoint, so on a job whose checkpoint grows through the run — rows accumulated per page, say — each step costs more than the last. job.append writes the step instead. Declare apply, the reducer that folds one entry into the checkpoint:

const Sync = defineJob({
    type: 'Sync',
    apply: (rows: Row[] | undefined, row: Row) => {
        (rows ??= []).push(row);
        return rows;
    },
    run: async (job, input: Input) => {
        const rows = job.resumedFrom ?? [];      // a detached copy — yours to keep folding
        for (let page = rows.length; page < input.pages; page++) {
            const row = await fetchPage(input, page);
            rows.push(row);                      // your fold…
            await job.append(row);               // …and the durable one: O(row), not O(rows)
            if (page % 100 === 99) await job.checkpoint(rows);   // compaction every N
        }
        return rows.length;
    },
});

apply(checkpoint, entry) returns a value to replace the checkpoint, or nothing to say it was mutated in place. A checkpoint that is still undefined — no checkpoint() or append() yet — can only be created by returning one. It runs at every append and again on every activation, replaying the appended entries onto the stored checkpoint, so it must be a pure function of its two arguments.

Each append is durable when it resolves, etag-CAS like every write. After a crash, resumedFrom is the last checkpoint with every appended entry folded in, in order — the same value the body saw. checkpoint(), pause() and the terminal transition are full saves, and therefore the compaction points: checkpointing every N steps bounds what a resume replays. Measured on the jobs/checkpoint-growth bench, an append per step stays flat at roughly 8–10 µs from the head of a 300-step run to its tail, where a checkpoint per step grows about five-fold.

The entry type E is inferred from apply — defineJob, JobOptions and JobHandle carry it as a trailing type parameter — so without apply, job.append takes never and cannot be called. A late append from a body that is winding down is dropped, like a late progress(). On a storage without appendText (fileStorage) every append is a full save: the recipe still holds, at O(checkpoint) cost.

Deferring a checkpoint#

await job.checkpoint({ cursor: i + 1 }, { durability: 'eventual' });

checkpoint(cp, options?) takes the same SaveOptions as ctx.save() (re-exported from @sigx/actors/job). 'immediate', the default, is durable when the promise resolves — one encode and one etag-CAS per step. 'eventual' marks the state for the host's write-behind debounce (50 ms) and resolves at once, so a burst of N steps costs one save. pause(), the terminal transition and a graceful deactivation still save synchronously, so the record that decides a job's fate is never eventual.

The trade is crash distance. A host death between an eventual checkpoint and its flush resumes from the last checkpoint that reached storage, which may be several steps back. Use it for steps that are cheap to redo and idempotent; keep 'immediate' for a step that commits an external side effect. On Cloudflare, where eviction is not deactivation, there is no final flush at all — prefer 'immediate' there.

Projecting status outside the job#

onSettled fires on every terminal transition, which is what makes it safe to keep a projection — a status row in your own database, a metric, a notification — in step with the job:

export const ImportJob = defineJob({
    type: 'Import',
    async run(job) { /* … */ },
    async onSettled(control, info) {
        await db.imports.update(info.key, { status: info.status });
    },
});

Two of those transitions cannot be observed from inside run() at all, which is the reason the hook exists:

• The maxAttempts give-up. The body is re-entered, sees attempts > maxAttempts, and finishes as failed without ever calling your run — so there is no body turn to write from.
• cancel() on a paused job. A paused job holds no task, so nothing in the body is around to notice.

Without the hook, a projection written only from run() asserts "still running" forever after either one.

The contract:

• It runs after the save. The transition is already durable and info is the final state. A handler that throws is caught, logged in dev, and swallowed — it cannot unwind the settle.
• It fires for completed too, not only the runtime-driven cases, so handlers should be idempotent.
• No self-dispatch, the same rule as onReminder: it receives a JobControl, not a client.
• Awaiting slow I/O holds the actor's turn. Hand long work to a task rather than blocking the settle.

migrateState is not supported#

defineJob does not accept migrateState. JobOptions has no such option and the underlying defineActor config forwards nothing.

The reasoning is worth a sentence: a job's stored record is the job envelope — status, progress, checkpoint and the rest, with your own state under extra. A hook over that would hand you a runtime shape you do not own. Migrating the extra half wants its own option, and is not part of this.

See State & persistence for migration on ordinary actors.

Recipes, not API#

A singleton queue-worker with strict ordering and bounded concurrency, a cron-on-reminders scheduler, and the Cloudflare Durable Object posture are all patterns rather than features. They are written up in docs/job-recipes.md in the actors repo, deliberately left out of the API so the job layer stays one thing done well.

Next steps#

• Tasks — the primitive underneath, and the crash-resume machinery.
• Timers & reminders — pausing with a timeout.
• Clustering — where a resumed job comes back.