Persistence Stores

StateStore protocol with Memory, File and SQLite backends — the create → act → persist → discard model for web workers, Celery and multi-process apps.

An interpreter is process memory. Under gunicorn or uvicorn workers, in a Celery task, or when two requests arrive for the same order, “keep it in RAM” is wrong. Every durable system converges on the same four-step loop:

flowchart LR
    L["1 · load<br/><small>snapshot → started interpreter</small>"] --> A["2 · act<br/><small>send one event (or a batch)</small>"]
    A --> P["3 · persist<br/><small>save with expected_version</small>"]
    P --> D["4 · discard<br/><small>stop(); drop the object</small>"]
    P -. "ConflictError" .-> L

xstate_statemachine.persistence gives that loop a store: a small protocol with three zero-dependency backends, an optimistic-locking token on every record, and a pair of helpers that do steps 1 and 3 for you. Redis ([redis], see the Redis guide), SQLAlchemy and Django ship backends implementing the same protocol.

The loop in six lines

from xstate_statemachine import MachineLogic, create_machine
from xstate_statemachine.persistence import (
    ConflictError,
    MemoryStore,
    load_interpreter,
    save_interpreter,
)

cfg = {
    "id": "order",
    "initial": "cart",
    "context": {
        "items": 0,
    },
    "states": {
        "cart": {
            "on": {
                "ADD": {
                    "actions": "add",
                },
                "CHECKOUT": "paid",
            },
        },
        "paid": {
            "type": "final",
        },
    },
}


def add(i, ctx, e, a):
    ctx["items"] += 1


machine = create_machine(
    cfg,
    logic=MachineLogic(
        actions={
            "add": add,
        }
    ),
)
store = MemoryStore()  # swap for FileStore / SQLiteStore


def handle(key: str, event: str) -> None:
    while True:  # the optimistic retry loop
        interp, version = load_interpreter(store, key, machine)  # started
        try:
            interp.send(event)
            save_interpreter(store, key, interp, expected_version=version)
            return
        except ConflictError:
            continue  # someone saved first: reload, redo
        finally:
            interp.stop()  # discard


handle("order-42", "ADD")
handle("order-42", "ADD")
interp, version = load_interpreter(store, "order-42", machine)
assert interp.context["items"] == 2 and version == 2
interp.stop()

load_interpreter returns a started SyncInterpreter and the record’s version — 0 when the key did not exist yet, so the first save_interpreter(expected_version=0) succeeds only if nobody else created the record in between. For the async engine use await aload_interpreter(...).

Choosing a backend

Every backend passes the same contract test suite (tests/persistence/test_store_contract.py), so switching is a one-line change.

What a record holds

from xstate_statemachine import SyncInterpreter, create_machine
from xstate_statemachine.persistence import MemoryStore, StoredSnapshot

machine = create_machine(
    {
        "id": "m",
        "version": "3",
        "initial": "a",
        "states": {
            "a": {},
        },
    }
)
interp = SyncInterpreter(machine).start()
store = MemoryStore()

version = store.save(
    "m-1", interp.get_snapshot(), machine_version=machine.version or ""
)
record = store.load("m-1")
assert isinstance(record, StoredSnapshot)
assert (record.key, record.version, record.machine_version) == ("m-1", 1, "3")
assert record.deadlines == ()  # durable timers land here (#264)
interp.stop()

Optimistic vs pessimistic

Optimistic (save(..., expected_version=n)) is the default and the right choice almost always: no lock is held while your event runs, and a lost race costs one reload. ConflictError carries .expected and .actual so a supervisor can see how contended a key is.

Pessimistic (with store.lock(key, timeout=10):) is for the rare step that must not be retried — an action with an external side effect that is not idempotent. LockTimeoutError is raised if the lock cannot be taken in time; for SQLiteStore this also covers SQLite’s own database is locked, so you never see a bare sqlite3.OperationalError. The lock is per key on MemoryStore / FileStore and database-wide on SQLiteStore (SQLite has no row locks — the honest granularity).

from xstate_statemachine.persistence import MemoryStore, LockTimeoutError
import threading

store = MemoryStore()
with store.lock("order-1", timeout=1):
    def contender():
        try:
            with store.lock("order-1", timeout=0.1):
                raise AssertionError("must not get in")
        except LockTimeoutError:
            print("blocked, as expected")
    t = threading.Thread(target=contender); t.start(); t.join()

persisted() — the safe pattern as a with block

The retry loop above is what every handler ends up writing. persisted() writes it for you, with a lock strategy deciding how concurrent writers are coordinated:

from xstate_statemachine import MachineLogic, create_machine
from xstate_statemachine.persistence import MemoryStore, persisted

cfg = {
    "id": "order",
    "initial": "cart",
    "context": {
        "items": 0,
    },
    "states": {
        "cart": {
            "on": {
                "ADD": {
                    "actions": "add",
                },
                "PAY": "paid",
            },
        },
        "paid": {
            "type": "final",
        },
    },
}


def add(i, ctx, e, a):
    ctx["items"] += 1


machine = create_machine(
    cfg,
    logic=MachineLogic(
        actions={
            "add": add,
        }
    ),
)
store = MemoryStore()

with persisted(store, "order:42", machine) as order:  # load (started)
    order.send("ADD")  # act
    order.send("ADD")
# persist on clean exit; the interpreter is stopped (discarded)

with persisted(store, "order:42", machine) as order:
    assert order.context["items"] == 2
    receipt = order.send("PAY", wait=True)
    assert receipt.changed
assert store.load("order:42").version == 2

Two rules make the block predictable:

1. An exception inside the block writes nothing. The store is exactly as it was; the interpreter is stopped. Your try/except around the block is the transaction boundary.
2. A block cannot be re-run. Under the default OptimisticLock, if another writer saved between your load and your exit, the exit raises ConflictError and you retry the whole block. When you want the library to retry for you, hand it a callable instead: persisted_retry(store, key, machine, lambda i: i.send("ADD")).

from xstate_statemachine import MachineLogic, create_machine
from xstate_statemachine.persistence import (
    ConflictError,
    MemoryStore,
    persisted,
    persisted_retry,
)

cfg = {
    "id": "c",
    "initial": "s",
    "context": {
        "n": 0,
    },
    "states": {
        "s": {
            "on": {
                "T": {
                    "actions": "inc",
                },
            },
        },
    },
}


def inc(i, ctx, e, a):
    ctx["n"] += 1


machine = create_machine(
    cfg,
    logic=MachineLogic(
        actions={
            "inc": inc,
        }
    ),
)
store = MemoryStore()

with persisted(store, "k", machine):
    pass  # create the record (version 1)

try:
    with persisted(store, "k", machine) as i:
        persisted_retry(
            store, "k", machine, lambda x: x.send("T")
        )  # a concurrent writer
        i.send("T")
except ConflictError as exc:
    print(
        f"conflict: expected v{exc.expected}, found v{exc.actual}"
    )  # block's work discarded

persisted_retry(
    store, "k", machine, lambda i: i.send("T")
)  # retries until it wins
with persisted(store, "k", machine) as i:
    assert i.context["n"] == 2

For the async engine use async with apersisted(store, key, machine) as interp: — store may be a sync store (calls go through the executor) or an as_async() adapter.

The async block settles before it saves (#263 battle). await interp.send("PAY", wait=True) resolves at the end of that event’s macrostep. When the step entered a state with a plain def service, the service’s completion is the next macrostep — per SCXML — and the run loop starts it at once; if that step enters a second def invoke (authorise → capture), the machine is mid-step exactly when your block exits. The sync engine’s send() drains the whole chain before returning, so persisted() never had this problem; apersisted() now closes the gap by awaiting interp.await_settled(settle_timeout) before the snapshot (default DEFAULT_SETTLE_TIMEOUT, 30 s — it only ever waits while a service is genuinely running). The same settle runs inside the Starlette / FastAPI / Litestar / Quart registries before the receipt body is built, so the 200 a caller reads is what the store holds. settle_timeout=0 opts out: a mid-step machine is then refused with SnapshotMidStepError, as before — loud, never torn. A fired or armed after is not owed work (a machine sitting on a live SLA timer is settled), so a block on a timer-heavy chart does not wait out the timer.

What persisted() promises

Every row is asserted by tests/persistence/test_battle_260_locking_semantics.py on MemoryStore, FileStore and SQLiteStore, sync and async:

Cost. persisted() is 1.00–1.09× a hand-rolled load → from_snapshot → send → save(expected_version) on MemoryStore and SQLiteStore; PessimisticLock on SQLite is faster (0.89×) because BEGIN IMMEDIATE makes the load and the save one transaction.

When something fails inside the block — each row injected by tests/persistence/test_battle_260_locking_failures.py on every store, sync and async:

load_interpreter(create_if_missing=True) returns a started interpreter at version 0 and does not save until you do. OptimisticLock(retries=…, backoff=…) raises TypeError for a non-int retries (bool included) or a non-RetryPolicy backoff, and ValueError for a negative retries. apersisted() on a sync store closes the as_async adapter it created, on success and on error (1 000 cycles: thread count flat).

Concurrency: choosing a lock

Each persisted() block (and each call of the callable under persisted_retry) gets its own interpreter; the engine is never shared across threads.

OptimisticLock + FileStore under heavy same-key contention. Each FileStore.save takes the OS file lock for its read-compare-write and waits at most 10 s for it. Sixteen threads hammering one key with fsync’d saves starved a few calls past that on a slow CI runner: they raised LockTimeoutError — loud, never a lost update, every call that returned was counted exactly once. If you expect that contention profile on one key, use PessimisticLock (one lock around the whole step, writers queue in order) or SQLiteStore (24 µs saves, no starvation observed).

import threading
from xstate_statemachine import MachineLogic, create_machine
from xstate_statemachine.persistence import (
    MemoryStore,
    OptimisticLock,
    PessimisticLock,
    persisted,
    persisted_retry,
)

cfg = {
    "id": "c",
    "initial": "s",
    "context": {
        "n": 0,
    },
    "states": {
        "s": {
            "on": {
                "T": {
                    "actions": "inc",
                },
            },
        },
    },
}


def inc(i, ctx, e, a):
    ctx["n"] += 1


machine = create_machine(
    cfg,
    logic=MachineLogic(
        actions={
            "inc": inc,
        }
    ),
)

for lock in (OptimisticLock(retries=100), PessimisticLock(timeout=30)):
    store = MemoryStore()

    def worker():
        for _ in range(25):
            persisted_retry(
                store, "k", machine, lambda i: i.send("T"), lock=lock
            )

    threads = [threading.Thread(target=worker) for _ in range(8)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    with persisted(store, "k", machine, lock=lock) as i:
        assert i.context["n"] == 200, type(lock).__name__  # no lost updates

Safety rails every store enforces

• Size cap. max_snapshot_bytes (default 1 MiB) is checked on save and load — SnapshotTooLargeError. A record another writer poisoned cannot make a reader allocate unboundedly.
• Key validation. Non-empty str, ≤ 200 characters, no NUL — InvalidKeyError. FileStore additionally refuses anything path-like and never puts the raw key in a path: keys are percent-encoded (reversibly, case-preserving, so Order and order never collide on a case-insensitive filesystem) and Windows device names are prefixed.
• forget(key) erases the record and everything auxiliary (deadlines, lock files) and reports counts — the right-to-erasure hook.
• Codec seam. codec= takes any encode(str) -> str / decode(str) -> str pair, so compression or encryption at rest is a constructor argument, not a backend fork.
• Atomic writes. FileStore writes to a temp file in the same directory, fsyncs, then os.replaces — a reader sees the old record or the new one, never a torn one. A fault hook in the tests simulates a crash between fsync and rename and asserts the previous record is intact.
• Permissions. FileStore directory 0700, files 0600; SQLiteStore creates the database (and its -wal / -shm siblings) 0600.
• Schema versioning. SQLiteStore keeps an xsm_schema(version) table with explicit upgrade steps; a database written by a newer library is refused, never guessed at. Current schema is v2 (adds the deadlines(key) index); a v1 file from 0.11.0 is upgraded in place on open.
• health() on every store — a cheap liveness probe for your readiness endpoint.

What every backend agrees on

The battle test for the stores (tests/persistence/test_battle_259_stores_corruption_scaling.py) drives the same contract matrix through MemoryStore, FileStore, SQLiteStore and the as_async view of each, and asserts identical answers. This is the contract you can code against without knowing which backend is configured:

Two costs differ and are worth knowing before you pick a backend. list_keys is O(total keys) on MemoryStore and FileStore however small limit is (Memory sorts every key; File lists and decodes the whole directory — 1 ms at 100 keys, 24 ms at 2 000, 14 s to fill 10 000) and flat on SQLiteStore (16 µs). And FileStore.save is milliseconds (2.8 ms, fsync off; an atomic temp-write-replace) where SQLite is tens of microseconds (24 µs) and Memory is one. save / load / delete are O(1) in store size on all three in code; FileStore.save renames into a directory of n entries, which is flat on NTFS and ext4 but read ~8× slower at 2 000 keys than at 100 on APFS (macOS CI) — a filesystem property worth knowing before putting tens of thousands of instances in one FileStore directory. Memory does not grow over 10 000 save→load→delete cycles on any backend (0 B attributed to the library); SQLite keeps one connection per thread and close() really closes it; file handles are stable.

Operations: what the failures look like

Every row below was injected by tests/persistence/test_battle_259_stores_crash_concurrency.py — real kill -9 of a writer child at every step of FileStore.save, ENOSPC at every syscall, Windows sharing violations, two processes on one lock — and the behaviour described is what the test asserts.

asyncio

The stdlib stores are synchronous (file and SQLite I/O block). as_async(store) runs each call in the default executor and exposes the same surface with await; lock() becomes an async with, acquired and released on one dedicated worker thread because file and SQLite locks are thread-affine.

import asyncio
from xstate_statemachine import create_machine
from xstate_statemachine.persistence import (
    MemoryStore,
    aload_interpreter,
    as_async,
    save_interpreter,
)

machine = create_machine(
    {
        "id": "m",
        "initial": "a",
        "states": {
            "a": {
                "on": {
                    "GO": "b",
                },
            },
            "b": {},
        },
    }
)
store = MemoryStore()


async def main():
    interp, version = await aload_interpreter(
        store, "m-1", machine
    )  # started Interpreter
    await interp.send("GO", wait=True)
    save_interpreter(store, "m-1", interp, expected_version=version)
    await interp.stop()

    astore = as_async(store)
    record = await astore.load("m-1")
    assert record.version == 1
    async with astore.lock("m-1"):
        assert (await astore.list_keys(prefix="m-")) == ["m-1"]


asyncio.run(main())

Idempotency: the inbox

Every broker, webhook provider (Stripe, GitHub, Twilio) and retrying HTTP client delivers at least once. Duplicate deliveries double-credit wallets and double-ship orders. The fix is deduplication at the consumer keyed by a stable event id — the inbox pattern — and IdempotencyPlugin makes it an attachable plugin: duplicates are answered from the inbox before the machine sees them, with the same receipt the first delivery got.

from xstate_statemachine import (
    MachineLogic,
    SyncInterpreter,
    create_machine,
    receipt_to_status,
)
from xstate_statemachine.persistence import IdempotencyPlugin, MemoryInbox

# A Stripe-shaped webhook consumer: `event.id` ("evt_…") is the natural key.
cfg = {
    "id": "wallet",
    "initial": "open",
    "context": {
        "balance": 0,
    },
    "states": {
        "open": {
            "on": {
                "payment_intent.succeeded": {
                    "actions": "credit",
                },
            },
        },
    },
}


def credit(i, ctx, e, a):
    ctx["balance"] += e.payload["amount"]


machine = create_machine(
    cfg,
    logic=MachineLogic(
        actions={
            "credit": credit,
        }
    ),
)

inbox = MemoryInbox()  # or SQLiteInbox(store)
plugin = IdempotencyPlugin(
    inbox,
    principal=lambda e: e.payload["account"],  # X0.2: scope by tenant, always
    key=lambda e: e.payload.get("id"),  # Stripe's event id
)
wallet = SyncInterpreter(machine).use(plugin).start()


def webhook(body: dict) -> int:
    receipt = wallet.send(body["type"], wait=True, **body)
    return receipt_to_status(receipt)


first = {
    "type": "payment_intent.succeeded",
    "id": "evt_1",
    "account": "acct_A",
    "amount": 500,
}
assert webhook(first) == 200 and wallet.context["balance"] == 500
assert (
    webhook(first) == 200 and wallet.context["balance"] == 500
)  # redelivery: same answer, credited once
assert (
    webhook(
        {
            **first,
            "amount": 999,
        }
    )
    == 422
)  # same id, different payload: refused
assert (
    webhook(
        {
            **first,
            "account": "acct_B",
        }
    )
    == 200
)  # another tenant's evt_1 is a new event
assert wallet.context["balance"] == 1000

What the plugin does, on both engines:

A refusal is a receipt, not an exception: plugin hooks are contained (a raising hook would admit the event), so the plugin answers with a receipt whose error says why. Use send(wait=True) at an ingress and read receipt.error / receipt_to_status.

Scope is principal / machine id / instance key — never just the actor id. principal is required: two tenants reusing evt_1 must not collide, and a tenant must not be able to replay another’s outcome. The instance key defaults to the store key persisted() loaded the machine under.

Fingerprint is sha256(event type + canonical JSON payload minus the key field), so a reused key with a different body is detectable. Keys are ≤ 255 printable ASCII characters.

TTL defaults to 7 days (Stripe’s window); inbox.purge_expired() reclaims space; inbox.forget(scope) erases a tenant.

Guarantees — at-least-once + inbox, never exactly-once

What you get: a duplicate delivery is answered without the machine processing it again, and the answer is the one the first delivery produced. What you do not get: exactly-once side effects. The inbox covers the state transition; an action that calls an external API is still an action that ran once per admitted event. Put side effects in services, or behind an outbox.

Crash consistency (X0.3) — the mark must be visible iff the snapshot containing the effect is:

1. Shared backend. When the inbox and the state store share a database (SQLiteInbox(store) on the same SQLiteStore), persisted() buffers the mark and writes it right after the snapshot save — inside the same BEGIN IMMEDIATE transaction under PessimisticLock, so they commit or roll back together.
2. Separate backends. Save, then mark. The window between them is covered by a bounded ring of the last 64 processed keys that travels inside the snapshot (context["__xsm_processed_ids__"]): a redelivery that finds the key in flight in the inbox but processed in the snapshot is the crash window, and is answered as a duplicate and the inbox repaired.

Three fault-injection tests pin this: crash before save (no mark, retry is a first delivery), crash between save and mark (caught by the ring), crash after mark (plain duplicate).

When the inbox itself is down — on_inbox_error

The trade-off, stated plainly. Plugin hooks are fail-open by design (a raising hook admits the event). Before this battle test, that meant an inbox backend failure — get or claim raising StoreError / OSError / sqlite3.Error, or a SQLite lock timeout — let the keyed event run with no deduplication, with a perfectly normal receipt (changed=True, duplicate=False) and on_plugin_error as the only trace. The #261 battle test drove a dead inbox and watched the action run twice for one key with nothing on either receipt; it also found that apersisted + PessimisticLock + a shared SQLiteInbox hit that path on every send.

The default is now on_inbox_error="refuse": the receipt carries InboxUnavailableError (receipt_to_status → 503, “retry later”), the action never runs, nothing is claimed. The whole point of the plugin is to not run twice, so a dead inbox must not quietly switch dedup off. Since 0.12.0 the plugin’s own callables are held to the same rule: a key, principal or instance_key function that raises (a KeyError on payload["user"], an auth lookup that is down) refuses the event, with the receipt’s error naming the callable, instead of admitting it undeduplicated. IdempotencyPlugin(..., on_inbox_error="admit") opts back into availability for a webhook endpoint whose sender retries anyway and tolerates a duplicate — alert on on_plugin_error there, it is the only signal. The plugin’s own refusals — mismatch (422), in-flight (409), an un-fingerprintable payload — are the same in both modes.

Crash windows, as measured. A child process is killed (os._exit(9)) at each step of claim → action → save → mark, the parent restarts on the same database and redelivers the same payload, then a different payload on a copy. In every cell the committed effect is exactly one or zero; the action never runs twice on top of a committed save, sync and apersisted alike:

(same payload / different payload). Two things the table makes visible: a kill right after the claim leaves that key 409 for the full ttl_s (7 days by default — there is no shorter lease on in-flight claims; purge or forget it, or set a shorter ttl_s), and a PessimisticLock body on a shared SQLite inbox is one transaction, so every kill before the commit is a clean redelivery.

TTL semantics (ttl_s, default 7 days): both the in-flight claim and the final mark expire after ttl_s; after that the key is new — so ttl_s bounds the dedup window and the “same original receipt” promise. None never expires (a crashed worker then blocks that key until you purge() or forget() it). 0 makes every send fresh. Negative, NaN or non-numeric → ValueError at construction (they used to silently disable dedup). A wall clock jumping backwards never purges early; forwards past ttl_s purges. Keys are at most 255 printable ASCII characters, case-sensitive and byte-exact; non-ASCII and control characters are refused with a typed receipt. Scope footgun: two in-memory interpreters of one machine that share an id or instance_key share an inbox scope — pass instance_key= per instance (a persisted() block uses its store key automatically).

Cost (Windows, 3.14): bare send 14 µs; with the plugin on MemoryInbox 64 µs; on SQLiteInbox 178 µs. MemoryInbox ≈ 377 B per entry and purge_expired() reclaims it; a 10 000-row SQLiteInbox is 2.8 MB, purge of 10 000 expired rows 6.8 ms (indexed), 0 expired rows 0.03 ms.

from xstate_statemachine import MachineLogic, create_machine
from xstate_statemachine.persistence import (
    IdempotencyPlugin,
    SQLiteInbox,
    SQLiteStore,
    persisted,
)
import tempfile, pathlib

cfg = {
    "id": "wallet",
    "initial": "open",
    "context": {
        "balance": 0,
    },
    "states": {
        "open": {
            "on": {
                "CREDIT": {
                    "actions": "credit",
                },
            },
        },
    },
}


def credit(i, ctx, e, a):
    ctx["balance"] += e.payload["amount"]


machine = create_machine(
    cfg,
    logic=MachineLogic(
        actions={
            "credit": credit,
        }
    ),
)

db = pathlib.Path(tempfile.mkdtemp()) / "app.db"
store = SQLiteStore(db)
inbox = SQLiteInbox(store)  # shares the file AND the transaction
dedupe = IdempotencyPlugin(inbox, principal=lambda e: "acct_A")

for _ in range(3):  # the same webhook, delivered three times
    with persisted(store, "wallet:acct_A", machine, plugins=[dedupe]) as w:
        w.send("CREDIT", idempotency_key="evt_1", amount=500)

with persisted(store, "wallet:acct_A", machine, plugins=[dedupe]) as w:
    assert w.context["balance"] == 500  # credited exactly once
store.close()

Audit log & replay

“Who approved this and why?” is answered by grepping Slack in most codebases. AuditPlugin writes an append-only record for every processed event — including the ones that were refused — from the same plugin hook that observed the outcome, so the audit can never disagree with the state. The log doubles as event sourcing lite: replay() re-runs the recorded events from context₀ and asserts every step lands where it was recorded.

from xstate_statemachine import MachineLogic, SyncInterpreter, create_machine
from xstate_statemachine.persistence import AuditPlugin, MemoryLog, replay

# An approval workflow: every decision carries who made it and why.
cfg = {
    "id": "expense",
    "initial": "submitted",
    "context": {
        "amount": 1200,
    },
    "states": {
        "submitted": {
            "on": {
                "APPROVE": {
                    "target": "approved",
                    "guard": "underLimit",
                },
                "ESCALATE": "escalated",
                "REJECT": "rejected",
            },
        },
        "escalated": {
            "on": {
                "APPROVE": "approved",
                "REJECT": "rejected",
            },
        },
        "approved": {
            "type": "final",
        },
        "rejected": {
            "type": "final",
        },
    },
}
logic = MachineLogic(
    guards={
        "underLimit": lambda ctx, e: ctx["amount"] <= 1000,
    }
)
machine = create_machine(cfg, logic=logic)

log = MemoryLog()  # or JSONLinesLog(path) / SQLiteLog(store)
expense = SyncInterpreter(machine).use(AuditPlugin(log)).start()

expense.send(
    "APPROVE", actor="alice", reason="looks fine"
)  # guard says no: recorded as denied
expense.send("ESCALATE", actor="alice", reason="over my limit")
expense.send(
    "APPROVE",
    actor="bob",
    reason="Q3 travel budget",
    correlation_id="req-7f3a",
)

for r in log.read("expense"):
    print(
        r.seq,
        r.event_type,
        r.disposition,
        r.actor,
        r.reason,
        "->",
        r.to_states[-1].split(".")[-1],
    )
# 1 APPROVE denied alice looks fine -> submitted
# 2 ESCALATE transition alice over my limit -> escalated
# 3 APPROVE transition bob Q3 travel budget -> approved

rebuilt = replay(
    machine, log.read("expense")
)  # re-run from context₀ on a SimulatedClock
assert rebuilt.current_state_ids == expense.current_state_ids

Each TransitionRecord carries: machine_id (the store key under persisted(), else the interpreter id), a gap-free seq, ts (interpreter.wall_now()), event_type and a redacted event_payload, from_states / to_states (leaves), the actions that ran, a disposition (transition · denied · unhandled · deferred · error · duplicate), actor / reason / correlation_id, machine_version, engine=True for timer and service-completion events, and origin ("external" for a caller’s send(), "internal" for a raise, a self-send() from an action, or a re-released deferred event — replay() re-sends only external records). Records are plain JSON and round-trip through JSONLinesLog.

actor and reason come from the event payload (actor_key= / reason_key= configurable); correlation_id from the payload or, failing that, from the correlation_id_var contextvar — set it in your request middleware and every record in that request carries it.

All three: read(machine_id, after_seq=, limit=), purge_older_than(cutoff_ts), forget(machine_id). append(rec, connection=None) is the seam a Django or SQLAlchemy log store uses to join the store’s transaction.

What replay() is — and is not

replay(machine, records, *, upto=None, logic=None, verify=True, key=None, snapshot=None) returns a started SyncInterpreter positioned after the last record. The #262 battle test recorded a chart exercising every outcome kind — denied, deferred, errored under continue and rollback, a 3-deep raise chain, an always chain, two after rungs, invoke with onDone and onError, nested + parallel + deep history, final — on both engines and all three log stores, and asserted the replay lands on the identical configuration:

This is event sourcing lite: the log is a faithful record and a replayable one, but the snapshot stays the source of truth for the current state. There is no projection framework, no upcasting of old events, and no guarantee about actions’ external effects — those belong to services and an outbox.

The log and the snapshot after a crash

Records are written only for committed steps. Inside a persisted() block they are buffered and released at the snapshot save — inside the same BEGIN IMMEDIATE transaction when SQLiteLog shares the SQLiteStore under PessimisticLock, via after_commit otherwise — so a lost optimistic attempt, a block that raised, or a writer killed before the save leaves no audit rows. (Before the battle test, records were appended as each event settled, so the log could run ahead of the state.) Child processes killed with os._exit(9) after the save and after the append, step 1 committed, step 2 in flight:

seq is assigned atomically by the store (append_next: MAX(seq)+1 and the insert in one statement on SQLite; under the process lock on Memory/JSONL) — 16 threads × 100 under OptimisticLock, PessimisticLock and NoLock, and 2 processes on one SQLiteLog key, are all gap-free. The NoLock + audit “unique seq constraint” limitation listed in the handover is fixed, not documented. JSONLinesLog.append_next is atomic within one process only; use SQLiteLog for multi-process writers.

Backend contract, as measured

replay() loads and sorts the supplied records (O(n) memory, ~9 s and ~8 MB per 100 000 records). Plugin memory is flat over 10 000 sends with a durable store; MemoryLog grows by exactly the records and purge_older_than() reclaims.

Versioning in-flight instances

You ship v2 of a machine while 10,000 orders are mid-flight on v1. Temporal needs explicit patched() / worker versioning; XState has an open bug restoring child-actor snapshots; no Python FSM library handles it at all. We do not automatically migrate. We promise the two things that are honest and valuable:

1. Every snapshot knows which machine produced it — machine_hash (the structural fingerprint, since 0.8.0) and machine_version (the chart’s "version" label, the thing a human reads).
2. A mismatch fails loudly — MachineVersionMismatchError (a SnapshotDriftError) — unless you registered an upcaster for exactly that hop.

from xstate_statemachine import SyncInterpreter, create_machine
from xstate_statemachine.persistence import (
    MachineVersionMismatchError,
    SnapshotMigrator,
)

v1 = {
    "id": "order",
    "version": "1.0",
    "initial": "paying",
    "states": {
        "paying": {
            "on": {
                "OK": "done",
            },
        },
        "done": {
            "type": "final",
        },
    },
}
v2 = {
    "id": "order",
    "version": "2.0",
    "initial": "payment",
    "states": {
        "payment": {
            "initial": "card",
            "states": {
                "card": {
                    "on": {
                        "OK": "#order.done",
                    },
                },
            },
        },
        "done": {
            "type": "final",
        },
    },
}

blob = (
    SyncInterpreter(create_machine(v1)).start().get_snapshot()
)  # written by v1

try:
    SyncInterpreter.from_snapshot(
        blob, create_machine(v2)
    )  # default: refuse loudly
except MachineVersionMismatchError as exc:
    print(exc.found, "->", exc.expected)  # 1.0 -> 2.0

migrator = SnapshotMigrator()


@migrator.register("1.0", "2.0")
def rename_paying(blob: dict) -> dict:
    blob["state_ids"] = [
        "order.payment.card" if s == "order.paying" else s
        for s in blob["state_ids"]
    ]
    return blob  # `configuration` is rebuilt from the leaves against v2


order = SyncInterpreter.from_snapshot(
    blob, create_machine(v2), migrator=migrator
)
assert order.current_state_ids == {"order.payment.card"}

What happens on restore, in order: the label is compared (on_version_mismatch: "error" default · "warn" restores as-is · "migrate", the default when a migrator= is given); a migration applies the shortest chain of registered steps (1.0 → 2.0 → 3.0) to a copy of the blob, rewrites machine_version and drops machine_hash — the structure changed by definition; then the structural hash is checked (unless migrated), the layout is upcast, and the blob is validated against this machine exactly like any other: every state id must exist (StateNotFoundError, never a silent skip), the configuration must be legal, strict and event schemas still apply to restored events. A missing hop is NoMigrationPathError. Child actors are restored with the same migrator and policy — steps can be scoped with machine_id="kid".

What a step may and may not do — as battle-tested (#263). Every row is asserted by tests/persistence/test_battle_263_migration_semantics.py and _runtime.py:

Labels compare as strings: a chart "version": 1 and a blob "machine_version": "1" are one label; 1.0 becomes "1.0", a different label; "" is a mismatch. A blob with "machine_version": null is unlabelled — the library itself writes null for a chart with no "version" — and restores like an absent key. The “carries no machine_version” warning is logged once per (machine id, expected label) per process, not once per restore. Child actors: a child with no path fails the whole restore, naming the child; the parent’s step may rewrite actors[*].src; a migrated grandchild re-saves under its own new label.

Under persisted() / apersisted() with OptimisticLock, N workers opening one stale key all run the step in memory but exactly one commits (the record’s version counter advances by one, the label is the new one); under PessimisticLock the step runs once. A step that raises leaves the stored record byte-identical. The DueTimerScanner on a stale instance: without a migrator each scan records one MachineVersionMismatchError in ScanResult.errors and the deadline stays (no hot loop); with one, the timer fires and the record re-saves at the new label. Cost: a 1-hop migration of a 50-state blob is 1.65× a plain restore (the deep copy), 62 µs vs 38 µs.

Blobs written before labels existed (0.10.x) restore with a warning — they cannot be checked; a chart that declares no "version" never mismatches.

Rolling out a new version

• Additive changes first. A new state, a new event, a new context key with a default: the hash changes but no snapshot needs rewriting — pass verify_machine_hash=False for that deploy and bump the label with on_version_mismatch="warn", or register a no-op step so the label is rewritten on the next save.
• Renames and moves need a step. Write the upcaster against the blob (state ids, configuration, context), register it for the exact hop, and let persisted(..., migrator=migrator) re-save each instance at the new label the first time it is touched.
• Dual-read window. Deploy readers that carry the migrator before writers that emit the new label; old readers refuse new blobs loudly rather than half-restoring them (see the snapshots guide for the layout-level version of the same rule).
• Find what is still stale. xsm snapshots --store sqlite:///app.db machine.json --stale lists the keys whose machine_version differs from the chart’s (--json for scripts) — the drain list for a deploy; --fail-if-stale exits 1 while any remain, the gate before retiring the old workers. It reads the label index, never the blobs (10 000 keys in 0.1 s on SQLite). An unlabelled record is not stale and a chart with no "version" has nothing stale — the same rules restore applies. See the CLI page.

Guarantees. A snapshot is refused, never silently mis-restored, when its label or structure does not match — unless you said how to bridge the gap. Migration steps are your code; the library validates their output but cannot know your domain. Child-actor migration is best effort (each child is checked with its own machine; a child with no path fails the whole restore).

Durable timers

after timers are the library’s superpower for retries, SLAs, escalations and reminders — and until 0.11.0 they were in-memory tasks that died with the process. Under create → act → persist → discard the interpreter is discarded after every request, so a 24-hour after never fired. Now the deadline lives in the snapshot (deadlines, layout v4), indexed by the store, and a scanner wakes the machine when it matures. This deliberately reverses the 0.8.0 “timers are not persisted” decision.

import time
from xstate_statemachine import create_machine
from xstate_statemachine.persistence import (
    DueTimerScanner,
    MemoryStore,
    persisted,
)

cfg = {
    "id": "trial",
    "initial": "active",
    "states": {
        "active": {
            "after": {
                "3600000": "reminded",
            },
        },  # 1 hour
        "reminded": {
            "type": "final",
        },
    },
}
machine = create_machine(cfg)
store = MemoryStore()

with persisted(
    store, "user:42", machine
):  # arms the timer; the deadline is persisted
    pass
now = time.time()
(deadline,) = store.load("user:42").deadlines
assert abs(deadline.due_at_wall - (now + 3600)) < 5

scanner = DueTimerScanner(store, lambda key: machine)  # "another process"
assert scanner.run_once(now=now + 1800) == 0  # not early
assert scanner.run_once(now=now + 3601) == 1  # fires once due
assert scanner.run_once(now=now + 3601) == 0  # idempotent
with persisted(store, "user:42", machine) as trial:
    assert trial.matches("trial.reminded")

Each persisted Deadline carries state_id, entry_seq (the state-entry generation, so a deadline armed by an earlier visit to the same state is recognisably stale), due_at_wall (epoch seconds from interpreter.wall_now()), the resolved delay_ms (a named or dynamic delay is a number by then) and event_type. interpreter.pending_deadlines() lists them, soonest first, on both engines.

Restoring: restart_timers

On the sync engine timers fire from send() / tick(), so "fire_due" pumps once inside start(); on the async engine the loop runs the zero-delay callbacks and settles before start() returns. A deadline whose state no longer exists after a migration fails loudly (StateNotFoundError) rather than being dropped.

What a restore does with the record — as battle-tested (#264). Asserted by tests/persistence/test_battle_264_timer_semantics.py and _child_timers.py on both engines:

The scanner

DueTimerScanner(store, machine_for_key, *, lock=OptimisticLock(), plugins=(), now=time.time, skew_tolerance_s=0, prefix="", limit=1000, migrator=None, on_version_mismatch=None) is the zero-dependency driver: run_once(now) reads the store’s deadline index, and for every key with a matured deadline opens persisted(..., restart_timers="fire_due") under the lock strategy — the transition fires, the snapshot is saved, remaining deadlines are re-indexed. scan() returns a ScanResult (scanned, due, woken, skipped_stale, errors, max_lag_s — the “how late are we” metric); scanner.last_result keeps the most recent ScanResult; migrator / on_version_mismatch are forwarded to persisted(); run_forever(interval_s) / stop() for a dedicated process. Celery Beat, APScheduler or cron adapters just call run_once().

Under the lock the scanner re-reads the record and wakes only if a matured deadline is still there — a machine another worker already advanced is skipped (skipped_stale), never double-fired — and the save is fenced on the version that re-read saw, so a writer that slips in between the re-read and the save produces a ConflictError the scanner also counts as skipped_stale. skew_tolerance_s absorbs clock skew between the writing host and the scanning one; one key’s failure is recorded in errors and does not stop the pass.

limit caps machines woken per tick, earliest deadline first. Before the #264 battle it capped the keys scanned in name order, so a scheduler that came back to a backlog larger than the limit woke the same first N keys every tick and the rest never fired. A tick now reads the whole deadline index (one query on SQLite / SQLAlchemy / Redis / Django, a dict walk on Memory — due_keys(until_wall, limit=)), takes the limit earliest, and repeated ticks drain the backlog oldest-first. ScanResult.scanned is what was inspected (all of it); woken ≤ limit; max_lag_s on the first tick after an outage is the whole outage — that is your alert. FileStore has no deadline index: a tick reads every record (~2.4 ms each, ≈ 4 min per tick at 100 000 records). Use SQLite, SQLAlchemy or Redis for a scanned store beyond a few thousand records.

Two schedulers by accident, concurrency, crashes — as measured (tests/persistence/test_battle_264_scanner_concurrency.py, _stores_crash.py, and the fastapi_orders scheduler-outage scenario):

Side effects are at-least-once under OptimisticLock. A scanner that loses the race has already run the transition — actions included — before its save is refused: the 8-thread runs counted 61 309 uncommitted in-memory runs on Memory, 562 on SQLite, 782 on File, and 0 under PessimisticLock. Make timer actions idempotent, or use the pessimistic strategy. The idempotency inbox does not dedupe a re-fired after — an after event has no payload, so there is no key to derive; it protects the keyed events your timer action sends onward. Redis ttl_s: when the hashes expire the zset entry stays, and the scanner skips that key every tick without firing it.

Guarantees. A timer fires no earlier than its deadline (minus skew_tolerance_s) and no later than the next scanner tick after it; a restore never arms it for longer than its declared delay. Each deadline is committed exactly once under OptimisticLock or PessimisticLock, however many scanners and workers race; under OptimisticLock the losers may have run the transition in memory first (side effects at-least-once — see above); under NoLock there is no exactly-once. A crash between the fire and the snapshot save re-fires on the next tick (at-least-once): make the timer’s transition idempotent, or use PessimisticLock. A guarded after whose guard refuses is consumed like any denied event; the state’s timer is re-armed from zero on the next hydration, so the guard is asked again one delay later. Only the root machine’s deadlines are indexed — a child actor’s timer fires when its parent is woken, but cannot itself wake the parent.

xsm simulate runs on a SimulatedClock and shows after timers firing as you advance it; persisted deadlines are a runtime concern and are not part of the simulator.

Writing your own backend

Subclass BaseStore and implement the _load_raw / _save_raw / _delete_raw / _list_keys_raw / _lock_raw primitives on raw strings; the base class applies key validation, the size cap and the codec around them so no backend can forget a rail. Then add your factory to STORE_FACTORIES in the contract test suite and run it — that is the definition of a conforming store.