symfonic.capabilities.memory.commit

commit

One transaction, one lease lock, one cycle -- or a refusal at composition.

The journal makes a cycle's mutations arrive together. Two things were still missing from "atomic", and both were the kind of gap that reads as safe:

A plain read is not authority. Checking the lease with a bare SELECT answers about the instant it ran. The cycle then applies its batch, and between those two moments the lease can expire and another worker can acquire the row and start work of its own. Running the check on the same connection as the writes does not close that -- the window is between the check and the commit, not between connections. So the check takes the row lock (:meth:~symfonic.capabilities.memory.leases.LeasePort.hold_for_update) and holds it until the transaction ends, which makes a rival's acquisition queue behind the commit instead of slipping inside it.

"Atomic" must not degrade quietly. The first version accepted whatever it was given: a store with no transaction had its operations applied one after another, so a failure on the third left the first two behind; two stores on different pools got two independent transactions, and the inner one could commit while the outer rolled back. Both are best-effort wearing the word atomic, which is worse than not claiming it.

So a cycle now runs in exactly one transactional domain, established at composition and verified here. Everything that participates -- the graph the phases mutate, the lease that authorises them, the records the write coordinator publishes -- must belong to it. A composition that cannot offer one is refused when the coordinator is built, before any cycle runs, rather than producing a report that says clean about a half-applied batch.

The one domain that is not a database is a single in-process store, and it offers a real transaction too: :class:~symfonic.memory.backends.in_memory_transaction.SnapshotTransaction restores its tables if the batch raises. Non-interleaving alone was never enough -- a batch whose third operation raises has already applied the first two -- so the in-process domain is accepted because it can roll back, not because nothing else can run.

commit_cycle async

commit_cycle(journal: Any, fence: Any, flush: Any = None, *, domain: Any = None) -> tuple[int, tuple[str, ...]]

Check the lease under a lock and write the batch, in one transaction.

Parameters:

Name	Type	Description	Default
journal	Any	the cycle's deferred mutations.	required
fence	Any	authority. None for a cycle running without a lease.	required
flush	Any	async () -> tuple[str, ...], publishing the staged records.	None
domain	Any	the transaction domain established at composition. Required whenever there is anything to write: without it there is no statement to make about atomicity.	None

Returns:

Type	Description
tuple[int, tuple[str, ...]]	(mutations_written, published_record_ids).

Raises:

Type	Description
LeaseLost	the scope is no longer this cycle's. The transaction rolls back, so nothing is written.
MemoryContractError	the journal touched a store outside domain.

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

async def commit_cycle(
    journal: Any, fence: Any, flush: Any = None, *, domain: Any = None
) -> tuple[int, tuple[str, ...]]:
    """Check the lease under a lock and write the batch, in one transaction.

    Args:
        journal: the cycle's deferred mutations.
        fence: authority. ``None`` for a cycle running without a lease.
        flush: ``async () -> tuple[str, ...]``, publishing the staged records.
        domain: the transaction domain established at composition. Required
            whenever there is anything to write: without it there is no
            statement to make about atomicity.

    Returns:
        ``(mutations_written, published_record_ids)``.

    Raises:
        LeaseLost: the scope is no longer this cycle's. The transaction rolls
            back, so nothing is written.
        MemoryContractError: the journal touched a store outside ``domain``.
    """
    if journal.pending == 0 and flush is None:
        # Nothing to write, so nothing to be atomic about. Authority is still
        # checked by the caller, which is what makes a read-only cycle that
        # lost its scope report ``lease_lost`` rather than ``clean``.
        return 0, ()
    # The stores the cycle actually wrote to must agree with each other and
    # with whatever composition declared. Resolved to their *domains* rather
    # than used directly: a Postgres backend's rollback belongs to its pool,
    # not to the backend object.
    touched = require_one_domain(
        {f"graph store #{index}": store for index, store in enumerate(journal.stores())}
    )
    if domain is None:
        domain = touched
    elif touched is not None and touched is not domain:
        raise MemoryContractError(
            "this cycle wrote to a store outside the transaction domain it was "
            "composed with, so its batch could not be committed with the rest. "
            "The composition-time check should have refused this; reaching here "
            "means a phase was handed a graph nobody declared."
        )
    async with one_transaction(domain), journalled(None):
        # Unbound for the duration: applying the batch must reach the store, and
        # the staged flush writes through the very backend the journal wraps.
        # Left bound, the flush would be deferred into a journal that has just
        # been emptied and its records would silently never land.
        if fence is not None:
            # Inside the transaction, and holding the lease row: a rival's
            # acquisition blocks behind this commit rather than landing between
            # the answer and the writes it authorised.
            await fence.check_locked("applying this cycle's mutations")
        written = await journal.apply()
        published = await flush() if flush is not None else ()
    return written, published

domain_of

domain_of(participant: Any) -> Any

The transaction domain participant belongs to.

Walks the public surface -- pool, durable, backend, graph, unfenced -- because the participants are wrappers over wrappers: a fenced view of a store over a journalled backend over a pool. Every hop is a published property, so this is a walk over the API rather than over somebody's private attributes.

Returns the object that owns the rollback -- a connection pool, or a store that can open its own transaction -- or None when no such owner can be found, which callers must treat as a refusal rather than as "probably fine".

An object, never a label. A shared "in-process" string collapsed two different in-memory stores into one fictitious domain: a graph on backend A and a write coordinator on backend B both answered "in-process", the composition was accepted, and the commit snapshotted A while the flush wrote to B -- so a failure rolled back half of it. Identity is the whole property being asserted, so identity is what this returns.

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

def domain_of(participant: Any) -> Any:
    """The transaction domain ``participant`` belongs to.

    Walks the public surface -- ``pool``, ``durable``, ``backend``, ``graph``,
    ``unfenced`` -- because the participants are wrappers over wrappers: a
    fenced view of a store over a journalled backend over a pool. Every hop is
    a published property, so this is a walk over the API rather than over
    somebody's private attributes.

    Returns **the object that owns the rollback** -- a connection pool, or a
    store that can open its own transaction -- or ``None`` when no such owner
    can be found, which callers must treat as a refusal rather than as
    "probably fine".

    An object, never a label. A shared ``"in-process"`` string collapsed two
    different in-memory stores into one fictitious domain: a graph on backend A
    and a write coordinator on backend B both answered "in-process", the
    composition was accepted, and the commit snapshotted A while the flush
    wrote to B -- so a failure rolled back half of it. Identity is the whole
    property being asserted, so identity is what this returns.
    """
    seen: list[Any] = []
    current = participant
    for _ in range(8):  # wrappers are shallow; the bound stops a cycle of them
        if current is None or any(current is item for item in seen):
            return None
        seen.append(current)
        pool = getattr(current, "pool", None)
        if pool is not None:
            return pool
        for hop in ("unfenced", "durable", "graph", "backend"):
            nxt = getattr(current, hop, None)
            if nxt is not None and nxt is not current:
                current = nxt
                break
        else:
            # No further wrapper. A store that can open a transaction of its
            # own is its own domain, and *is* it: two in-memory backends are
            # two domains, not one. Tested by capability rather than by class
            # name -- what matters is that a failed batch can be rolled back.
            #
            # Asked only once the unwrapping is exhausted, because a wrapper
            # forwards ``transaction`` to what it wraps: asking earlier made
            # every wrapper its own domain, so two views of one backend looked
            # like two stores that could not commit together.
            return current if callable(getattr(current, "transaction", None)) else None
    return None

require_domain_for

require_domain_for(leases: Any, runtime: Any, declared: Any) -> Any

The domain this coordinator commits in, or a refusal explaining why not.

Established here because here is where a deployment is composed. A cycle that discovered at commit time that its lease and its graph were on different pools would already have run seventeen phases and spent whatever model calls they make, and the only honest thing left to report would be that none of it could be applied.

Every participant is compared, including the in-process ones. An earlier version skipped those, on the reasoning that a single process needs no coordination -- which let a Postgres transaction= sit beside phases on an in-memory graph, and let a graph on one in-memory backend sit beside a write coordinator on another. Neither pair shares a rollback.

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

def require_domain_for(leases: Any, runtime: Any, declared: Any) -> Any:
    """The domain this coordinator commits in, or a refusal explaining why not.

    Established here because here is where a deployment is composed. A cycle
    that discovered at commit time that its lease and its graph were on
    different pools would already have run seventeen phases and spent whatever
    model calls they make, and the only honest thing left to report would be
    that none of it could be applied.

    Every participant is compared, including the in-process ones. An earlier
    version skipped those, on the reasoning that a single process needs no
    coordination -- which let a Postgres ``transaction=`` sit beside phases on
    an in-memory graph, and let a graph on one in-memory backend sit beside a
    write coordinator on another. Neither pair shares a rollback.
    """
    stores = _store_participants(runtime)
    store_domain = require_one_domain(stores)
    lease_domain = domain_of(leases)
    lease_is_pooled = lease_domain is not None and lease_domain is not leases

    if lease_is_pooled and declared is None:
        raise MemoryContractError(
            "this consolidation coordinator has a lease port with a connection "
            "pool but was given no ``transaction=`` domain, so its cycles could "
            "not commit atomically: the graph mutations, the lease check and the "
            "staged records would land as separate writes and a failure partway "
            "would leave the earlier ones behind. Pass the same pool the graph "
            "backend and the lease port use."
        )
    if lease_is_pooled and declared is not None and lease_domain is not declared:
        raise MemoryContractError(
            "the lease port and the declared transaction domain are different "
            "pools. Nothing can hold the lease row and write the batch together "
            "across two of them, so the check-to-write race the lock exists to "
            "close would stay open."
        )

    domain = declared if declared is not None else store_domain
    if domain is None:
        domain = lease_domain if lease_is_pooled else None
    if store_domain is not None and domain is not None and store_domain is not domain:
        raise MemoryContractError(
            f"the {stores.odd_one} is in a different transaction domain from "
            "the one this cycle commits in, so it could not land in the same "
            "transaction as the rest of the cycle: one half would commit and "
            "the other roll back. Compose the graph, the lease and the write "
            "coordinator on one pool -- or, in one process, on one store."
        )
    return domain

require_one_domain

require_one_domain(participants: dict[str, Any]) -> Any

The single domain every participant shares, or a refusal naming the odd one.

Parameters:

Name	Type	Description	Default
participants	dict[str, Any]	role -> object, so the error can say which one broke the requirement. "two domains" sends a reader looking; "the lease is on a different pool from the graph" does not.	required

Raises:

Type	Description
MemoryContractError	when a domain cannot be established, or when the participants span more than one.

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

def require_one_domain(participants: dict[str, Any]) -> Any:
    """The single domain every participant shares, or a refusal naming the odd one.

    Args:
        participants: role -> object, so the error can say *which* one broke
            the requirement. "two domains" sends a reader looking; "the lease
            is on a different pool from the graph" does not.

    Raises:
        MemoryContractError: when a domain cannot be established, or when the
            participants span more than one.
    """
    domains: list[tuple[str, Any]] = []
    for role, participant in participants.items():
        if participant is None:
            continue
        domain = domain_of(participant)
        if domain is None:
            raise MemoryContractError(
                f"the {role} ({type(participant).__name__}) does not declare a "
                "transaction domain, so a consolidation cycle over it cannot be "
                "atomic: a batch would be applied one operation at a time and a "
                "failure partway would leave the earlier ones behind. Publish a "
                "``pool`` property naming the pool it commits through, or "
                "compose the cycle on a store that has one."
            )
        domains.append((role, domain))
    if not domains:
        return None
    first_role, first = domains[0]
    for role, domain in domains[1:]:
        if domain is not first:
            raise MemoryContractError(
                f"the {role} and the {first_role} are in different transaction "
                "domains, so nothing can commit them together: one would land "
                "and the other roll back. A consolidation cycle needs its "
                "graph, its lease and its staged records on one pool -- or, "
                "in one process, on one and the same store. Two in-memory "
                "backends are two domains."
            )
    return first