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 tenant lock decides. Acquisition takes a transaction-scoped advisory lock for the tenant, then inserts only when no live ancestor or descendant row exists. The lock serialises different scope rows that an exact-key conflict cannot coordinate. Sibling scopes remain independent holders.

The clock 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. Acquisition samples clock_timestamp() after waiting for the tenant lock; now() would still mean when its transaction began.

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 table and native index under a schema-scoped DDL lock.

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 table and native index under a schema-scoped DDL lock.

    ``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:
            async with conn.transaction():
                # Functions and expression indexes race just like tables.
                await conn.execute(
                    "SELECT pg_advisory_xact_lock(2147483646, hashtext(current_schema()))"
                )
                await conn.execute(LEASES_DDL)
                await conn.execute(NATIVE_INDEX_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.
    """
    scope = scope_from_path(lease.scope_path)
    conn = await self._pool.acquire()
    try:
        # Acquisition of a child row also needs to wait for this commit.
        await conn.execute(_LOCK_TENANT, scope.tenant)
        row = await conn.fetchval(
            _HOLDS_FOR_UPDATE, lease.scope_path, lease.owner
        )
    finally:
        await self._pool.release(conn)
    return row is not None