Skip to content

symfonic.memory.backends.postgres_journal_base_schema

postgres_journal_base_schema

Operator-installed owner keys and writer fence for v1 invocation journal base.

Installation requires drained writers/erasers and an external DDL barrier. This module does not admit production QUICK or wire journal erasure into the graph.

install async 

install(conn: Any, schema: str) -> None

One-shot transactional migration; scans only under operator DDL barrier.

Source code in src/symfonic/memory/backends/postgres_journal_base_schema.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
async def install(conn: Any, schema: str) -> None:
    """One-shot transactional migration; scans only under operator DDL barrier."""
    s = quote(schema)
    async with conn.transaction():
        await conn.execute(f"LOCK TABLE {s}.memory_nodes IN EXCLUSIVE MODE")
        if await conn.fetchval(
            f"SELECT EXISTS (SELECT 1 FROM {s}.memory_erase_jobs WHERE permanent)"
        ):
            raise ValueError("CORE-460: drain permanent jobs before journal migration")
        if set(await present(conn, schema)) != set(TABLES):
            raise ValueError("CORE-460: install all five journal base tables first")
        await conn.execute(
            f"CREATE FUNCTION {s}.{OWNER}(levels jsonb, tenant text) RETURNS text "
            "LANGUAGE plpgsql IMMUTABLE SET search_path=pg_catalog AS $owner$"
            + OWNER_BODY + "$owner$"
        )
        await conn.execute(
            f"CREATE FUNCTION {s}.{FENCE}() RETURNS trigger LANGUAGE plpgsql "
            "SET search_path=pg_catalog AS $fence$" + fence_body(schema) + "$fence$"
        )
        for name in TABLES:
            table = f"{s}.{quote(name)}"
            await conn.execute(f"LOCK TABLE {table} IN SHARE ROW EXCLUSIVE MODE")
            await conn.execute(f"ALTER TABLE {table} ADD COLUMN erase_scope text COLLATE \"C\"")
        journal = f"{s}.{codec.JOURNAL}"
        if await conn.fetchval(
            f"SELECT EXISTS (SELECT 1 FROM {journal} WHERE schema<>$1)", codec.SCHEMA
        ):
            raise ValueError("CORE-460: non-native journal row")
        mismatch = await conn.fetchval(
            f"SELECT EXISTS (SELECT 1 FROM {s}.{completion.COMPLETIONS} c JOIN {journal} j "
            "USING (tenant_id, invocation_id) WHERE c.scope_levels::jsonb IS DISTINCT "
            "FROM j.scope_levels)"
        )
        if mismatch:
            raise ValueError("CORE-460: completion owner differs from journal")
        for name in TABLES:
            table = f"{s}.{quote(name)}"
            if name == codec.JOURNAL:
                levels = "t.scope_levels"
            elif name == completion.SEQUENCES:
                levels = "t.scope_levels::jsonb"
            else:
                levels = (f"(SELECT j.scope_levels FROM {journal} j WHERE j.tenant_id=t.tenant_id "
                          "AND j.invocation_id=t.invocation_id)")
            await conn.execute(
                f"UPDATE {table} t SET erase_scope={s}.{OWNER}({levels}, t.tenant_id)"
            )
            await conn.execute(f"ALTER TABLE {table} ALTER COLUMN erase_scope SET NOT NULL")
            await conn.execute(f"CREATE INDEX idx_core460_{TABLES[name]}_owner_c ON {table} "
                               '(erase_scope COLLATE "C")')
            await conn.execute(f"CREATE TRIGGER {FENCE} BEFORE INSERT OR UPDATE ON {table} "
                               f"FOR EACH ROW EXECUTE FUNCTION {s}.{FENCE}()")
            await conn.execute(f"ALTER TABLE {table} ENABLE ALWAYS TRIGGER {FENCE}")