Skip to content

symfonic.memory.backends.in_memory_transaction

in_memory_transaction

Snapshot transactions sharing the graph backend's writer lock.

A transaction holds the lock from snapshot through rollback or commit. The owner is explicit: an awaited child joins by entering transaction(), while a separate writer waits even if it inherited the parent's context variable. A consolidation cycle rejects multiple graph domains; callers that manually nest distinct backends must use a consistent lock order.

snapshot async 

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

Restore the full preimage on failure; nested entries join the owner.

Source code in src/symfonic/memory/backends/in_memory_transaction.py 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
@asynccontextmanager
async def snapshot(store: Any) -> AsyncIterator[Any]:
    """Restore the full preimage on failure; nested entries join the owner."""
    if _context_owner(store):
        task = current_task()
        counts = store._write_owner_tasks  # noqa: SLF001
        counts[task] = counts.get(task, 0) + 1
        joined_token = _JOINED.set((store, store._write_owner, task, _JOINED.get()))  # noqa: SLF001
        try:
            yield store
        finally:
            counts[task] -= 1
            if not counts[task]:
                del counts[task]
            _JOINED.reset(joined_token)
        return

    _refuse_expired_join(store)
    async with store._write_lock:  # noqa: SLF001
        original_nodes = {t: dict(rows) for t, rows in store._nodes.items()}  # noqa: SLF001
        original_edges = {t: dict(rows) for t, rows in store._edges.items()}  # noqa: SLF001
        saved = (
            deepcopy(original_nodes),
            deepcopy(original_edges),
            dict(store._workset_cursors),  # noqa: SLF001
        )
        owner = object()
        token = _OWNER.set((store, owner, _OWNER.get()))
        store._write_owner = owner  # noqa: SLF001
        store._write_owner_tasks = {current_task(): 1}  # noqa: SLF001
        try:
            yield store
            if any(task is not current_task() for task in store._write_owner_tasks):  # noqa: SLF001
                raise RuntimeError("joined child transaction remains open")
        except BaseException:
            try:
                for tenant, rows in original_nodes.items():
                    for identifier, original in rows.items():
                        _restore_row(original, saved[0][tenant][identifier])
                for tenant, rows in original_edges.items():
                    for identifier, original in rows.items():
                        _restore_row(original, saved[1][tenant][identifier])
            finally:
                store._nodes = defaultdict(dict, original_nodes)  # noqa: SLF001
                store._edges = defaultdict(dict, original_edges)  # noqa: SLF001
                store._workset_cursors = saved[2]  # noqa: SLF001
            raise
        finally:
            store._write_owner = None  # noqa: SLF001
            store._write_owner_tasks = {}  # noqa: SLF001
            _OWNER.reset(token)

writer async 

writer(store: Any) -> AsyncIterator[None]

Share the transaction lock only with tasks that joined its ownership.

Source code in src/symfonic/memory/backends/in_memory_transaction.py 
52
53
54
55
56
57
58
59
60
@asynccontextmanager
async def writer(store: Any) -> AsyncIterator[None]:
    """Share the transaction lock only with tasks that joined its ownership."""
    if _owns(store):
        yield
    else:
        _refuse_expired_join(store)
        async with store._write_lock:  # noqa: SLF001
            yield