Skip to content

symfonic.capabilities.memory.journal_seal

journal_seal

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 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
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 
14
15
16
17
18
19
20
21
22
23
24
25
26
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