Skip to content

symfonic.capabilities.memory.publish

publish

How a cycle becomes durable, and what happens when it may not.

Split from :mod:symfonic.capabilities.memory.consolidation along the seam the design actually has: that module runs a roster, this one decides whether what the roster produced is allowed to land. The two questions have different readers -- a phase author needs the first, a reviewer of the atomicity contract needs the second -- and keeping them in one file made the runtime the longest module in the capability.

Everything here answers to :mod:.commit, which owns the rule that a cycle commits in exactly one transaction domain or is refused before it starts.

publish_cycle async 

publish_cycle(scope: MemoryScope, state: ConsolidationState, journal: CycleJournal, fence: Fence | None, domain: Any = None, writes: Any = None) -> tuple[str, ...]

Make this cycle's work durable -- all of it, or none of it.

Both halves go together: the graph mutations the phases made through the journal, and the records they staged through the write coordinator. Publishing them separately is the split this mechanism exists to close -- the runtime used to withhold the staged half on a failure while the direct half was already in the store, so a cycle that half-failed left a visible half.

A cycle that failed, was abandoned, or lost its lease discards the journal instead. What it did is then not merely unpublished, it never happened: the next clean cycle starts from the state this one found.

Source code in src/symfonic/capabilities/memory/publish.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
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
async def publish_cycle(
    scope: MemoryScope,
    state: ConsolidationState,
    journal: CycleJournal,
    fence: Fence | None,
    domain: Any = None,
    writes: Any = None,
) -> tuple[str, ...]:
    """Make this cycle's work durable -- all of it, or none of it.

    Both halves go together: the graph mutations the phases made through
    the journal, and the records they staged through the write
    coordinator. Publishing them separately is the split this mechanism
    exists to close -- the runtime used to withhold the staged half on a
    failure while the direct half was already in the store, so a cycle
    that half-failed left a visible half.

    A cycle that failed, was abandoned, or lost its lease discards the
    journal instead. What it did is then not merely unpublished, it never
    happened: the next clean cycle starts from the state this one found.
    """
    if not journal.complete:
        state.errors = (*state.errors, "cycle: bounded adjacency evidence was incomplete")
    if state.lease_lost or not state.clean:
        dropped = journal.discard()
        if dropped:
            logger.info(
                "consolidation cycle=%s scope=%s discarded %d graph "
                "mutation(s): %s",
                state.cycle.value,
                scope.path,
                dropped,
                "lease lost" if state.lease_lost else "a phase failed",
            )
        return ()
    flush = None
    if writes is not None:
        async def flush() -> tuple[str, ...]:  # noqa: RUF029 - closure shape
            receipt = await writes.flush(scope)
            return receipt.committed

    try:
        _written, published = await commit_cycle(
            journal, fence, flush, domain=domain
        )
    except LeaseLost as exc:
        state.lease_lost = True
        state.errors = (*state.errors, f"commit: {exc}")
        journal.discard()
        return ()
    return published

still_ours async 

still_ours(state: ConsolidationState, fence: Fence | None) -> None

Ask the fence once more, whatever the phases did or did not write.

Not redundant with the per-mutation checks: a cycle whose phases all read and none wrote reaches the end having never asked the fence anything, and reporting clean for a cycle that had no authority would tell a scheduler the scope was consolidated when another worker was in the middle of consolidating it.

Source code in src/symfonic/capabilities/memory/publish.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
async def still_ours(state: ConsolidationState, fence: Fence | None) -> None:
    """Ask the fence once more, whatever the phases did or did not write.

    Not redundant with the per-mutation checks: a cycle whose phases all
    read and none wrote reaches the end having never asked the fence
    anything, and reporting ``clean`` for a cycle that had no authority
    would tell a scheduler the scope was consolidated when another worker
    was in the middle of consolidating it.
    """
    if fence is None or state.lease_lost:
        return
    try:
        await fence.check("finishing the cycle")
    except LeaseLost as exc:
        state.lease_lost = True
        state.errors = (*state.errors, f"cycle: {exc}")