Do slow work outside the database.

Pick a pattern

A worker is your own process that does slow or external work. A live query tells it what to do; a mutation decides whether to accept the result. Nothing new runs inside the database.

Results land in ordinary collections. To run and scale workers, see Worker pools.

Keep a result current

Store each result with the exact input that produced it. When the source changes, the old result stops counting immediately.

The complete example keeps a document’s SHA-256 digest current.

const desired = derive("worker.desired", (ctx, id: string) => {
  const document = ctx.get(documents, id);
  if (!document) return null;
  const input = { recipe: "sha256-v1", text: document.text };
  return { key: canonicalJson([id, input]), input };
});

const pending = query("worker.pending", (ctx, id: string) => {
  const task = ctx.get(desired, id);
  return task && ctx.get(results, id)?.key !== task.key ? task : null;
});

• Put everything that affects the result in the input: source values, recipe, model and prompt versions, locale, config, tenant. Don’t use a global database revision.
• Key it with canonical JSON, or a hash for large inputs.
• The input must not depend on its own result. Store outside facts as versioned records; Flower can’t react to changes it never sees.

Check freshness when publishing

const publish = mutation("worker.publish", (ctx, args: {
  id: string; key: string; result: string;
}) => {
  const task = ctx.get(desired, args.id);
  if (!task || task.key !== args.key) return { accepted: false };
  if (ctx.get(results, args.id)?.key !== task.key) {
    ctx.set(results, args.id, { key: task.key, result: args.result });
  }
  return { accepted: true };
});

The check and write commit together: a stale result gets accepted: false, and among racing workers the first wins. Downstream code must read the guarded result, not the raw record:

const currentResult = derive("document.currentResult", (ctx, id: string) => {
  const task = ctx.get(desired, id);
  const stored = ctx.get(results, id);
  return task && stored?.key === task.key ? stored.result : null;
});

• null means “not ready” here. If a result can be null, use {status: "ready", value: null}.
• If the input goes A → B → A, the stored A result counts again. Add a generation number to force fresh work.

Try it: keep a digest current

Run the document reconciler against a fresh local app. It needs Node.js and no API keys.

Start a local node, then run from the checkout:

npm run build
node sdk/cli.ts deploy docs/reactive-worker.ts
node sdk/cli.ts call document.put '{"id":"notes","text":"Hello, garden."}'   --request-id notes-v1
node sdk/cli.ts watch document.get '"notes"' 

node docs/reactive-worker-client.ts notes
# FLOWER_URL defaults to http://127.0.0.1:7101.
# Start the same command in another terminal to exercise competing workers.

node sdk/cli.ts call document.put '{"id":"notes","text":"The garden has changed."}'   --request-id notes-v2

The view shows a digest only for the current text. Stop all workers, edit, and restart one: it catches up from its first snapshot. Download the application and client.

To run a pool of queue workers instead, see Worker pools.

Complete every action

When every action matters, create a job in the same mutation as the business change. Workers lease jobs, do the work, and record the outcome.

For example, the mutation that finalizes an invoice also calls jobs.enqueue(ctx, operationId, payload), so a committed invoice always has its receipt job. Put the full payload in the job.

examples/workers.ts exposes jobs.enqueue, jobs.ready, jobs.claim, jobs.complete, jobs.fail, jobs.retry and jobs.backlog. Its readiness query checks for pending jobs and expired leases:

const ready = query("internal.jobs.ready", (ctx) => {
  const pending = ctx.range(jobs.records.by("pending").range({
    prefix: ["", "pending"], limit: 1,
  }));
  const expired = ctx.range(jobs.records.by("leased").range({
    prefix: ["", "leased"], lte: ctx.now(), limit: 1,
  }));
  return pending.rows.length > 0 || expired.rows.length > 0;
});

Add jobs.records to define({collections}). The watch refreshes as time passes, so a crashed worker’s job wakes another worker with no new writes.

1. Wake and claim. When the watch reports ready, call jobs.claim. Claim only what you can start now; readiness reserves nothing.
2. Do the work. Check the lease deadline first. Send a stable business ID as the idempotency key.
3. Finish through Flower. Complete or fail with the lease exactly as returned. The same mutation can update business records.
4. Drain and re-arm. Keep claiming until a claim returns null, then open a new readiness watch.

const identity = {
  id: lease.id, owner: lease.owner, token: lease.token,
  ...(lease.history ? { history: lease.history } : {}),
};
// Keep the same arguments and request ID on uncertain replies.
await client.mutate("jobs.complete", {
  ...identity, result: { delivered: true },
}, { requestId: completionId });

• Request IDs: each new claim needs a new request ID, even after a null. When a reply is uncertain, retry with the same ID and arguments. The returned lease may already be expired.
• Idempotency: use the business operation’s ID across retries and leases. A lease token identifies one attempt, not the operation. Lease expiry cannot stop a paused process, so the receiving service must deduplicate or fence. If it fences, keep enforcing after a disaster restore.
• Lease length: leases can’t be renewed. Keep tasks shorter, or split them.
• Retries: backoff and retry limits are up to you; a timer can call jobs.retry. Look up an unknown outcome by business ID before calling it failed.