Work queues
queue from @flower-js/sdk/temporal hands each job to one worker at a time, retries failures, and rejects results from a worker whose lease was lost. It is a component. See the queues guide for a walkthrough.
| API | Contract |
|---|---|
queue<P = Json, R = Json>(name, options?): Queue<P, R> | Creates a queue; add it to define({ uses }). P and R come from the payload and result schemas when given. Names starting with $flower. are reserved. |
QueueOptions<P, R> | Optional lease: { defaultMs, maxMs } (30,000 and 300,000), retry, and payload and result schemas, which check every enqueue and completion (INVALID_ARGUMENT). |
QueueRetry | maxAttempts (5), initialDelayMs (1,000) and maxDelayMs (60,000). After attempt n fails, the job waits min(maxDelayMs, initialDelayMs · 2^(n−1)). retry: false makes every failure final. |
Queue<P, R> | A component, plus the QueueView methods for the default scope "", name, records, scope and http. |
QueueView<P, R> | The job methods below, for one scope. Call the writing ones from a mutation. |
enqueue(ctx, id, payload, { delayMs?, at?, replace? }?): Job | Adds a pending job, available now, after delayMs, or at at (not both). Fails with JOB_EXISTS while the ID is pending or leased; replace: true replaces a completed or failed job. |
claim(ctx, owner, { leaseMs? }?): Claim | null | Leases the job that has waited longest, including jobs whose lease expired, with the next fencing token, and counts an attempt. Null when nothing is ready. LEASE_TOO_LONG above lease.maxMs. |
renew(ctx, lease, { leaseMs? }?): Claim | Extends a current lease from now. The token stays the same. |
complete(ctx, lease, result): Job | Stores the result if the lease is still current and unexpired. |
fail(ctx, lease, error, { retry?, delayMs? }?): Job | Same check, then requeues the job after the backoff, or delayMs. With retry: false, a disabled policy or no attempts left, the job is failed with error. |
retry(ctx, id, { delayMs? }?): Job | Moves a failed job back to pending with a fresh attempt count. Other states fail with JOB_NOT_FAILED. |
cancel(ctx, id): boolean | Deletes the job, in any state. Returns whether it existed. |
get(ctx, id): Job | null | The job’s current state. An expired lease shows as pending, or failed when no attempts are left, with a LEASE_EXPIRED error. |
scan(ctx): Job[] | Every job in this scope, by ID. |
ready(ctx): boolean | Whether a claim would succeed now. It changes as time passes, so a watch on it wakes workers when a lease expires. |
stats(ctx): QueueStats | Readiness, backlog age and the next wake-up. |
queue.records | The backing collection, keyed [scope, id] and shared by all scopes. Writing it directly can break the queue. |
queue.scope(name): QueueView | The same queue for one namespace, with its own fencing tokens. Scopes don’t restrict access. |
queue.http(prefix, options?): QueueHttp | Public methods to spread into define({ http }), named `${prefix}.claim` and so on. Default set: claim, renew, complete, fail, get, ready, stats. Arguments are checked: { owner, leaseMs? }, a lease identity plus result or error, { id }; ready and stats take null. |
QueueHttpOptions | Optional methods (add enqueue, retry or cancel here), scope ("argument" adds a required scope argument; a function of ctx picks it, for example from ctx.principal()) and access for every generated method. |
QueueMethodName | "enqueue" | "claim" | "renew" | "complete" | "fail" | "retry" | "cancel" | "get" | "ready" | "stats". |
QueueMethods<P, R>, QueueHttp<Prefix, P, R, M> | The types of the generated methods, keyed by name and by alias. |
QueueStats | ready; oldestReadyAt, when the longest-waiting ready job became available; nextAvailableAt, when a delayed job or running lease next makes work available. |
Lease | owner, token, expiresAt and, once retention is initialized, history. |
LeaseIdentity | id, owner, token, and optional history. Pass it unchanged from the claim. |
Claim<P = Json> | A Lease plus scope, id, payload and attempt. |
Job<P = Json, R = Json> | The stored job: scope, id, payload, state (pending, leased, completed or failed), availableAt, leaseExpiresAt, lease, attempts, timestamps, result and error. |
Per-tenant scopes chosen by the caller
import { define, v } from "@flower-js/sdk";
import { queue } from "@flower-js/sdk/temporal";
const renders = queue("renders", {
payload: v.object({ url: v.string() }),
result: v.object({ bytes: v.int({ min: 0 }) }),
lease: { defaultMs: 15_000, maxMs: 60_000 },
});
export default define({
uses: [renders],
http: renders.http("renders", { methods: ["enqueue", "claim", "renew", "complete", "fail", "ready"], scope: "argument" }),
});
// Workers: runQueueWorker(client, { queue: "renders", scope: "tenant-a", work, signal })- A stale, expired or replaced lease fails with
LEASE_LOST. An expired lease counts as a failed attempt. - A claim retried with the same request ID returns its original lease, which may have expired already. Check
expiresAtbefore starting work. - Flower can’t make external side effects exactly-once. Give external systems the job ID as an idempotency key, or the token as a fence.
- Completed and failed jobs stay until you cancel or replace them.
- Pass a
scopeargument only to methods generated withscope: "argument"; elsewhere it is rejected as an unexpected property.
Expiring records
expiringCollection from @flower-js/sdk/temporal stores records that stop being readable after a deadline. It is a component.
| API | Contract |
|---|---|
expiringCollection<T = Json>(name, { expiration?, value? } = {}) | Creates the collection; add it to define({ uses }). Default expiration is null. A value schema checks every set. Names starting with $flower. are reserved. |
ExpiringCollection<T> | A component, plus the methods below. |
Expiration | null (never), {at}, {afterCreationMs} or {afterUpdateMs}. Nonnegative integer milliseconds. |
ExpiringEntry<T> | value, createdAt, updatedAt and expiresAt (or null). |
records: Collection<ExpiringEntry<T>> | The backing collection. Reading it directly can return expired entries; use the methods instead. |
entry(ctx, key): ExpiringEntry<T> | null | The live entry with timestamps, or null once expired. |
get(ctx, key): T | null | The live value, or null. |
scan(ctx): Row<T>[] | All live records. Cost includes expired rows not yet deleted. |
set(ctx, key, value, expiration?): ExpiringEntry<T> | Writes with this call’s expiration or the default. Keeps createdAt if the old entry is still live; an expired key starts fresh. |
delete(ctx, key): void | Removes the record, live or expired. |
- A record is expired when
now >= expiresAt. Reads enforce this immediately. - A maintenance task deletes expired records as they come due, in pages of 64.
- Expiry doesn’t run your code. To act at a deadline, use the scheduler.
Delayed callbacks
scheduler from @flower-js/sdk/scheduler runs a mutation after a deadline. The timer commits in the same transaction as the write that schedules it. It is a component.
| API | Contract |
|---|---|
scheduler(name, handlers, options = {}): Scheduler | Creates timers; add them to define({ uses }). handlers maps names to mutations, which don’t need to be public. |
SchedulerOptions | Optional maxAttempts (3), retryDelayMs (1,000) and maxRetryDelayMs (60,000). |
Scheduler<Handlers> | A component, plus the methods below. Handler names and their argument types are checked. |
ScheduledTimer<A = Json> | state (pending or failed), handler, args, dueAt, attempts, error, createdAt, updatedAt. |
Timer<A = Json> | A ScheduledTimer with its id. |
records: Collection<ScheduledTimer> | The backing collection. Don’t write it directly. |
after(ctx, id, delayMs, handler, args): Timer | Schedules delayMs after ctx.now(). Reusing an ID replaces the earlier timer (debounce) and resets its attempts. Arguments that fail the handler’s schema fail now, with INVALID_ARGUMENT. |
at(ctx, id, dueAt, handler, args): Timer | Schedules at an absolute time in milliseconds. A past time runs at the next chance. |
get(ctx, id): Timer | null | Reads one timer. |
scan(ctx, { state? } = {}): Timer[] | Lists timers, optionally one state, ordered by dueAt then ID. |
cancel(ctx, id): boolean | Deletes a timer. Returns whether it existed. |
retry(ctx, id, delayMs = 0): Timer | Reschedules a failed timer with a fresh attempt count, using the currently deployed handler. Other states fail with TIMER_NOT_FAILED. |
The order and its timer commit together
import { collection, define, mutation, v } from "@flower-js/sdk";
import { scheduler } from "@flower-js/sdk/scheduler";
const orders = collection("orders", v.object({ state: v.enum(["baking", "ready"]) }));
const bake = mutation("internal.bake", { args: v.string() }, (ctx, id) => {
if (ctx.get(orders, id)) ctx.set(orders, id, { state: "ready" });
return null;
});
const oven = scheduler("oven", { bake });
const place = mutation("orders.place", { args: v.string({ min: 1 }) }, (ctx, id) => {
ctx.set(orders, id, { state: "baking" });
oven.after(ctx, `bake:${id}`, 5_000, "bake", id);
return { id };
});
export default define({ uses: [oven], http: { place } });- A deadline means “not before”. Load, elections and outages can delay a timer.
- If the handler succeeds, its writes and the timer’s removal commit together.
- If it fails, its writes roll back and it retries with doubling delay, up to
maxAttempts. Then it stays as a failed timer, with the error, that you can inspect and retry. - A timer whose handler is no longer deployed fails with
SCHEDULER_HANDLER_MISSING. - A handler can run more than once before it commits. Put external side effects in a queue.
- By default the leader checks for due timers every 250 ms. That’s an operator setting.