Durable steps

A durable step is a checkpoint inside a single task body. step.run executes a piece of work once, commits its result next to the job, and every later attempt of that job returns the committed value instead of running it again. step.sleep ends the attempt entirely and brings the job back later, having replayed everything that already ran.

from flexiq import Queue, current_job

queue = Queue(db_path="tasks.db")

@queue.task(max_retries=3)
def checkout(order_id: str) -> str:
    step = current_job.step
    charge = step.run(
        "charge",
        lambda: stripe.charge(order_id, idempotency_key=step.idempotency_key),
    )
    step.sleep("1h")                                   # attempt ends here
    step.run("receipt", lambda: send_receipt(charge))  # runs an hour later
    return charge["id"]

import { currentJob } from "@byteveda/flexiq";

queue.task(
  "checkout",
  async (orderId: string) => {
    const { step } = currentJob()!;
    const charge = await step.run("charge", () =>
      stripe.charge(orderId, { idempotencyKey: step.idempotencyKey }),
    );
    await step.sleep("1h"); // attempt ends here
    await step.run("receipt", () => sendReceipt(charge)); // runs an hour later
    return charge.id;
  },
  { maxRetries: 3 },
);

@TaskHandler("checkout")
public String checkout(String orderId) throws Exception {
    StepContext step = JobContext.current().step();
    Charge charge = step.run("charge", Charge.class,
            () -> stripe.charge(orderId, step.idempotencyKey()));
    step.sleep(Duration.ofHours(1));            // attempt ends here
    step.run("receipt", () -> sendReceipt(charge));  // runs an hour later
    return charge.id();
}

Retry that job after the receipt fails and the card is not charged a second time: charge is a memo hit, and only receipt runs.

Steps or a workflow?

Both are durable multi-step execution, and they are not redundant. The difference is where the structure lives.

The rule: if the sequence is a property of the data and you want to look at it — as a graph, in the dashboard, resumed on another machine, with steps running in parallel — it is a workflow. If the sequence is a property of the code and you would have written it as an ordinary function anyway, it is a durable step.

A long ETL fan-out over 500 shards is a workflow. "Charge, wait an hour, email" is three lines with two step.run calls, and forcing it into a DAG buys nothing but three extra dispatches.

A step runs once per job, not once per attempt

Only a body that returns is memoized. One that throws commits nothing and runs again on the next attempt, and so does one whose process died before the commit.

A memo hit returns the value decoded from its stored bytes, which went through the queue's serializer — codec chain included, so an encrypting codec reaches the step store with no extra plumbing. Anything the serializer does not round-trip exactly comes back in its decoded shape. Return something the serializer preserves, or a handle to it.

Steps run one at a time. A step's position in the sequence is what identifies it, so a second step started while the first is uncommitted has no position to take: asyncio.gather over two arun callsPromise.all over two step.run callstwo steps issued from different threads fails the attempt rather than interleaving. Issue them in order.

Memoization is not exactly-once

This is the part that decides whether the page's opening example is actually correct.

Between "the payment API returned 200" and "the step row committed" there is an instant where the process can die. The replay has no record that the call happened, so it makes it again:

Nothing on this side of the network closes that window. The only thing that does is a key the other service dedupes on, and the only key that works is one this job mints the same way every time it runs. That is step.idempotency_keystep.idempotencyKeystep.idempotencyKey():

018f…c2:charge#0
└ run key ┘└ step ┘

Derived from the run's identity and the step's position, and from nothing else — no clock, no payload, no serializer, no codec. That is the contrast with the auto-derived dedup key, which hashes the serialized payload and therefore moves if a codec does.

charge = current_job.step.run(
    "charge",
    lambda: stripe.Charge.create(
        amount=order.total,
        idempotency_key=current_job.step.idempotency_key,  # 018f…c2:charge#0
    ),
)

const { step } = currentJob()!;
const charge = await step.run("charge", () =>
  stripe.charges.create(
    { amount: order.total },
    { idempotencyKey: step.idempotencyKey }, // 018f…c2:charge#0
  ),
);

StepContext step = JobContext.current().step();
Charge charge = step.run("charge", Charge.class, () ->
        stripe.charges().create(order.total(), step.idempotencyKey()));  // 018f…c2:charge#0

The key names the step that is running, so it is only readable inside a step body — outside one there is no step for it to name, and reading it there raises.

It is stable across an ordinary retry, across a step.sleep wake and across an operator's dead-letter retry from the dashboard. That last one mints a new job id, so the run key is carried on the new job rather than re-derived: retrying a dead-lettered charge three days later does not charge the customer twice.

Step identity

By default a step is identified by its name and how many times that name has already been used in this attempt: charge#0, fetch#0, fetch#1. The name is required and positional — never inferred from the callable, because an inferred name changes the moment a lambda is renamed or inlined.

name#occurrence is stable only while the surrounding code asks for the same names in the same order. That holds for a list. It does not hold for a set, an unordered query, or anything whose iteration order can move — reorder the items and fetch#1 now answers a different question, with a key sequence that looks identical. So pin identity to the data instead:

for order_id in order_ids:                 # iteration order is arbitrary
    current_job.step.run("fetch", lambda i=order_id: load_order(i), key=order_id)   # → fetch:1234

for (const orderId of orderIds) {
  // iteration order is arbitrary
  await step.run("fetch", () => loadOrder(orderId), { key: orderId }); // → fetch:1234
}

for (String orderId : orderIds) {           // iteration order is arbitrary
    step.run("fetch", Order.class, () -> loadOrder(orderId), StepOptions.key(orderId));  // → fetch:1234
}

A keyed step is matched by its key wherever it sits in the recorded sequence; an unkeyed one is matched at its position. The two forms count independently — a keyed call does not advance the name's occurrence counter:

The alternative would make adding a keyed call shift the key of every later unkeyed call of the same name — a divergence caused by an edit that changed nothing about the unkeyed steps.

Names are capped at 128 bytes and explicit keys at 256. Two steps deriving the same key in one attempt are refused, not silently merged.

Divergence: changing code while jobs are in flight

The recorded step sequence is compared against the running one at every step.run and step.sleep, in memory, against a snapshot loaded once at attempt start. Reach a position whose recorded key is not the key the code just asked for and the attempt fails:

step sequence changed for job 018f…c2 at position 2
  recorded: charge#0, notify#0, receipt#0
  running:  charge#0, notify#0, audit#0
  step 2 was 'receipt#0', now 'audit#0'
A memoized result would answer a different question than the step asking for it.
Drain or dead-letter this task's in-flight jobs before deploying a change to its
step sequence.

This failure is not retried. It goes straight to the dead-letter queue, because the code will not change between attempts and a retry would only burn the budget reproducing it. Handing back a wrong memoized result is worse than either.

Two changes are not divergences:

• Steps added at the end. The snapshot runs out, the new steps run and commit. That is just an attempt getting further than the last one.
• Steps removed from the end. The attempt finishes having consumed fewer steps than were committed. Logged as a warning naming both sequences, never a failure — the orphaned rows' side effects already happened, and the new code has no use for their values. The rows are dropped with the job.

Deploying a change to a task's step sequence is therefore a drain, not a rolling restart: let the in-flight jobs of that task finish (or dead-letter them) before the new code starts picking them up.

step.sleep ends the attempt

A sleep is not a sleep() call. The job's deadline is committed as a step row, the execution claim is released, and the job goes back to Pending at that instant. The worker moves on to other jobs immediately.

@queue.task()
def onboarding(user_id: str) -> None:
    step = current_job.step
    step.run("welcome", lambda: send_welcome(user_id))
    step.sleep("3d", name="cool_off")            # attempt ends
    step.run("nudge", lambda: send_nudge(user_id))
    step.sleep_until(next_billing_date(user_id), name="until_billing")
    step.run("invoice", lambda: invoice(user_id))

queue.task("onboarding", async (userId: string) => {
  const { step } = currentJob()!;
  await step.run("welcome", () => sendWelcome(userId));
  await step.sleep("3d", { name: "coolOff" }); // attempt ends
  await step.run("nudge", () => sendNudge(userId));
  await step.sleepUntil(nextBillingDate(userId), { name: "untilBilling" });
  await step.run("invoice", () => invoice(userId));
});

@TaskHandler("onboarding")
public void onboarding(String userId) throws Exception {
    StepContext step = JobContext.current().step();
    step.run("welcome", () -> sendWelcome(userId));
    step.sleep(Duration.ofDays(3), StepOptions.named("cool_off"));   // attempt ends
    step.run("nudge", () -> sendNudge(userId));
    step.sleepUntil(nextBillingDate(userId), StepOptions.named("until_billing"));
    step.run("invoice", () -> invoice(userId));
}

On wake the job replays from the top of the task body. welcome is a memo hit, the sleep is a memo hit and returns immediately, and execution carries on at nudge.

The deadline is fixed by the first commit. Replaying a "1h" sleep wakes at the original instant rather than an hour later each time, which is what stops a crash loop from producing a sleep that outlives the job. Reach for sleep_untilsleepUntilsleepUntil when the deadline means something outside the job — a billing date, a market open — because an absolute instant does not care how many times the attempt replayed.

Name your sleeps. An unnamed one is numbered sleep#0, sleep#1, sleep#2, and a divergence report reading step 2 was 'sleep#1' tells nobody which one moved.

Size caps

Step results live in the database and the whole set is loaded at attempt start, so they are capped. All three are enforced on the encoded bytes — post-serializer, post-codec — because that is what is stored, and a value that fits before encryption may not fit after.

Going over refuses the commit rather than spilling — there is nowhere to spill to, and the same database under a different key is not a spill, just the same bytes with less visibility:

step 'render#0' exceeds the step bytes limit: 1468006 > 262144

The answer is not a bigger cap. Write the value where it belongs — object storage, a table of your own — and memoize the handle:

# Wrong: the rendered PDF is not a checkpoint.
pdf = current_job.step.run("render", lambda: render_report(report_id))

# Right: the step's result is the pointer.
key = current_job.step.run("render", lambda: upload(render_report(report_id)))
current_job.step.run("email", lambda: send_link(key))

// Wrong: the rendered PDF is not a checkpoint.
const pdf = await step.run("render", () => renderReport(reportId));

// Right: the step's result is the pointer.
const key = await step.run("render", async () => upload(await renderReport(reportId)));
await step.run("email", () => sendLink(key));

// Wrong: the rendered PDF is not a checkpoint.
byte[] pdf = step.run("render", byte[].class, () -> renderReport(reportId));

// Right: the step's result is the pointer.
String key = step.run("render", String.class, () -> upload(renderReport(reportId)));
step.run("email", () -> sendLink(key));

Where the claim is not this process's

Every step write is fenced on the job's execution claim, so a step is only as durable as the thing holding it. Two situations where that is not the process running the task body:

• Attached executors. An executor holds no claim of its own, so the claim a step is fenced on is the scheduler's. It commits through the connection instead — the same rules, one round trip per step. A scheduler that offers no step store refuses rather than running the step un-memoized, retryably, because a fleet mid-rollout may place the next attempt somewhere that can commit.
• A lost claim. If a reaper or another worker takes the job over while a step is in flight, the commit is refused and the attempt reports a failure — which the scheduler then drops, because it fences on (owner, attempt) before mutating anything. The run proceeding elsewhere is untouched.

Task<String> checkout = Task.of("checkout", String.class);

try (FlexiQ queue = InMemoryFlexiQ.open()) {
    AtomicInteger charges = new AtomicInteger();
    String id = queue.enqueue(checkout, "go");

    try (Worker worker = queue.worker()
            .handle(checkout, payload -> {
                JobContext ctx = JobContext.current();
                ctx.step().run("charge", Integer.class, charges::incrementAndGet);
                return "ok";
            })
            .start()) {
        queue.awaitJob(id, Duration.ofSeconds(5));
    }

    assertEquals(1, charges.get()); // once per job, across every attempt
}

FlexiQ queue = FlexiQ.builder().sqlite(tempDir + "/steps.db").open();

Signals you cannot catch away

A step sleep and a step failure both unwind the task body, and neither may be swallowed by user code: a swallowed sleep runs the rest of the task with no claim, and a swallowed divergence returns a memoized answer to a different question.

So they are not ordinary errors.

And because that alone cannot stop a determined catch-everything handler, the worker also latches: a body that swallowed a signal and returned anyway fails the attempt regardless, naming what it swallowed.

Whether a step failure retries is decided by the core's classification, not by a task's retry filters — every worker path reads it before consulting them.