Skip to content

symfonic.memory.backends.postgres_edge_publication_schema

postgres_edge_publication_schema

Explicit dark publication DDL; no backfill, default migration or fleet admission.

Only standalone ordinary tables and the declared publication trigger are supported. Extra triggers/rules/inheritance require a separate composition review; installation refuses them without removing existing database objects. Working endpoint/expiry trigger composition is deliberately unsupported here; it requires a separate integration review before this port can serve that fleet. Persistent catalog drift is detected. Any role with TRIGGER or DDL privileges can defeat between-call attestation via transient install/write/remove; owner disable/write/restore is also outside authority. Both require a fleet barrier.

attest async 

attest(conn, schema: str, *, working_guard=None) -> None

Refuse absent, partially installed, altered or disabled guards.

Source code in src/symfonic/memory/backends/postgres_edge_publication_schema.py 
125
126
127
128
129
async def attest(conn, schema: str, *, working_guard=None) -> None:
    """Refuse absent, partially installed, altered or disabled guards."""
    async with conn.transaction():
        await conn.execute("SET LOCAL search_path TO pg_catalog, pg_temp")
        await _attest(conn, schema, working_guard=working_guard)

install async 

install(conn, schema: str, *, working_guard=None) -> None

Opt-in additive DDL only; never invoked by backend ensure_schema.

Source code in src/symfonic/memory/backends/postgres_edge_publication_schema.py 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
async def install(conn, schema: str, *, working_guard=None) -> None:
    """Opt-in additive DDL only; never invoked by backend ensure_schema."""
    namespace = quoted(schema)
    table = f"{namespace}.memory_edges"
    async with conn.transaction():
        await conn.execute("SET LOCAL search_path TO pg_catalog, pg_temp")
        await _require_tables(conn, schema)
        if working_guard is not None:
            from symfonic.memory.backends.postgres_edge_publication_composed import (
                attest_working_profile,
            )
            await conn.execute("SET LOCAL lock_timeout = '5s'")
            await conn.execute(
                "SELECT pg_catalog.pg_advisory_xact_lock("
                "pg_catalog.hashtextextended('symfonic-working-guard-install', 0))"
            )
            await conn.execute(
                f"LOCK TABLE {namespace}.memory_nodes, {table} "
                "IN SHARE ROW EXCLUSIVE MODE"
            )
            installed = await conn.fetchval(
                "SELECT EXISTS(SELECT 1 FROM pg_catalog.pg_trigger "
                "WHERE tgrelid=$1::regclass AND tgname=$2)", table, _FUNCTION,
            )
            if installed:
                await _attest(conn, schema, working_guard=working_guard)
                return
            await attest_working_profile(conn, schema, working_guard, publication=False)
        await conn.execute(f"LOCK TABLE {table} IN SHARE ROW EXCLUSIVE MODE")
        await conn.execute(f"ALTER TABLE {table} " + ", ".join(
            f"ADD COLUMN IF NOT EXISTS {column} {kind} NULL"
            for column, kind in (("publication_format", "SMALLINT"),
                                 ("publication_incarnation", "UUID"),
                                 ("publication_revision", "BIGINT"))))
        exists = await conn.fetchval(
            "SELECT EXISTS(SELECT 1 FROM pg_catalog.pg_constraint "
            "WHERE conrelid=$1::regclass AND conname=$2)", table, _CHECK)
        if not exists:
            await conn.execute(f"ALTER TABLE {table} ADD CONSTRAINT {_CHECK} {_EXPRESSION}")
        exists = await conn.fetchval(
            "SELECT pg_catalog.to_regprocedure($1)", f"{namespace}.{_FUNCTION}()")
        if exists is None:
            await conn.execute(f"CREATE FUNCTION {namespace}.{_FUNCTION}() RETURNS trigger "
                               "LANGUAGE plpgsql SET search_path = pg_catalog, pg_temp "
                               f"AS $body${_BODY}$body$")
        exists = await conn.fetchval(
            "SELECT EXISTS(SELECT 1 FROM pg_catalog.pg_trigger "
            "WHERE tgrelid=$1::regclass AND tgname=$2)", table, _FUNCTION)
        if not exists:
            await conn.execute(f"CREATE TRIGGER {_FUNCTION} BEFORE INSERT OR UPDATE ON {table} "
                               f"FOR EACH ROW EXECUTE FUNCTION {namespace}.{_FUNCTION}()")
            await conn.execute(f"ALTER TABLE {table} ENABLE ALWAYS TRIGGER {_FUNCTION}")
        await attest(conn, schema, working_guard=working_guard)