Transactions, or: How to Un-Write Half a Load

A flow that half-succeeds leaves your warehouse lying to you, and a retry only makes it worse. Prefect gives tasks commit-or-roll-back semantics — the one thing Airflow has no answer for at all.

Every failure mode this series has covered so far is a failure of execution. The task threw, the worker died, the API timed out — something didn’t finish, and you retried it or you cached past it or you let it fail loudly. In every one of those stories, the world outside your pipeline was still telling the truth. The load didn’t happen, so the table doesn’t claim it did.

This chapter is about the other kind of failure — the one where the pipeline stops halfway and the world is now wrong. The staging table holds rows that correspond to no completed load. The Parquet file sits in the warehouse prefix but the manifest was never updated. The vendor’s API accepted your “order shipped” call and then the database write that was supposed to record it blew up, so your system and theirs now disagree about reality. Nothing is running. Nothing is retrying. The pipeline is red and quiet, and the damage is already done.

Airflow’s answer to this is: write the cleanup yourself. There’s a cleanup task, a TriggerRule.ONE_FAILED to make it fire, and a long careful think about the exact set of upstream states in which it should and shouldn’t run — and if you get the trigger rule wrong you either clean up work that succeeded or leave behind work that didn’t. It’s not that Airflow does compensation badly. It’s that Airflow has no concept of compensation at all, so you build one out of graph edges every single time.

Prefect has the concept. It’s called a transaction, and it is the headline feature of Prefect 3.

The failure that retries make worse

Make it concrete with the bookshop load. Two tasks. stage transforms the day’s orders and writes them to a staging table. load copies staging into the warehouse table and updates a watermark. Between them, a fifteen-second window in which the staging table contains a write that the warehouse hasn’t consumed.

@task
def stage(orders: list[dict]) -> str:
    write_staging(orders)          # side effect #1
    return "staged"

@task
def load(_staged: str) -> None:
    load_warehouse()               # side effect #2 — and it fails
    advance_watermark()

load raises. Now think about what a retry does for you here, because the instinct — “it’s fine, it’ll retry” — is exactly wrong. The retry re-runs load, which runs load_warehouse() again, on top of a staging table that is still holding yesterday’s half-copied rows plus today’s. If load_warehouse isn’t idempotent, you’ve now double-loaded. If it is idempotent, congratulations: you’ve correctly loaded from a staging table that’s in an undefined state. The retry didn’t heal anything. It re-ran the second half of a two-half operation whose first half nobody ever agreed to keep.

What you actually want is the word every database engineer already knows: either both writes land or neither does. That’s not a retry policy. That’s a transaction, and until Prefect 3 no orchestrator gave you one that spanned tasks.

with transaction(): — the smallest version that works

Here is the whole idea in one block:

from prefect import flow, task
from prefect.transactions import transaction

@task
def stage(orders: list[dict]) -> str:
    write_staging(orders)
    return "staged"

@stage.on_rollback
def undo_stage(txn):
    clear_staging()

@task
def load(_staged: str) -> None:
    load_warehouse()

@flow
def daily_load(orders: list[dict]):
    with transaction():
        staged = stage(orders)
        load(staged)

Both tasks run inside with transaction():. If load raises, the transaction never commits — and every task that already succeeded inside that block fires its on_rollback hook on the way out. undo_stage runs, the staging table gets cleared, and the exception propagates to the flow as usual. The pipeline still fails. It just doesn’t leave a mess behind when it does.

@stage.on_rollback is a decorator that registers a compensating function against the task. It takes exactly one argument — the transaction — and Prefect calls it for you. You never call it. You never wire an edge to it. You never reason about trigger rules. The undo for a task lives next to the task, which is the only place it was ever going to stay correct.

That’s the pitch. Now let’s take the machine apart, because every single knob on it changes the failure semantics, and the defaults are only right until they aren’t.

commit_mode: when the work actually lands

A transaction has a commit_mode that decides when the commit — and therefore when the on_commit hooks, and the write of the staged result — actually happens. There are exactly three values, and I want to name them precisely because older material (including, for a while, this series) has been sloppy about it:

from prefect.transactions import CommitMode

CommitMode.EAGER
CommitMode.LAZY
CommitMode.OFF

That’s the complete list. If you’ve read somewhere that there’s an ATOMIC mode, there isn’t — CommitMode has three members and ATOMIC is not one of them. The atomicity people are reaching for when they say that word is what LAZY already gives you.

LAZY is the default, and it means: stage everything, commit nothing, and when the outermost with block exits cleanly, commit the whole tree at once. Every task inside stages its result and its hooks; nothing is finalized. Then the root exits, and the commit sweeps through — children first, then the root’s own commit hooks, then the write to the record store. This is the all-or-nothing you came for.

EAGER commits a transaction the instant its own block exits successfully, without waiting for anything above it. An eager transaction that finishes has cashed out for good. A later failure elsewhere in the flow — even in the parent that contains it — will not roll it back. Use it for a step that is genuinely independent and shouldn’t be held hostage to work that comes after it.

OFF is the one nobody mentions and it’s a genuinely useful escape hatch: a root transaction with commit_mode=CommitMode.OFF rolls back even on success. Read that again. If nobody takes responsibility for committing — no parent, and the mode says don’t — the exit path rolls the whole thing back. That is a dry-run switch. Run the transaction, do the work, fire the rollback hooks, undo it all, and confirm your compensation logic actually compensates. If you have ever wanted to test that your cleanup code works without waiting for a real 3 a.m. failure to find out, OFF is that test.

The inheritance rule matters as much as the values: a nested transaction inherits its parent’s commit mode unless you set one explicitly. A root with no commit_mode gets LAZY; every child under it gets LAZY too, which is precisely why the tree commits as a unit by default.

Nesting, and which way rollback flows

Transactions nest, and the nesting is where the model earns its keep:

@flow
def nightly():
    with transaction():              # root
        extract_and_stage()
        with transaction():          # child
            transform()
            load()                   # ← this raises

Walk the failure. load raises inside the child. The child’s __exit__ sees the exception, rolls itself back (running the rollback hooks of everything that succeeded inside it, transform included), and re-raises. The child then hands itself to its parent — and here’s the part worth memorizing: when a rolled-back child resets into its parent, it rolls the parent back too. The root’s rollback runs, extract_and_stage’s hook fires, and the exception continues up into your flow.

Rollback flows outward. A child cannot fail quietly inside a successful parent.

Two ordering details you’ll be glad to know before you’re debugging a compensation at midnight:

  • Rollback hooks run in reverse. The transaction unwinds its hooks LIFO, and rolls back its children in reverse order too. That’s the right semantics — you undo the last thing you did first — and it means your compensations compose like a stack, not like a list.
  • Commit runs forward, children first. On a successful root commit, children commit in order, then the parent’s own on_commit hooks fire, then the parent writes its record.

And the escape hatch: an EAGER child opts out of the whole arrangement. It commits when its own block ends, so when the root later fails, that child is already committed and is not rolled back. Here’s that difference, run for real:

@flow
def eager_child():
    with transaction():                                   # root, LAZY
        with transaction(commit_mode=CommitMode.EAGER):   # child, EAGER
            stage(4)
        load("x", boom=True)                              # root fails
BODY stage
HOOK stage.on_commit     ← child committed immediately, before load even ran
BODY load
flow caught: warehouse said no
                         ← and no rollback hook fires. The stage is kept.

Compare the same shape with a default (LAZY) child:

BODY stage
BODY load
HOOK stage.on_rollback   ← root failed, so the child never committed
flow caught: warehouse said no

Same code, one keyword apart, opposite outcomes. EAGER inside a LAZY root is a deliberate hole you punch in the atomicity, and it should be a decision you can defend in review, not something you copied off a blog.

Both halves of the hook pair: on_commit and on_rollback

on_rollback gets all the attention because it’s the flashy one, but a task has a matching on_commit hook, and the pair is what makes real compensation logic tractable. The pattern is older than Prefect and you already know it under another name: a write-ahead log. Do the reversible thing during the task. Do the irreversible thing only on commit. Undo the reversible thing on rollback.

from prefect import flow, task
from prefect.transactions import transaction, get_transaction

@task
def write_temp(orders: list[dict]) -> str:
    path = f"/staging/{today()}.parquet"
    write_parquet(path, orders)          # reversible: it's just a temp file
    get_transaction().set("tmp_path", path)
    return path

@write_temp.on_commit
def promote(txn):
    tmp = txn.get("tmp_path")
    move(tmp, tmp.replace("/staging/", "/warehouse/"))   # irreversible, and safe now

@write_temp.on_rollback
def discard(txn):
    delete(txn.get("tmp_path"))                          # never happened

@flow
def publish_day(orders: list[dict]):
    with transaction():
        path = write_temp(orders)
        validate(path)          # if this raises, the temp file is deleted
        register(path)          # if this raises, the temp file is deleted

The task body never performs the move. The move happens in on_commit, which only runs when the entire transaction succeeded — so validate and register both get a veto over whether the file is ever visible in the warehouse. If either one raises, discard deletes the temp file and the warehouse never knew anything was in flight.

Notice the timing, because it’s the whole point of LAZY: under the default commit mode, promote does not run when write_temp returns. It runs when the with block exits cleanly, after register. I ran this, and the trace is unambiguous:

BODY stage
BODY load
HOOK stage.on_commit      ← fired at block exit, not at task exit

The commit hook is your “now it’s safe to do the irreversible thing” callback: publish the file, flip the pointer, send the email, call the vendor’s confirm endpoint. Anything you would regret doing before you knew the rest of the block was going to work belongs there.

The scratch space between a task and its hooks

You saw txn.set() and txn.get() above and might have skimmed past them. Don’t — they solve a problem that otherwise pushes you straight back to global variables.

A rollback hook needs to know what to undo. Which file did the task write? Which row IDs did it insert? Which idempotency token did it send the vendor? That information exists inside the task body and nowhere else, and the hook is a separate function that Prefect calls later, possibly after the task’s stack frame is long gone. So the transaction doubles as a key/value scratch space that the task and its hooks share:

from prefect.transactions import get_transaction

@task
def charge_card(order_id: str, cents: int) -> str:
    charge_id = payments.charge(order_id, cents)
    txn = get_transaction()
    txn.set("charge_id", charge_id)     # the hook will need this
    return charge_id

@charge_card.on_rollback
def refund(txn):
    payments.refund(txn.get("charge_id"))

get_transaction() returns whichever transaction is currently active — the innermost one — so inside a task body it hands you exactly the transaction that will later call your hook. txn.set(name, value) stashes; txn.get(name) retrieves, with an optional default (without one, an unknown key raises rather than silently handing you None, which is the correct and slightly rude choice).

Two behaviors worth knowing because they will otherwise surprise you:

get() falls through to the parent. A child transaction that doesn’t have a key asks its parent for it. That makes the root a natural place to stash run-wide context — a batch id, a target prefix — that every nested compensation can read without you threading it through five function signatures.

get() deep-copies. Mutating what get() hands back does not update the stored value. If you pull a list out, append to it, and expect the transaction to have noticed, it hasn’t. Set it back explicitly:

rows = txn.get("row_ids", default=[])
rows.append(new_id)
txn.set("row_ids", rows)      # required — the mutation above touched a copy

This is a small paper cut with a good reason behind it (a shared mutable scratch space across nested transactions and async hooks would be a race condition factory), but it is a paper cut, and it’s the one people file bugs about.

isolation_level, and the lock manager it demands

Now the concurrency problem. Two runs — a scheduled one and a nervous human clicking “Run” — both enter a transaction keyed on today’s date at the same time. Both check the store, both see no committed record, both proceed, and both do the work. Whatever “at most once” meant, you didn’t get it.

isolation_level is the fix, and it has exactly two values:

  • READ_COMMITTED (the default) does no locking. It guarantees you won’t read a half-written record, and that’s all it guarantees. Two concurrent first-runs of the same key will both proceed. It’s fine when your transactions are naturally serialized — one scheduler, one deployment, non-overlapping runs.
  • SERIALIZABLE takes a lock on the transaction record for the life of the block. The second runner for the same key blocks at the with line until the first releases — at which point it sees a committed record and skips, or sees a rollback and takes its turn.

And here’s the part everybody gets wrong the first time: SERIALIZABLE requires a store with a lock manager, and Prefect will refuse to run without one. Not a warning. A ConfigurationError at __enter__:

ConfigurationError: Isolation level SERIALIZABLE is not supported by provided
configuration. Please ensure you've provided a lock file directory or lock
manager when using the SERIALIZABLE isolation level.

So you build the store yourself:

from pathlib import Path
from prefect import flow
from prefect.filesystems import LocalFileSystem
from prefect.locking.filesystem import FileSystemLockManager
from prefect.results import ResultStore
from prefect.transactions import transaction, IsolationLevel

@flow
def daily_load(day: str, orders: list[dict]):
    store = ResultStore(
        result_storage=LocalFileSystem(basepath="/srv/prefect/records"),
        lock_manager=FileSystemLockManager(
            lock_files_directory=Path("/srv/prefect/locks"),
        ),
    )

    with transaction(
        key=f"bookshop-load-{day}",
        store=store,
        isolation_level=IsolationLevel.SERIALIZABLE,
    ) as txn:
        if txn.is_committed():
            return                      # someone already did this day
        staged = stage(orders)
        load(staged)

FileSystemLockManager takes a lock_files_directory and it wants a Path, not a string. It coordinates every process that can see that directory — which on one box is every process, and across a fleet is nobody, because your workers don’t share a filesystem. For a real distributed setup you want RedisLockManager from prefect-redis, so the lock lives somewhere every worker can actually see it. A filesystem lock on ephemeral container storage is a lock that protects one container from itself, which is not the race you were worried about.

There’s also MemoryLockManager from prefect.locking.memory, which coordinates within a single process. Useful in tests, useful for threads inside one flow run, useless for anything that spans processes. Choose the lock manager whose blast radius matches the race you’re actually trying to prevent, and be honest that “we run on Kubernetes and use the filesystem lock manager” is a sentence that means “we have no lock.”

The record store: where a committed transaction actually lives

A transaction with a key writes a record when it commits, and that record is what makes the key mean anything on a later run. So: where does it go?

It goes in a ResultStore — the same abstraction from the results chapter. If you pass a key and no store, Prefect grabs the ambient result store from the run context, which by default is a LocalFileSystem rooted at ~/.prefect/storage. The record is a file. I checked: commit a transaction keyed ser-2 against a store rooted at ./.res, and afterwards ./.res/ contains a file named ser-2. That’s the whole magic trick. The transaction record store is your result store, and a transaction record is a result keyed by your string.

Which produces two consequences you should hold onto:

The idempotency is exactly as durable as the storage. Local disk means the record dies with the box. Point result_storage at S3 or GCS and the “we already loaded 2026-06-04” fact survives a redeploy, a new worker, a different region. If your transaction keys are meant to prevent double-loading across a fleet, and your record store is a laptop’s home directory, they prevent nothing.

All the workers need to see the same store. Same rule as results, same failure mode as caching: two workers with two private stores never see each other’s records, so every run is a first run, and your at-most-once transaction runs once per worker. This is the single most common way a correct-looking transaction key does nothing at all.

Two more knobs on transaction() round it out. overwrite=True ignores an existing record and re-does the work — the transaction-level equivalent of refresh_cache, and your manual “no really, run it again” override. And write_on_commit=False gives you the hooks and the atomicity without persisting a record at all, when you want the rollback semantics but not the idempotency.

Transaction key and cache key are the same machinery

This is the section that will save you the most confusion, so I’m going to be blunt about it: a task’s cache key is its transaction key. They are not analogous. They are not similar. They are literally the same string, computed by the same code path, stored in the same store.

Every task run in Prefect 3 opens its own transaction. The engine computes that transaction’s key by calling cache_policy.compute_key(...) — the exact machinery from the caching chapter. Then, on entry, the transaction checks whether a record already exists for that key. If it does, the transaction begins life already COMMITTED, the task body is skipped, and the state you see in the UI is named… Cached.

That’s not a metaphor for how caching works. That is how caching works. refresh_cache on a task is passed straight through as overwrite on its transaction. A cache hit is a transaction that found its record already written. The Cached state is a transaction that was born committed.

So why do you have two words for it? Because they operate at different scopes, and the scope is the entire distinction:

  • A cache policy keys the transaction that wraps one task. Its unit of atomicity is a single function call. Its question is “did we already compute this?”
  • A with transaction(key=...) block keys a transaction that wraps a group of tasks, with your own hooks attached to it. Its unit of atomicity is the block. Its question is “did we already commit this whole thing, and if the middle of it blows up, how do I undo the beginning?”

Caching gives you skip-the-work. Transactions give you skip-the-work plus undo-the-work. Same store, same key mechanism, one extra guarantee — and that guarantee is the only reason to reach for the heavier tool.

One honest correction while we’re here, because I’ve seen this claimed (and this series claimed it too, in an earlier draft): a group-level transaction whose key is already committed does not magically skip the body of your with block. Python doesn’t work that way. What happens is that the transaction enters in a COMMITTED state, and commit() becomes a no-op — the hooks don’t re-fire, the record isn’t rewritten. The statements inside the block still execute, and the tasks inside still get evaluated (they’ll each hit their own caches, if their own policies say so). If you want the block to actually short-circuit, you have to ask:

with transaction(key=f"bookshop-load-{day}") as txn:
    if txn.is_committed():
        logger.info("day %s already committed; nothing to do", day)
        return
    stage(orders)
    load(staged)

I ran the version without that if, twice, and watched the task body execute both times. txn.is_committed() is not decoration. It’s the branch.

What transactions are not

Time to be a bad salesman, because the fastest way to get burned by this feature is to hear the word “transaction” and import your database intuitions wholesale.

They are not database transactions. There is no ACID here, no MVCC, no write-ahead log in the storage engine sense. Prefect is not intercepting your INSERTs. If two tasks in a Prefect transaction each write to Postgres, Postgres has no idea they’re related; a reader querying between them sees a half-done world, exactly as it would have without the with block. Prefect gives you compensation, not isolation from readers. If you need other sessions to never observe the intermediate state, you need a database transaction — and the right move is often to put a real BEGIN/COMMIT inside a single task, with the Prefect transaction wrapping the coarser, cross-system stuff around it.

They do not roll back side effects you didn’t write a hook for. This is the big one. Prefect does not know what your task did. It cannot infer that write_staging() should be undone by clear_staging(). The rollback is entirely the code in your on_rollback function, and a task with no rollback hook rolls back by doing precisely nothing, silently, with no warning. The framework gives you the call; you still write the undo. A transaction full of tasks with no hooks is an expensive way to type with.

The task that failed does not get rolled back. Look closely at that trace from earlier. load raised — and load’s own rollback hook never ran. Only stage’s did. That’s not a bug; it’s the semantics. A task’s hooks are registered with the transaction when the task succeeds and stages its result. A task that dies partway through its own side effect never staged anything, so it has nothing to roll back, and Prefect will not guess. Compensation covers completed work, not the half-finished work of the task that broke. If load_warehouse() can fail after writing some rows, cleaning that up is your job, inside load — a try/except around your own partial write. The transaction protects the steps around the explosion, not the inside of the explosion.

Not everything is reversible. You cannot un-send an email. You cannot un-call a partner’s POST /orders. When a step is irreversible, the transaction’s job is not to undo it — it’s to make sure you never do it until the commit. That’s what on_commit is for, and if you find yourself writing an on_rollback that tries to walk back an irrevocable action, you’ve put the action in the wrong place. Move it to the commit hook and let the transaction decide whether it happens at all.

And hooks can themselves fail. A rollback hook that raises will be logged, and the rollback will abort. Write them the way you’d write any cleanup: idempotent, defensive, DELETE ... IF EXISTS, safe to run twice. Your compensation code runs on the worst day your pipeline has; that is a bad day to discover it assumed a happy path.

Final thoughts

The reason Airflow doesn’t have this isn’t a missing feature — it’s a missing idea. Airflow’s model is a graph of tasks the scheduler moves through, and a task’s relationship to the outside world is entirely opaque to it. There is no place in that model to hang “and here is how to undo it,” because the graph is a plan for what to run, not a record of what happened. So Airflow engineers do what they’ve always done: they draw the compensation as more graph, wire it with a trigger rule, and hold it right in their heads.

Prefect can offer transactions because it already treats a task run as a durable, keyed, recorded thing — the same substrate that made results and caching work in the last chapter. Once a task run has a record in a store, “has this committed?” is a lookup, “commit these together” is a tree walk, and “undo this” is a hook you attach next to the code it undoes. It’s not magic. It’s just the third feature you get from the same foundation, and it’s the one that turns a red pipeline from a mess you clean up into a mess that cleans up after itself.

Use it where it earns its place: multi-step operations that write to more than one system, side effects you’d have to explain to someone in a post-mortem, days you must load exactly once. Skip it where a plain retry will do. And write the hooks — because a transaction with no hooks is a promise you made to yourself that nobody, including Prefect, has any intention of keeping.

Next: Automations, Events, and Webhooks

Comments