Skip to content

symfonic.memory.backends.postgres_journal_base_retirement

postgres_journal_base_retirement

Bounded erasure seam for operator-migrated PostgreSQL invocation journal base.

Not wired into graph retirement: staged/proof rows and all unverified states remain refused. The caller must hold the graph retirement transaction/lock.

bounded_sql 

bounded_sql(table: str) -> str

Owner-indexed C-range probe for bounded retirement work.

Source code in src/symfonic/memory/backends/postgres_journal_base_retirement.py 
86
87
88
89
90
def bounded_sql(table: str) -> str:
    """Owner-indexed C-range probe for bounded retirement work."""
    return (f'SELECT ctid FROM {table} WHERE erase_scope COLLATE "C">=$1 '
            'AND erase_scope COLLATE "C"<$2 ORDER BY erase_scope COLLATE "C" '
            f"LIMIT {BUDGET + 1}")

erase async 

erase(conn: Any, schema: str, scope: Any) -> None

Called inside initial permanent transaction after graph EXCLUSIVE lock.

Source code in src/symfonic/memory/backends/postgres_journal_base_retirement.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
async def erase(conn: Any, schema: str, scope: Any) -> None:
    """Called inside initial permanent transaction after graph EXCLUSIVE lock."""
    tables = await require_ready(conn, schema)
    await _selected(conn, tables, scope)
    async with index_probes(conn, ordered=True, presorted=True):
        for table in tables:
            count = await conn.fetchval(
                "WITH selected AS MATERIALIZED (" + bounded_sql(table) + "), "
                f"erased AS (DELETE FROM {table} t USING selected s WHERE t.ctid=s.ctid "
                "RETURNING 1) SELECT count(*) FROM erased",
                scope.subtree_prefix, scope.subtree_prefix[:-1] + "0",
            )
            if count > BUDGET:
                raise MemoryContractError(EXCEEDED)

preflight async 

preflight(conn: Any, schema: str, scope: Any) -> None

Call before permanent intent; never silently truncate large histories.

Source code in src/symfonic/memory/backends/postgres_journal_base_retirement.py 
102
103
104
async def preflight(conn: Any, schema: str, scope: Any) -> None:
    """Call before permanent intent; never silently truncate large histories."""
    await _selected(conn, await require_ready(conn, schema), scope)

require_ready async 

require_ready(conn: Any, schema: str) -> list[str]

Attest exact installed writer fences before any retirement claim.

Source code in src/symfonic/memory/backends/postgres_journal_base_retirement.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
async def require_ready(conn: Any, schema: str) -> list[str]:
    """Attest exact installed writer fences before any retirement claim."""
    from symfonic.memory.backends.postgres_erase_ownership import _INDEX_CHECK

    if await migration.present(conn, schema, migration.STAGED):
        from symfonic.memory.backends import postgres_staged_auxiliary_schema as staged

        await staged.require_ready(conn, schema)
    names = await migration.present(conn, schema)
    if not names:
        return []
    if set(names) != set(migration.TABLES):
        raise MemoryContractError(UNREADY)
    s = migration.quote(schema)
    tables = [f"{s}.{migration.quote(name)}" for name in ORDER]
    if await conn.fetchval(
        _FUNCTIONS, schema, f"{s}.memory_nodes", migration.OWNER, migration.OWNER_BODY,
        migration.FENCE, migration.fence_body(schema),
    ) is not True:
        raise MemoryContractError(UNREADY)
    for table in tables:
        if await conn.fetchval(_TRIGGER, table, migration.FENCE) is not True:
            raise MemoryContractError(UNREADY)
    await conn.execute("LOCK TABLE " + ", ".join(tables) + " IN SHARE UPDATE EXCLUSIVE MODE")
    wanted = [(name, f"idx_core460_{migration.TABLES[name]}_owner_c", ["erase_scope"])
              for name in ORDER]
    rows = await conn.fetch(_INDEX_CHECK, s, json.dumps(wanted))
    if len(rows) != len(wanted) or any(row["valid"] is not True for row in rows):
        raise MemoryContractError(UNREADY)
    for table in tables:
        if await conn.fetchval(_TRIGGER, table, migration.FENCE) is not True:
            raise MemoryContractError(UNREADY)
    return tables