symfonic.capabilities.memory.leases

leases

One consolidation per scope, across processes rather than within one.

A coordinator holding an asyncio.Lock per scope is green in every test and wrong in production. The shipped topology is Celery beat firing a nightly job while an operator runs the manual backfill, and those are different processes: an in-process lock coordinates neither of them with the other, and two Deep Sleeps run over one scope while every test that could have caught it passes.

So the exclusion is a port, and the thing it is keyed on is the scope.

A lease, not a lock. A lock is held until released, which means a worker that is killed holds it forever and that scope never consolidates again. A lease expires, so the failure mode of a dead worker is a delay rather than a permanent outage.

With an owner token, because expiry creates a second holder. Once a lease can expire, two workers can believe they hold the same scope: the first is merely slow, the second took over after the deadline. Release therefore names who is releasing, and a release from the previous holder is refused rather than applied -- otherwise the slow worker's finally block frees the new holder's lease and a third worker walks in.

Renewable, because a slow worker is not a dead one. Deep Sleep can run for minutes and five of its phases may call a model. Without renewal the TTL has to be longer than the slowest imaginable cycle -- which makes a real crash cost that long -- or a living worker loses its scope to its own model call, and expiry stops meaning "this worker died". :meth:LeasePort.renew re-stamps the deadline while the work is going, so expiry means only what it should.

And a fence, because renewal can still fail. A suspended process, an unreachable database, a worker that genuinely stalled: then the lease does lapse mid-cycle, and the danger is that the worker carries on writing under an authority it no longer has. :meth:LeasePort.holds is the check, and :class:~.fencing.FencedGraph is where it is made -- before every durable mutation, rather than once before publishing. Ten of the roster's phase modules write to the graph directly and never touch the write coordinator, so a fence at the end would have guarded almost nothing.

InProcessLeases

InProcessLeases(*, single_process: bool, now: Any = None)

A lease table in this process's memory. Single-process use only.

Correct for a dev server, a test, and a deployment that genuinely runs one worker -- and silently wrong for every other, which is why it takes single_process=True rather than defaulting to convenient. The scaffold composes the Postgres adapter; this one exists so a laptop does not need a database to run a cycle, and so the port has an implementation whose behaviour a test can pin without one.

now is injected for the same reason the schedule's is: a test that had to sleep through a TTL would be a slow test asserting a timeout.

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

def __init__(self, *, single_process: bool, now: Any = None) -> None:
    if not single_process:
        raise MemoryContractError(
            "InProcessLeases coordinates one process and this deployment "
            "did not declare itself to be one. Celery beat and a manual "
            "backfill are different processes, and an in-memory table "
            "excludes neither from the other -- so two Deep Sleeps would "
            "run over one scope while every test passed. Pass "
            "single_process=True to say you mean it, or compose the "
            "Postgres adapter."
        )
    import time

    self._now = now or time.monotonic
    self._held: dict[str, tuple[str, float]] = {}

hold_for_update async

hold_for_update(lease: Lease) -> bool

The same answer as :meth:holds, and correctly so.

One process, one event loop, and a batch this table's owner applies without awaiting anything that yields: there is no instant between the check and the write for a rival to occupy. The lock PostgresLeases needs exists because there the rival is another process.

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

async def hold_for_update(self, lease: Lease) -> bool:
    """The same answer as :meth:`holds`, and correctly so.

    One process, one event loop, and a batch this table's owner applies
    without awaiting anything that yields: there is no instant between the
    check and the write for a rival to occupy. The lock ``PostgresLeases``
    needs exists because *there* the rival is another process.
    """
    return await self.holds(lease)

Lease dataclass

Lease(scope_path: str, owner: str, ttl_seconds: float)

One scope, held by one owner, until one deadline.

LeasePort

Bases: Protocol

Exclusion for one scope, across whatever processes serve it.

acquire async

acquire(scope: MemoryScope, *, owner: str, ttl_seconds: float) -> Lease | None

Take the lease for scope, or None if someone else holds it.

None is the loser's answer and it is not an error: the scope is being consolidated right now, by somebody, which is the outcome asked for. An expired lease is available -- that is what makes a dead worker a delay rather than an outage.

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

async def acquire(self, scope: MemoryScope, *, owner: str, ttl_seconds: float) -> Lease | None:
    """Take the lease for ``scope``, or ``None`` if someone else holds it.

    ``None`` is the loser's answer and it is not an error: the scope is
    being consolidated right now, by somebody, which is the outcome asked
    for. An expired lease is available -- that is what makes a dead worker
    a delay rather than an outage.
    """
    ...

hold_for_update async

hold_for_update(lease: Lease) -> bool

Whether lease is ours, and keep it ours until the transaction ends.

The commit-time fence, and a different question from :meth:holds. holds answers about the instant it ran: a caller that then writes has a window in which the lease can lapse and a rival can take it, and running both on one connection does not close it -- the window is between the check and the commit. This one takes a lock the rival's acquisition must wait for, so the answer is still true when the batch lands.

Only meaningful inside a transaction. An implementation with no transactions to speak of may return :meth:holds, but a deployment composed on one is refused before a cycle runs rather than told afterwards that "atomic" meant something weaker.

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

async def hold_for_update(self, lease: Lease) -> bool:
    """Whether ``lease`` is ours, and keep it ours until the transaction ends.

    The commit-time fence, and a different question from :meth:`holds`.
    ``holds`` answers about the instant it ran: a caller that then writes
    has a window in which the lease can lapse and a rival can take it, and
    running both on one connection does not close it -- the window is
    between the check and the commit. This one takes a lock the rival's
    acquisition must wait for, so the answer is still true when the batch
    lands.

    Only meaningful inside a transaction. An implementation with no
    transactions to speak of may return :meth:`holds`, but a deployment
    composed on one is refused before a cycle runs rather than told
    afterwards that "atomic" meant something weaker.
    """
    ...

holds async

holds(lease: Lease) -> bool

Whether lease is still this owner's, right now.

The fence. A holder calls it before a mutation it cannot take back, so a worker whose lease expired mid-cycle stops rather than writing under an authority it lost.

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

async def holds(self, lease: Lease) -> bool:
    """Whether ``lease`` is still this owner's, right now.

    The fence. A holder calls it before a mutation it cannot take back, so
    a worker whose lease expired mid-cycle stops rather than writing under
    an authority it lost.
    """
    ...

release async

release(lease: Lease) -> bool

Give up lease. False when it was not this owner's to give.

Refused rather than applied, because by the time a slow worker reaches its finally the lease may belong to whoever took over -- and releasing it there would hand a third worker a scope two are already writing to.

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

async def release(self, lease: Lease) -> bool:
    """Give up ``lease``. ``False`` when it was not this owner's to give.

    Refused rather than applied, because by the time a slow worker reaches
    its ``finally`` the lease may belong to whoever took over -- and
    releasing it there would hand a third worker a scope two are already
    writing to.
    """
    ...

renew async

renew(lease: Lease) -> bool

Push lease's deadline out by its own TTL. False if it lapsed.

Owner-checked like :meth:release: a worker whose scope was taken over must not extend the deadline of whoever holds it now. False is the answer that says "you no longer have this", and a heartbeat that gets it should stop rather than retry -- the scope is somebody else's.

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

async def renew(self, lease: Lease) -> bool:
    """Push ``lease``'s deadline out by its own TTL. ``False`` if it lapsed.

    Owner-checked like :meth:`release`: a worker whose scope was taken over
    must not extend the deadline of whoever holds it now. ``False`` is the
    answer that says "you no longer have this", and a heartbeat that gets
    it should stop rather than retry -- the scope is somebody else's.
    """
    ...

new_owner_token

new_owner_token() -> str

A token this holder can prove and another cannot guess.

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

def new_owner_token() -> str:
    """A token this holder can prove and another cannot guess."""
    return uuid.uuid4().hex