symfonic.capabilities.memory.mutations

mutations

One ordering point for memory publication and subtree erasure.

lifecycle_fence async

lifecycle_fence(graph: Any, vectors: Any, scope: MemoryScope, fence: Any, *, operation: Any = None, verb: Any = None) -> AsyncIterator[None]

Bound the review lock and one PG graph-only transaction together.

Source code in src/symfonic/capabilities/memory/mutations.py

@asynccontextmanager
async def lifecycle_fence(
    graph: Any, vectors: Any, scope: MemoryScope, fence: Any,
    *, operation: Any = None, verb: Any = None,
) -> AsyncIterator[None]:
    """Bound the review lock and one PG graph-only transaction together."""
    from symfonic.capabilities.memory.graph_rows import _adapter_scope
    from symfonic.memory.layers.procedural.review_lock import review_lock_for

    lock = review_lock_for(graph)
    if isinstance(fence, _ReceiptMutationFence):
        async with lock, fence.hold(scope, operation=operation, verb=verb):
            yield
        return
    native_pg = False
    if scope.identity_version and vectors is None and isinstance(fence, _PostgresMutationFence):
        from symfonic.capabilities.memory.graph_native_admission import exact_pg_graph

        native_pg = exact_pg_graph(graph)
    path = {"path": scope.path} if scope.identity_version == 1 else {}
    if not native_pg:
        async with lock, fence.hold(_adapter_scope(scope), **path):
            yield
        return
    from symfonic.memory.backends.postgres_native_subtree_query import NATIVE_SUBTREE_TIMEOUT_MS

    deadline_at = asyncio.get_running_loop().time() + NATIVE_SUBTREE_TIMEOUT_MS / 1000
    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:
        async with fence.hold(_adapter_scope(scope), deadline_at=deadline_at, **path):
            yield
    finally:
        lock.release()

mutation_fence_for

mutation_fence_for(graph: Any, vectors: Any = None) -> Any

Derive the strongest fence the composed stores can honestly offer.

Source code in src/symfonic/capabilities/memory/mutations.py

def mutation_fence_for(graph: Any, vectors: Any = None) -> Any:
    """Derive the strongest fence the composed stores can honestly offer."""
    from symfonic.capabilities.memory.commit import domain_of

    if type(graph).__module__ == "symfonic.memory.backends.mongo_native_receipt":
        from symfonic.memory.backends.mongo_native_read import read_snapshot_for

        read_snapshot_for(graph, vectors)
        return _NativeReadOnlyFence()

    if getattr(graph, "transaction_domain", None) is not None:
        from symfonic.capabilities.memory.receipt_read_dispatch import read_snapshot_for

        read_snapshot_for(graph, vectors)
        return _ReceiptMutationFence(graph.transaction_domain)
    domain = domain_of(graph)
    pool = domain if hasattr(domain, "raw_pool") else None
    vector_pool = getattr(vectors, "pool", None)
    if pool is not None:
        if vector_pool is not None and vector_pool is not pool:
            raise ConfigurationError(
                "graph and vector memory use different Postgres pools, so "
                "publication and erasure cannot share one ordering point"
            )
        return _PostgresMutationFence(pool)
    if vector_pool is not None:
        raise ConfigurationError(
            "graph and vector memory do not share one Postgres pool, so "
            "publication and erasure cannot share one ordering point"
        )
    if domain is not None:
        return _InProcessMutationFence(domain)
    if vectors is not None:
        raise ConfigurationError(
            "a vector-backed memory store needs a durable mutation fence; "
            f"{type(graph).__name__} publishes no pool or in-process transaction"
        )
    return _InProcessMutationFence(graph)