|
| 1 | +import { getTransientDatabaseFailure } from '@sim/utils/errors' |
| 2 | +import { sleep } from '@sim/utils/helpers' |
| 3 | +import { backoffWithJitter } from '@sim/utils/retry' |
| 4 | +import type { Sql, TransactionSql } from 'postgres' |
| 5 | + |
| 6 | +type ProjectMembershipPhase = 'connector' | 'column' |
| 7 | + |
| 8 | +/** Reads durable authority without inferring it from partially copied membership. */ |
| 9 | +export async function readProjectMembershipPhase( |
| 10 | + sql: Sql | TransactionSql |
| 11 | +): Promise<ProjectMembershipPhase> { |
| 12 | + const [table] = |
| 13 | + await sql`SELECT to_regclass('public.project_membership_rollout') IS NOT NULL AS present` |
| 14 | + if (!table?.present) throw new Error('Project membership rollout marker is missing') |
| 15 | + const rows = await sql< |
| 16 | + { id: string; phase: string }[] |
| 17 | + >`SELECT id, phase FROM public.project_membership_rollout` |
| 18 | + const row = rows[0] |
| 19 | + if (rows.length !== 1 || row?.id !== 'membership' || !['connector', 'column'].includes(row.phase)) |
| 20 | + throw new Error('Project membership rollout marker is missing or invalid') |
| 21 | + return row.phase as ProjectMembershipPhase |
| 22 | +} |
| 23 | + |
| 24 | +/** Mutating maintenance never writes columns while compatible applications still use the connector. */ |
| 25 | +export async function assertProjectColumnAuthority(sql: Sql | TransactionSql): Promise<void> { |
| 26 | + if ((await readProjectMembershipPhase(sql)) !== 'column') |
| 27 | + throw new Error( |
| 28 | + 'Project membership authority must switch through db:migrate after verified server/worker drainage before maintenance writes' |
| 29 | + ) |
| 30 | +} |
| 31 | + |
| 32 | +/** Commits the forward-only authority change before any discovery, backfill, or repair. */ |
| 33 | +export async function switchProjectMembershipAuthority(sql: Sql): Promise<void> { |
| 34 | + for (let attempt = 0; ; attempt++) { |
| 35 | + try { |
| 36 | + await sql.begin(async (tx) => { |
| 37 | + await tx`SET LOCAL statement_timeout = '3s'` |
| 38 | + await tx`SELECT set_config(CASE WHEN current_setting('transaction_timeout', true) IS NULL |
| 39 | + THEN 'idle_in_transaction_session_timeout' ELSE 'transaction_timeout' END, '5s', true)` |
| 40 | + await tx`LOCK TABLE public.workspace IN ACCESS EXCLUSIVE MODE NOWAIT` |
| 41 | + const [operator] = await tx`SELECT EXISTS (SELECT 1 FROM pg_locks |
| 42 | + WHERE pid = pg_backend_pid() AND locktype = 'advisory' AND granted |
| 43 | + AND classid = ((hashtextextended('sim:project-backfill-operator',0) >> 32) & 4294967295)::oid |
| 44 | + AND objid = (hashtextextended('sim:project-backfill-operator',0) & 4294967295)::oid |
| 45 | + AND objsubid = 1) AS held` |
| 46 | + if (!operator?.held) |
| 47 | + throw new Error('Project authority switch requires the operator session lock') |
| 48 | + const phase = await readProjectMembershipPhase(tx) |
| 49 | + if (phase === 'column') return |
| 50 | + const [state] = await tx`SELECT |
| 51 | + to_regclass('public.project_workspace') IS NOT NULL AS connector, |
| 52 | + EXISTS (SELECT 1 FROM public.workspace WHERE project_id IS NOT NULL) AS populated` |
| 53 | + if (!state?.connector || state.populated) |
| 54 | + throw new Error( |
| 55 | + 'Connector authority requires an intact connector and no populated Project columns; reconcile the rollout state before switching' |
| 56 | + ) |
| 57 | + await tx`UPDATE public.project_membership_rollout SET phase = 'column' WHERE id = 'membership'` |
| 58 | + }) |
| 59 | + return |
| 60 | + } catch (error) { |
| 61 | + const transient = getTransientDatabaseFailure(error) |
| 62 | + if (attempt >= 2 || (transient !== 'capacity' && transient !== 'conflict')) throw error |
| 63 | + await sleep(backoffWithJitter(attempt, null, { baseMs: 100, maxMs: 500 })) |
| 64 | + } |
| 65 | + } |
| 66 | +} |
| 67 | + |
| 68 | +/** Only an empty final schema push may initialize a missing singleton directly in column mode. */ |
| 69 | +export async function bootstrapProjectColumnAuthority(sql: Sql): Promise<void> { |
| 70 | + await sql.begin(async (tx) => { |
| 71 | + await tx`SET LOCAL statement_timeout = '3s'` |
| 72 | + await tx`LOCK TABLE public.workspace IN ACCESS EXCLUSIVE MODE NOWAIT` |
| 73 | + const [state] = await tx`SELECT |
| 74 | + to_regclass('public.project_workspace') IS NULL AS retired, |
| 75 | + EXISTS (SELECT 1 FROM pg_attribute WHERE attrelid = 'public.workspace'::regclass |
| 76 | + AND attname = 'project_id' AND attnotnull AND NOT attisdropped) AS required, |
| 77 | + NOT EXISTS (SELECT 1 FROM public.workspace) AND NOT EXISTS (SELECT 1 FROM public.project) AS empty` |
| 78 | + if (!state?.retired || !state.required) |
| 79 | + throw new Error('Project authority bootstrap requires the fresh final schema') |
| 80 | + const [table] = |
| 81 | + await tx`SELECT to_regclass('public.project_membership_rollout') IS NOT NULL AS present` |
| 82 | + if (!table?.present) |
| 83 | + throw new Error('Project membership rollout table is missing from the final schema') |
| 84 | + const rows = await tx`SELECT id, phase FROM public.project_membership_rollout` |
| 85 | + if (!rows.length && state.empty) |
| 86 | + await tx`INSERT INTO public.project_membership_rollout (id, phase) VALUES ('membership', 'column')` |
| 87 | + await assertProjectColumnAuthority(tx) |
| 88 | + }) |
| 89 | +} |
0 commit comments