One-way journal seal and fixed-snapshot replay for native receipts.
replay
async
replay(journal: Any) -> int
Apply only the sealed snapshot when a receipt sealed this journal.
Source code in src/symfonic/capabilities/memory/journal_seal.py
| async def replay(journal: Any) -> int:
"""Apply only the sealed snapshot when a receipt sealed this journal."""
batches = journal._sealed
if batches is None:
batches = tuple((key, tuple(entries)) for key, entries in journal._ops.items())
written = 0
for _, entries in batches:
for durable, name, args, kwargs in entries:
result = await getattr(durable, name)(*args, **kwargs)
if name in {"update_owned_edge", "patch_owned_node"} and result is None:
journal._overlays[id(durable)][1].invalid = True
reason = ("edge owner changed before journal apply"
if name == "update_owned_edge"
else "patch_owned_node guard changed before journal apply")
raise MemoryContractError(reason)
if name == "_apply_edge" and result is False:
continue
written += 1
for durable, overlay in journal._overlays.values():
progress = getattr(overlay, "progress", {})
if journal._sealed is not None and progress:
raise MemoryContractError("sealed native receipt has staged progress")
for scope, expected, following in progress.values():
from symfonic.capabilities.memory.growth import following_progress
following = following_progress(overlay, scope, following, journal._edge_limit)
await durable.save_workset_progress(scope, journal._cycle, expected, following)
progress.clear()
journal._ops.clear()
journal._node_versions.clear()
if journal._sealed is not None:
journal._sealed = ()
return written
|
seal
seal(journal: Any) -> Operations
Capture all staged operations without yielding to an inherited task.
Source code in src/symfonic/capabilities/memory/journal_seal.py
| def seal(journal: Any) -> Operations:
"""Capture all staged operations without yielding to an inherited task."""
journal._require_open()
operations: Operations = tuple(
(key, tuple(
(entry[0], entry[1], deepcopy(entry[2]), deepcopy(entry[3]))
if type(entry) is tuple and len(entry) == 4 else entry
for entry in entries
))
for key, entries in journal._ops.items()
)
journal._sealed = operations
return operations
|