symfonic.memory.backends.postgres_staged_identity

postgres_staged_identity

Compose a staged writer and its publisher on exact, distinct database logins.

CORE-376 makes PostgreSQL refuse publication by the writer role, but only for a turn that actually runs as that role, against a database where the boundary is installed. This module is the composition seam that refuses anything else:

• staged_writer_journal builds the staged journal for a turn process. On a connection from the very pool the journal will use, it requires that session_user and current_user both equal the writer role and that require_boundary(writer, publisher) holds. It takes the publisher's role name, which is not a secret, and no publisher credential.
• StagedPublisherBinding composes PostgresStagedPublisher for an activation process. It requires the publisher login and the boundary when composed and again before every activation and every activation_status read, so a journal on the writer's pool, a dropped policy or a changed login refuses instead of publishing or reporting.

Scope. Composition attests one connection. Bound status and activation reads attest their own snapshot transaction and hold boundary table locks through the read; the publication transaction independently attests its proof. It cannot tell whether both credentials were loaded into one process: process separation, where the publisher DSN lives and how activation is triggered are deployment decisions outside this library. install is never called here; the boundary is an operator step run as the table owner.

StagedPublisherBinding

StagedPublisherBinding(publisher: Any, *, writer: str, role: str)

A PostgresStagedPublisher that only activates on the publisher login.

Build it with await StagedPublisherBinding.compose(...); the reader of committed local final receipts is bound there and nowhere else.

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

def __init__(self, publisher: Any, *, writer: str, role: str) -> None:
    from symfonic.memory.backends.postgres_staged_activation import (
        PostgresStagedPublisher,
    )

    if type(publisher) is not PostgresStagedPublisher:
        _refuse("an exact PostgresStagedPublisher is required")
    # The attested pool is the one activation writes through, never a second.
    self._pool = _pool_of(publisher._journal._domain)
    self._publisher, self._writer, self._role = publisher, writer, role

activate async

activate(invocation_id: Any) -> Any

Re-attest the login and the boundary, then activate one READY turn.

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

async def activate(self, invocation_id: Any) -> Any:
    """Re-attest the login and the boundary, then activate one READY turn."""
    return await self._publisher.activate(invocation_id, _snapshot=self._snapshot)

activation_status async

activation_status(invocation_id: Any) -> Any

Re-attest, then read one turn of the bound scope; never publishes.

A raw journal snapshot reports whatever the rows say, including an ACTIVE forged after the boundary was dropped. This answers only while the publisher login and the boundary still hold, and only for the scope fixed at composition.

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

async def activation_status(self, invocation_id: Any) -> Any:
    """Re-attest, then read one turn of the bound scope; never publishes.

    A raw journal snapshot reports whatever the rows say, including an
    ACTIVE forged after the boundary was dropped. This answers only while
    the publisher login and the boundary still hold, and only for the scope
    fixed at composition.
    """
    publisher = self._publisher
    return await self._snapshot(publisher._journal, invocation_id, publisher._scope)

require_login async

require_login(pool: Any, expected: str, *, writer: str, publisher: str) -> None

expected is both the session and the current role, behind the boundary.

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

async def require_login(pool: Any, expected: str, *, writer: str, publisher: str) -> None:
    """``expected`` is both the session and the current role, behind the boundary."""
    conn = await pool.acquire()
    try:
        await require_login_on(conn, expected, writer=writer, publisher=publisher)
    finally:
        await pool.release(conn)

require_login_on async

require_login_on(conn: Any, expected: str, *, writer: str, publisher: str) -> None

Attest on the connection whose transaction will consume the result.

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

async def require_login_on(conn: Any, expected: str, *, writer: str, publisher: str) -> None:
    """Attest on the connection whose transaction will consume the result."""
    if expected not in (writer, publisher):
        _refuse("the expected login must be the writer or the publisher")
    await require_boundary(conn, writer, publisher)  # names refuse before SQL
    row = await conn.fetchrow(_LOGIN)
    if row is None or row["session_name"] != expected or row["acting_name"] != expected:
        _refuse(f"pool does not log in exactly as {expected}")

staged_writer_journal async

staged_writer_journal(domain: Any, *, cipher: Any, writer: str, publisher: str) -> Any

The staged journal of a turn process, bound to the attested writer login.

No schema work is done: the writer may not run DDL, so the staged schema, the claim fence and the boundary must already exist.

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

async def staged_writer_journal(domain: Any, *, cipher: Any, writer: str,
                                publisher: str) -> Any:
    """The staged journal of a turn process, bound to the attested writer login.

    No schema work is done: the writer may not run DDL, so the staged schema,
    the claim fence and the boundary must already exist.
    """
    from symfonic.memory.backends.postgres_operation_journal import PostgresOperationJournal

    await require_login(_pool_of(domain), writer, writer=writer, publisher=publisher)
    return PostgresOperationJournal(domain, cipher=cipher, staged_publication=True)