Skip to content

symfonic.memory.backends.postgres_vector_retirement

postgres_vector_retirement

Current-code vector admission; permanent jobs apply to every owner version.

Vector retirement additionally requires the durable phase and ownership proof. Drain all older readers/writers/erasers before installing CORE-426.

authority async 

authority(conn: Any, schema: str | None = None)

Retirement-enabled schemas require the opt-in vector trigger contract.

Source code in src/symfonic/memory/backends/postgres_vector_retirement.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
async def authority(conn: Any, schema: str | None = None):
    """Retirement-enabled schemas require the opt-in vector trigger contract."""
    jobs = await jobs_table(conn, schema)
    if jobs is None or not getattr(jobs, "retirement", False):
        return None
    vector = (
        await conn.fetchval("SELECT format('%I.memory_vectors', $1::text)", schema)
        if schema
        else "memory_vectors"
    )
    nodes = (
        await conn.fetchval("SELECT format('%I.memory_nodes', $1::text)", schema)
        if schema
        else "memory_nodes"
    )
    if not await conn.fetchval(
        "SELECT v.relnamespace=n.relnamespace AND v.relkind='r' "
        "FROM pg_class v,pg_class n WHERE v.oid=to_regclass($1) "
        "AND n.oid=to_regclass($2)",
        vector,
        nodes,
    ):
        raise MemoryContractError("vector retirement requires the memory_nodes schema")
    if not await conn.fetchval(
        "SELECT count(*)=2 FROM pg_trigger t JOIN pg_proc p ON p.oid=t.tgfoid "
        "JOIN pg_class v ON v.oid=t.tgrelid WHERE v.oid=to_regclass($3) "
        "AND t.tgenabled='A' AND NOT t.tgisinternal AND t.tgqual IS NULL "
        "AND t.tgattr=''::int2vector AND p.pronamespace=v.relnamespace "
        "AND p.proowner=v.relowner AND NOT p.prosecdef AND p.provolatile='v' "
        "AND p.prolang=(SELECT oid FROM pg_language WHERE lanname='plpgsql') "
        "AND p.prorettype='trigger'::regtype AND p.pronargs=0 "
        "AND p.proconfig IS NULL AND v.relpersistence='p' AND NOT v.relrowsecurity "
        "AND ((t.tgname='memory_vectors_retirement_lock' AND t.tgtype=22 AND p.prosrc=$1) "
        "OR (t.tgname='memory_vectors_retirement_row' AND t.tgtype=23 AND p.prosrc=$2))",
        LOCK_BODY,
        ROW_BODY,
        vector,
    ):
        raise MemoryContractError("vector retirement trigger contract drift")
    return jobs

reader async 

reader(conn: Any) -> str

Return a statement-local predicate, including for legacy owner paths.

Source code in src/symfonic/memory/backends/postgres_vector_retirement.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
async def reader(conn: Any) -> str:
    """Return a statement-local predicate, including for legacy owner paths."""
    jobs = await authority(conn)
    if jobs is None:
        return "TRUE"
    if await conn.fetchval("SELECT current_setting('transaction_isolation')") != "read committed":
        raise MemoryContractError("vector retirement reads require READ COMMITTED isolation")
    return (
        f"NOT EXISTS (SELECT 1 FROM {jobs} j WHERE j.tenant_id=$1 AND j.permanent) "
        f"AND NOT EXISTS (SELECT 1 FROM {jobs} j WHERE j.tenant_id=$1 AND j.state='erasing' "
        "AND COALESCE((to_jsonb(j)->>'vector_forget')::boolean, false) "
        'AND (memory_vectors.scope_path COLLATE "C">=j.prefix COLLATE "C" '
        'AND memory_vectors.scope_path COLLATE "C"'
        '<(left(j.prefix,length(j.prefix)-1)||\'0\') COLLATE "C"))'
    )

writer async 

writer(conn: Any, tenant: str, *, native: bool = True) -> None

Native writes always lock; v0 needs a lock only with attested retirement.

Source code in src/symfonic/memory/backends/postgres_vector_retirement.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
async def writer(conn: Any, tenant: str, *, native: bool = True) -> None:
    """Native writes always lock; v0 needs a lock only with attested retirement."""
    if native:
        await share_lock_nodes(conn)
    jobs = await authority(conn)
    if jobs is None:
        return
    if not native:
        await share_lock_nodes(conn)
    if await conn.fetchval("SELECT current_setting('transaction_isolation')") != "read committed":
        raise MemoryContractError("vector retirement fence requires READ COMMITTED isolation")
    if await conn.fetchval(
        f"SELECT EXISTS (SELECT 1 FROM {jobs} WHERE tenant_id=$1 AND permanent)",
        tenant,
    ):
        raise MemoryContractError("native tenant permanently retired")