symfonic.memory.backends.postgres_leases

postgres_leases

The consolidation lease, in the one place every worker can see it.

The exclusion Deep Sleep needs is between processes -- Celery beat and a manual backfill are different ones -- so the table is the coordination point, not any object in any of them.

One statement decides. Acquisition is a single INSERT ... ON CONFLICT DO UPDATE ... WHERE expires_at <= now(): the row is the lock, Postgres serialises the writers, and the RETURNING clause tells the caller whether it won. Two workers racing produce one insert and one no-op, with no read-then-write window between them for a third to slip through.

now() is the database's. Two workers whose clocks disagree would otherwise disagree about whether a lease had expired -- the one running fast would take over a lease the holder still believes it has. Every comparison here happens server-side, so there is one clock.

Expiry is the crash recovery. A worker killed mid-cycle leaves its row behind; the next acquisition after expires_at takes it. That is why the condition is a timestamp rather than a deleted row: nothing has to run to clean up after a process that stopped existing.

PostgresLeases

PostgresLeases(pool: Any)

A :class:~symfonic.capabilities.memory.leases.LeasePort over Postgres.

Source code in src/symfonic/memory/backends/postgres_leases.py

def __init__(self, pool: Any) -> None:
    self._pool = pool

pool property

pool: Any

The pool this port speaks through, and so its transaction domain.

Public because the commit path has to prove that the lease, the graph and the staged records all live in one domain before it will call a cycle atomic. Reading that off a private attribute would make the proof depend on an implementation detail of this class.

ensure_schema async

ensure_schema() -> None

Create the lease table if it is not there.

IF NOT EXISTS is idempotent across time and not across concurrency: two sessions running it at the same instant both pass the existence check and both insert into pg_type, and one gets a duplicate-key error. Ordinarily an obscure wrinkle -- here it is the expected case, because the workers this table exists to coordinate are exactly the ones that start together.

So a failure is retried once against a table that exists by then, and only then re-raised. Caught by the first run of the cross-process test and by nothing else: every later run found the table already there.

Source code in src/symfonic/memory/backends/postgres_leases.py

async def ensure_schema(self) -> None:
    """Create the lease table if it is not there.

    ``IF NOT EXISTS`` is idempotent across time and *not* across
    concurrency: two sessions running it at the same instant both pass the
    existence check and both insert into ``pg_type``, and one gets a
    duplicate-key error. Ordinarily an obscure wrinkle -- here it is the
    expected case, because the workers this table exists to coordinate are
    exactly the ones that start together.

    So a failure is retried once against a table that exists by then, and
    only then re-raised. Caught by the first run of the cross-process test
    and by nothing else: every later run found the table already there.
    """
    for attempt in (1, 2):
        conn = await self._pool.acquire()
        try:
            await conn.execute(LEASES_DDL)
            return
        except Exception:
            if attempt == 2:
                raise
            # Somebody else is creating it right now. The retry either
            # finds it there or fails for a reason that is not a race.
            logger.debug(
                "lease table creation raced another worker; retrying once",
                exc_info=True,
            )
        finally:
            await self._pool.release(conn)

hold_for_update async

hold_for_update(lease: Lease) -> bool

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

The commit-time fence. :meth:holds answers a question about the past tense the moment it returns; this one takes the row lock, so a rival acquisition queues behind the caller's transaction instead of landing between the answer and the write it authorised.

Only meaningful inside a transaction -- outside one the lock is released with the statement and this is :meth:holds with extra cost. The commit path always calls it inside one; see :func:symfonic.capabilities.memory.commit.commit_cycle, which refuses to run without a transaction domain rather than calling this and pretending.

Source code in src/symfonic/memory/backends/postgres_leases.py

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

    The commit-time fence. :meth:`holds` answers a question about the past
    tense the moment it returns; this one takes the row lock, so a rival
    acquisition queues behind the caller's transaction instead of landing
    between the answer and the write it authorised.

    Only meaningful inside a transaction -- outside one the lock is released
    with the statement and this is :meth:`holds` with extra cost. The commit
    path always calls it inside one; see
    :func:`symfonic.capabilities.memory.commit.commit_cycle`, which refuses
    to run without a transaction domain rather than calling this and
    pretending.
    """
    conn = await self._pool.acquire()
    try:
        row = await conn.fetchval(
            _HOLDS_FOR_UPDATE, lease.scope_path, lease.owner
        )
    finally:
        await self._pool.release(conn)
    return row is not None