Skip to content

symfonic.memory.backends.postgres_erase_vector_subtree

postgres_erase_vector_subtree

Opt-in contract for ordinary durable graph and vector subtree erasure.

installed async 

installed(conn: Any, jobs: str) -> bool

Attest the opt-in marker and exact irreversible database guards.

Source code in src/symfonic/memory/backends/postgres_erase_vector_subtree.py 
 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
119
120
121
122
async def installed(conn: Any, jobs: str) -> bool:
    """Attest the opt-in marker and exact irreversible database guards."""
    present = await conn.fetchval(
        "SELECT EXISTS (SELECT 1 FROM pg_attribute WHERE attrelid=$1::regclass "
        "AND attname='vector_subtree_marker' AND NOT attisdropped)", jobs,
    )
    if not present:
        return False
    schema = jobs.rsplit(".", 1)[0]
    vector = f'{schema}."memory_vectors"'
    ready = await conn.fetchval(
        "SELECT EXISTS (SELECT 1 FROM pg_attribute WHERE attrelid=$1::regclass "
        "AND attname='vector_forget' AND atttypid='boolean'::regtype AND attnotnull) "
        "AND EXISTS (SELECT 1 FROM pg_attribute WHERE attrelid=$1::regclass "
        "AND attname='vector_subtree_marker' AND atttypid='boolean'::regtype "
        "AND attnotnull) "
        "AND EXISTS (SELECT 1 FROM pg_constraint WHERE conrelid=$1::regclass "
        "AND conname='memory_erase_jobs_vector_subtree_marker' AND convalidated "
        "AND pg_get_constraintdef(oid)='CHECK (vector_subtree_marker)') "
        "AND EXISTS (SELECT 1 FROM pg_trigger t JOIN pg_proc p ON p.oid=t.tgfoid "
        "WHERE t.tgrelid=$1::regclass AND t.tgname='memory_erase_jobs_phase_guard' "
        "AND t.tgenabled='A' AND t.tgtype=27 AND NOT t.tgisinternal "
        "AND t.tgqual IS NULL AND t.tgattr=''::int2vector "
        "AND p.pronamespace=(SELECT relnamespace FROM pg_class WHERE oid=$1::regclass) "
        "AND NOT p.prosecdef AND p.prosrc=$2) "
        "AND EXISTS (SELECT 1 FROM pg_trigger t JOIN pg_proc p ON p.oid=t.tgfoid "
        "WHERE t.tgrelid=$1::regclass AND t.tgname='memory_erase_jobs_vector_phase_guard' "
        "AND t.tgenabled='A' AND t.tgtype=23 AND NOT t.tgisinternal "
        "AND t.tgqual IS NULL AND t.tgattr=''::int2vector "
        "AND NOT p.prosecdef AND p.prosrc=$3) "
        "AND EXISTS (SELECT 1 FROM pg_trigger t JOIN pg_proc p ON p.oid=t.tgfoid "
        "WHERE t.tgrelid=$4::regclass AND t.tgname='memory_vectors_subtree_row' "
        "AND t.tgenabled='A' AND t.tgtype=23 AND NOT t.tgisinternal "
        "AND t.tgqual IS NULL AND t.tgattr=''::int2vector "
        "AND p.pronamespace=(SELECT relnamespace FROM pg_class WHERE oid=$4::regclass) "
        "AND NOT p.prosecdef AND p.prosrc=$5)",
        jobs, PHASE_BODY, VECTOR_BODY, vector, ROW_BODY,
    )
    if not ready:
        raise MemoryContractError(UNREADY)
    return True

preflight async 

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

Refuse ambiguous owners before the durable job can hide any vector.

Source code in src/symfonic/memory/backends/postgres_erase_vector_subtree.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
async def preflight(conn: Any, jobs: str, scope: Any) -> None:
    """Refuse ambiguous owners before the durable job can hide any vector."""
    from symfonic.memory.backends.postgres_erase_ownership import BUDGET, owners_sql
    from symfonic.memory.backends.postgres_native_subtree import _descends

    if not await installed(conn, jobs):
        raise MemoryContractError(UNREADY)
    table = await require_ready(conn, jobs.rsplit(".", 1)[0])
    tenant, prefix = scope.tenant_id, scope.subtree_prefix
    async with index_probes(conn, ordered=True, presorted=True):
        null = await conn.fetchval(
            f"SELECT EXISTS (SELECT 1 FROM {table} WHERE tenant_id=$1 AND scope_path IS NULL)",
            tenant,
        )
        crossed = await conn.fetch(owners_sql(table), prefix, prefix[:-1] + "0")
    if null or len(crossed) > BUDGET or any(
        row["tenant_id"] != tenant or not _descends(row["scope_path"], scope)
        for row in crossed
    ):
        raise MemoryContractError("native vector subtree ownership is ambiguous")

require_empty async 

require_empty(conn: Any, jobs: str, scope: Any) -> None

Independent physical seek while the active job still fences writers.

Source code in src/symfonic/memory/backends/postgres_erase_vector_subtree.py 
154
155
156
157
158
159
160
161
162
async def require_empty(conn: Any, jobs: str, scope: Any) -> None:
    """Independent physical seek while the active job still fences writers."""
    from symfonic.memory.backends.postgres_erase_ownership import range_sql

    table = await require_ready(conn, jobs.rsplit(".", 1)[0])
    prefix = scope.subtree_prefix
    async with index_probes(conn, ordered=True, presorted=True):
        if await conn.fetchval(range_sql(table), prefix, prefix[:-1] + "0"):
            raise MemoryContractError("native vector subtree physical rows remain")