Skip to content

symfonic.memory.backends.in_memory_transaction

in_memory_transaction

All of a batch, or none of it, for the backend that lives in this process.

The Postgres pools offer a transaction and the in-process backend needs the same word for the same promise, because the consolidation commit asks for one by name. "Nothing in the batch awaits anything that yields, so nothing can interleave" was only half the guarantee: a batch whose third operation raises has already applied the first two, and no amount of non-interleaving takes those back.

So the tables are snapshotted on entry and restored if the block raises. Deep, because update_node mutates the stored node in place -- a shallow copy of the dicts would leave every field change from the failed block sitting in the "restored" rows.

Restoring cannot lose a concurrent writer's work: nothing inside the block awaits anything that yields, so there is no concurrent writer during it. That is the part the earlier reasoning got right, and it is what makes snapshot-and-restore safe rather than merely convenient.

Its own module, and imported from inside the method rather than at module scope, so it stays out of the eager import graph an adopter pays for on from symfonic.agent import SymfonicAgent. Nothing needs it until a batch opens a transaction, and only a consolidation commit does.

snapshot async 

snapshot(store: Any) -> AsyncIterator[Any]

Run a block against store so its writes all land or none of them do.

Nested use joins the outer block rather than opening a savepoint: the callers want "all of this together", and an inner block that could commit on its own would be a quieter way of not having a transaction.

Reaches the tables directly because a snapshot of a store's rows is what this is; publishing them so an outsider could copy them would be a wider surface for no gain, since only this function may.

Source code in src/symfonic/memory/backends/in_memory_transaction.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
@asynccontextmanager
async def snapshot(store: Any) -> AsyncIterator[Any]:
    """Run a block against ``store`` so its writes all land or none of them do.

    Nested use joins the outer block rather than opening a savepoint: the
    callers want "all of this together", and an inner block that could commit
    on its own would be a quieter way of not having a transaction.

    Reaches the tables directly because a snapshot of a store's rows is what
    this is; publishing them so an outsider could copy them would be a wider
    surface for no gain, since only this function may.
    """
    if getattr(store, "_in_transaction", False):
        yield store
        return
    saved = (
        deepcopy({t: dict(rows) for t, rows in store._nodes.items()}),  # noqa: SLF001
        deepcopy({t: dict(rows) for t, rows in store._edges.items()}),  # noqa: SLF001
        dict(store._workset_cursors),  # noqa: SLF001
    )
    store._in_transaction = True  # noqa: SLF001
    try:
        yield store
    except BaseException:
        store._nodes = defaultdict(dict, saved[0])  # noqa: SLF001
        store._edges = defaultdict(dict, saved[1])  # noqa: SLF001
        store._workset_cursors = saved[2]  # noqa: SLF001
        raise
    finally:
        store._in_transaction = False  # noqa: SLF001