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, *, operation: MemoryOperation | None = 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

async def publish_cycle(
    scope: MemoryScope,
    state: ConsolidationState,
    journal: CycleJournal,
    fence: Fence | None,
    domain: Any = None,
    writes: Any = None,
    *, operation: MemoryOperation | None = 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.
    """
    validate_transaction_operation(
        domain, operation, MemoryOperationVerb.COMMIT_CYCLE, scope,
    )
    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
    native_outer = False
    if writes is not None:
        # A background graph write needs the store lock; join before the
        # transaction takes that lock, then keep the flush inside it.
        await writes.join()
        native_outer = transactions._native_graph_cycle(domain, scope, writes)

        async def flush() -> tuple[str, ...]:  # noqa: RUF029 - closure shape
            if native_outer:
                receipt = await writes.flush(scope, join=False, native_outer=True)
            else:
                receipt = await writes.flush(scope, join=False)
            return receipt.committed

    try:
        if native_outer:
            from symfonic.memory.backends.postgres_native_subtree_query import (
                NATIVE_SUBTREE_TIMEOUT_MS,
            )
            from symfonic.memory.layers.procedural.review_lock import review_lock_for

            deadline_at = asyncio.get_running_loop().time() + NATIVE_SUBTREE_TIMEOUT_MS / 1000
            lock = review_lock_for(writes._target.lifecycle.graph)
            try:
                async with asyncio.timeout_at(deadline_at):
                    await lock.acquire()
            except TimeoutError as exc:
                raise MemoryContractError(
                    "native lifecycle execution time budget exceeded"
                ) from exc
            try:
                _written, published = await commit_cycle(
                    journal, fence, flush, domain=domain,
                    native_scope=scope, native_writes=writes,
                    native_deadline_at=deadline_at, operation=operation,
                )
                writes.confirm_native(published)
            finally:
                lock.release()
        else:
            _written, published = await commit_cycle(
                journal, fence, flush, domain=domain,
                native_scope=scope, native_writes=writes, operation=operation,
            )
    except LeaseLost as exc:
        state.lease_lost = True
        state.errors = (*state.errors, f"commit: {exc}")
        journal.discard()
        return ()
    except (EdgeOwnershipConflict, VectorOwnershipConflict) as exc:
        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

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}")