Durable activation provenance for one staged PostgreSQL v1 turn (CORE-339).
An exact COMMITTED FLUSH receipt is not, by itself, evidence of activation: a
writer that held receipt write access while the role boundary was down can
insert one, and it survives the boundary's return. ACTIVE therefore also needs
a proof row, keyed by tenant and invocation, that binds the sealed READY row
(scope, run, output digest and a hash of the sealed bytes) to the exact FLUSH
operation id and intent digest. The hash of the sealed bytes must equal that
of the READY row whose per-record content seal (CORE-339a,
postgres_staged_content) the same transaction verified against the rows.
Only the receipt domain writes a proof, and only inside the transaction that
moves the armed FLUSH receipt from PENDING to COMMITTED: after the selected
publication completed and before the receipt update, so the graph release, the
vector writes, the proof and the receipt commit or roll back together. Before
inserting, that same transaction attests who is activating:
owner: the acting role is a superuser or owns the receipt table, and no
staged boundary policy is installed. The owner is trusted in any case.boundary: the boundary is installed; its writer is read from the policy
the installer created, and require_boundary(writer, current_user) holds,
so the activator is not the writer and the writer cannot read proofs.
Either way session_user must equal current_user and PUBLIC must hold no
write privilege on the proof table. Proof rows are never derived from a
receipt alone: a COMMITTED receipt without a matching proof, including every
receipt committed before this table existed, reads as UNKNOWN.
Not covered: a role an operator grants INSERT on the proof table, other than
the writer or PUBLIC, can forge a proof, exactly as it could release rows.
StagedProof
Bases: NamedTuple
What one activation arms: the exact staged FLUSH, its READY and content seal.
attest
async
attest(conn: Any) -> tuple[str, str, str | None]
(mode, activator, writer) for the acting login; any doubt refuses.
Source code in src/symfonic/memory/backends/postgres_staged_proof.py
| async def attest(conn: Any) -> tuple[str, str, str | None]:
"""``(mode, activator, writer)`` for the acting login; any doubt refuses."""
from symfonic.memory.backends.postgres_staged_roles import require_boundary
row = await conn.fetchrow(_WHO, _RECEIPTS, _WRITER_POLICY)
if row is None or row["session_name"] != row["acting_name"]:
_refuse("the activator must act as its own session login")
if await conn.fetchval(_PUBLIC_WRITE, PROOFS) is not False:
_refuse("PUBLIC may write activation proofs")
acting, writers = row["acting_name"], row["writers"]
if not row["bounded"]:
if row["trusted"] is not True:
_refuse("a non-owner activator requires the installed role boundary")
return "owner", acting, None
if writers is None or len(writers) != 1:
_refuse("the installed boundary does not name exactly one writer")
await require_boundary(conn, writers[0], acting) # refuses acting == writer
return "boundary", acting, writers[0]
|
ddl
ddl(staged_table: str) -> tuple[str, ...]
The proof table; it references the sealed READY row it proves.
Source code in src/symfonic/memory/backends/postgres_staged_proof.py
| def ddl(staged_table: str) -> tuple[str, ...]:
"""The proof table; it references the sealed READY row it proves."""
return (
f"""CREATE TABLE IF NOT EXISTS {PROOFS} (
tenant_id text NOT NULL,
invocation_id text NOT NULL,
operation_id text NOT NULL,
intent_digest text NOT NULL,
scope_levels text NOT NULL,
run_id text NOT NULL,
output_sha256 text NOT NULL,
sealed_sha256 text NOT NULL,
activator text NOT NULL,
attestation text NOT NULL CHECK (attestation IN ('owner', 'boundary')),
writer text,
PRIMARY KEY (tenant_id, invocation_id),
FOREIGN KEY (tenant_id, invocation_id) REFERENCES {staged_table}
)""",
f"REVOKE ALL ON TABLE {PROOFS} FROM PUBLIC",
)
|
matches
async
matches(conn: Any, expected: Any, stored: Any, ready: ReadyTurn, operation: MemoryOperation) -> bool
Exactly one proof binds this sealed READY row to this exact FLUSH.
Source code in src/symfonic/memory/backends/postgres_staged_proof.py
| async def matches(conn: Any, expected: Any, stored: Any, ready: ReadyTurn,
operation: MemoryOperation) -> bool:
"""Exactly one proof binds this sealed READY row to this exact FLUSH."""
from symfonic.memory.backends.postgres_journal_staged import _scope_key
# The turn writer must not read proof metadata across tenants. Its raw
# snapshot remains UNKNOWN; only the bound publisher can attest ACTIVE.
if await conn.fetchval("SELECT has_table_privilege(current_user, $1::regclass, 'SELECT')",
PROOFS) is not True:
return False
row = await conn.fetchrow(
f"SELECT * FROM {PROOFS} WHERE tenant_id=$1 AND invocation_id=$2",
expected["tenant"], expected["invocation"],
)
if row is None:
return False
sealed = hashlib.sha256(bytes(stored["sealed"])).hexdigest()
return (
row["operation_id"] == operation.operation_id
and row["intent_digest"] == operation.intent_digest
and (row["operation_id"], row["intent_digest"]) == tuple(ready.publication)
and row["scope_levels"] == stored["scope_levels"] == _scope_key(expected)
and row["run_id"] == ready.run_id
and secrets.compare_digest(row["output_sha256"], ready.output_sha256)
and secrets.compare_digest(row["sealed_sha256"], sealed)
and row["attestation"] in _MODES
and type(row["activator"]) is str and bool(row["activator"])
and (row["attestation"] == "owner") is (row["writer"] is None)
and row["writer"] != row["activator"]
)
|
preflight
async
preflight(pool: Any) -> None
Attest before the FLUSH id is spent, so drift refuses instead of UNKNOWN.
Advisory only: record attests again inside the receipt transaction.
Source code in src/symfonic/memory/backends/postgres_staged_proof.py
| async def preflight(pool: Any) -> None:
"""Attest before the FLUSH id is spent, so drift refuses instead of UNKNOWN.
Advisory only: ``record`` attests again inside the receipt transaction.
"""
conn = await pool.acquire()
try:
async with conn.transaction():
await attest(conn)
finally:
await pool.release(conn)
|
record
async
record(conn: Any, armed: StagedProof) -> None
Insert the proof on the receipt transaction's own connection.
Source code in src/symfonic/memory/backends/postgres_staged_proof.py
| async def record(conn: Any, armed: StagedProof) -> None:
"""Insert the proof on the receipt transaction's own connection."""
from symfonic.memory.backends.postgres_journal_staged import STAGED, _scope_key
operation, ready, content = require_armed(armed)
mode, activator, writer = await attest(conn)
expected = _expected(ready)
stored = await conn.fetchrow(
f"SELECT state, scope_levels, encode(sha256(sealed), 'hex') AS sealed_sha256 "
f"FROM {STAGED} WHERE tenant_id=$1 AND invocation_id=$2",
expected["tenant"], expected["invocation"],
)
if (stored is None or stored["state"] != "READY"
or stored["scope_levels"] != _scope_key(expected)
or not secrets.compare_digest(stored["sealed_sha256"], content.sealed_sha256)):
_refuse("the sealed READY row is absent or changed")
inserted = await conn.fetchval(
f"INSERT INTO {PROOFS} (tenant_id, invocation_id, operation_id, intent_digest, "
"scope_levels, run_id, output_sha256, sealed_sha256, activator, attestation, "
"writer) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11) "
"ON CONFLICT DO NOTHING RETURNING 1",
expected["tenant"], expected["invocation"], operation.operation_id,
operation.intent_digest, stored["scope_levels"], ready.run_id,
ready.output_sha256, stored["sealed_sha256"], activator, mode, writer,
)
if inserted != 1:
_refuse("an activation proof already exists for this invocation")
|
require_armed
require_armed(armed: Any) -> StagedProof
Exact types, and the FLUSH is the very publication the READY sealed.
Source code in src/symfonic/memory/backends/postgres_staged_proof.py
| def require_armed(armed: Any) -> StagedProof:
"""Exact types, and the FLUSH is the very publication the READY sealed."""
from symfonic.memory.backends.postgres_staged_content import ContentSeal
if type(armed) is not StagedProof:
_refuse("an exact StagedProof is required")
operation, ready, content = armed
if (type(operation) is not MemoryOperation or type(ready) is not ReadyTurn
or type(content) is not ContentSeal
or operation.verb is not MemoryOperationVerb.FLUSH
or operation.scope != ready.full_scope
or (operation.operation_id, operation.intent_digest) != tuple(ready.publication)):
_refuse("the armed FLUSH is not the sealed READY publication")
return armed
|