root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(postgres): allow fusion_runtime to write legacy-adoption drained marker

Browse Source 
Store-open adoption runs as fusion_runtime, which lacked grants on
public.fusion_schema_migrations, so the drained-marker write failed every
boot. Migration 0032 grants SELECT plus a SECURITY DEFINER helper limited
to the exact marker, and store-open calls that helper instead of raw INSERT.
This commit is contained in:
gsxdsm
2026-07-21 17:27:09 -07:00
parent 396090fc03
commit 4d588ad091
4 changed files with 111 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
packages/core/src/__tests__/legacy-adoption.test.ts
View File

@@ -326,15 +326,18 @@ function makeFakeStore(
db: {
execute: async (q: unknown) => {
const text = sqlText(q);
if (text.includes("SELECT")) {
if (opts?.markerReadThrows) throw new Error("marker read boom");
return markerPresent ? [{ version: "legacy-adoption-drained" }] : [];
}
if (text.includes("INSERT")) {
// FNXC:LegacyAdoption 2026-07-21-17:30: write path calls the SECURITY DEFINER
// helper (SELECT public.fusion_mark_legacy_adoption_drained()); the read path is
// SELECT version FROM … WHERE version = ….
if (text.includes("fusion_mark_legacy_adoption_drained")) {
markerWrites.push(text);
markerPresent = true;
return [];
}
if (text.includes("SELECT") && text.includes("version")) {
if (opts?.markerReadThrows) throw new Error("marker read boom");
return markerPresent ? [{ version: "legacy-adoption-drained" }] : [];
}
return [];
},
},
@@ -404,10 +407,9 @@ describe("adoptLegacyTaskRowsOnOpen — drained-marker completion short-circuit"
expect(await adoptLegacyTaskRowsOnOpen(store)).toBe(0);
// The sweep still ran (marker was absent) …
expect(listCalls.length).toBe(1);
// … and a clean drain recorded the durable marker exactly once, upsert-style.
// … and a clean drain recorded the durable marker exactly once via the SECURITY DEFINER helper.
expect(markerWrites.length).toBe(1);
expect(markerWrites[0]).toContain("INSERT");
expect(markerWrites[0]).toContain("ON CONFLICT");
expect(markerWrites[0]).toContain("fusion_mark_legacy_adoption_drained");
});
it("skips the sweep entirely when the marker is present", async () => {

38
packages/core/src/postgres/migrations/0032_legacy_adoption_drained_marker_runtime_grants.sql Normal file
View File

@@ -0,0 +1,38 @@
/*
FNXC:LegacyAdoption 2026-07-21-17:30:
Store-open adoption (adoptLegacyTaskRowsOnOpen) runs on the project-bound
fusion_runtime connection. public.fusion_schema_migrations only received
superuser grants, so the drained-marker SELECT/INSERT failed with permission
denied on every clean open (TUI spam: "Legacy-adoption drained-marker write
failed" with a bare Drizzle "Failed query" message).
Grant SELECT for the short-circuit read. Do NOT grant unrestricted INSERT —
runtime must not be able to stamp arbitrary numeric migration versions. Instead
expose a SECURITY DEFINER helper that can only write the exact non-numeric
LEGACY_ADOPTION_DRAINED_MARKER row.
*/
DO $$
BEGIN
IF to_regclass('public.fusion_schema_migrations') IS NULL
OR NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'fusion_runtime') THEN
RETURN;
END IF;
GRANT SELECT ON public.fusion_schema_migrations TO fusion_runtime;
CREATE OR REPLACE FUNCTION public.fusion_mark_legacy_adoption_drained()
RETURNS void
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = public
AS $fn$
BEGIN
INSERT INTO public.fusion_schema_migrations (version)
VALUES ('legacy-adoption-drained')
ON CONFLICT (version) DO NOTHING;
END;
$fn$;
REVOKE ALL ON FUNCTION public.fusion_mark_legacy_adoption_drained() FROM PUBLIC;
GRANT EXECUTE ON FUNCTION public.fusion_mark_legacy_adoption_drained() TO fusion_runtime;
END $$;

33
packages/core/src/postgres/schema-applier.ts
View File

@@ -40,8 +40,12 @@ Per-migration identities above stay fixed; only this latest-version marker moves
FNXC:WorkflowTaskContinuations 2026-07-21:
SCHEMA_BASELINE_VERSION advances to 0031 for durable, single-owner task
continuations at workflow column boundaries.
FNXC:LegacyAdoption 2026-07-21-17:30:
SCHEMA_BASELINE_VERSION advances to 0032 for fusion_runtime SELECT +
SECURITY DEFINER write access to the legacy-adoption drained marker.
*/
export const SCHEMA_BASELINE_VERSION = "0031";
export const SCHEMA_BASELINE_VERSION = "0032";
/** FNXC:SymbolLock 2026-07-31-10:00: upgrades need durable task declarations before admission resolves symbols. */
export const TASK_DECLARED_SYMBOLS_VERSION = "0028";
const INITIAL_SCHEMA_VERSION = "0000";
@@ -133,6 +137,11 @@ export const PLANNING_ACTIVE_TIMING_VERSION = "0029";
/** Dashboard health needs project-scoped, read-only runtime access to the SQLite cutover ledger. */
export const SQLITE_MIGRATION_RUNTIME_READ_VERSION = "0030";
export const WORKFLOW_TASK_CONTINUATIONS_VERSION = "0031";
/** FNXC:LegacyAdoption 2026-07-21-17:30: runtime role needs drained-marker read + restricted write. */
export const LEGACY_ADOPTION_DRAINED_MARKER_RUNTIME_GRANTS_VERSION = "0032";
/** SECURITY DEFINER helper that only inserts LEGACY_ADOPTION_DRAINED_MARKER. */
export const LEGACY_ADOPTION_DRAINED_MARKER_FUNCTION = "fusion_mark_legacy_adoption_drained";
/**
* Thrown when the database was migrated by a NEWER Fusion binary than the one now
@@ -326,6 +335,10 @@ const PLANNING_ACTIVE_TIMING_MIGRATION_PATH = join(MIGRATIONS_DIR, "0029_plannin
const TASK_DECLARED_SYMBOLS_MIGRATION_PATH = join(MIGRATIONS_DIR, "0028_task_declared_symbols.sql");
const SQLITE_MIGRATION_RUNTIME_READ_PATH = join(MIGRATIONS_DIR, "0030_sqlite_migration_runtime_read.sql");
const WORKFLOW_TASK_CONTINUATIONS_PATH = join(MIGRATIONS_DIR, "0031_workflow_task_continuations.sql");
const LEGACY_ADOPTION_DRAINED_MARKER_RUNTIME_GRANTS_PATH = join(
MIGRATIONS_DIR,
"0032_legacy_adoption_drained_marker_runtime_grants.sql",
);
/**
* Ensure the migration bookkeeping table exists. Lives in the public schema so
@@ -425,6 +438,9 @@ export async function applySchemaBaseline(
const planningActiveTimingAlreadyApplied = applied.includes(PLANNING_ACTIVE_TIMING_VERSION);
const sqliteMigrationRuntimeReadAlreadyApplied = applied.includes(SQLITE_MIGRATION_RUNTIME_READ_VERSION);
const workflowTaskContinuationsAlreadyApplied = applied.includes(WORKFLOW_TASK_CONTINUATIONS_VERSION);
const legacyAdoptionDrainedMarkerRuntimeGrantsAlreadyApplied = applied.includes(
LEGACY_ADOPTION_DRAINED_MARKER_RUNTIME_GRANTS_VERSION,
);
assertBinaryNotOlderThanDatabase(applied);
let schemaChanged = false;
@@ -881,6 +897,21 @@ export async function applySchemaBaseline(
schemaChanged = true;
}
/*
FNXC:LegacyAdoption 2026-07-21-17:30:
Explicit registration (migrations are never auto-discovered). Existing
clusters need fusion_runtime SELECT + SECURITY DEFINER write for the
drained-marker short-circuit before store-open adoption can stop spamming.
*/
if (!legacyAdoptionDrainedMarkerRuntimeGrantsAlreadyApplied) {
const migrationSql = await readFile(LEGACY_ADOPTION_DRAINED_MARKER_RUNTIME_GRANTS_PATH, "utf8");
await tx.execute(sql.raw(migrationSql));
await tx.execute(
sql`INSERT INTO public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} (version) VALUES (${LEGACY_ADOPTION_DRAINED_MARKER_RUNTIME_GRANTS_VERSION}) ON CONFLICT (version) DO NOTHING`,
);
schemaChanged = true;
}
return { applied: schemaChanged, pluginHooksRun: pluginHooks.length };
});
}

37
packages/core/src/task-store/lifecycle-ops.ts
View File

@@ -9,7 +9,11 @@
import {TaskStore, storeLog, RECONCILE_ORPHAN_TASK_DIR_MAX_AGE_MS, WORKFLOW_COMPILED_STEP_TEMPLATE_PREFIX} from "../store.js";
import {planLegacyAdoption} from "../legacy-adoption.js";
import {sql} from "drizzle-orm";
import {MIGRATION_BOOKKEEPING_TABLE, LEGACY_ADOPTION_DRAINED_MARKER} from "../postgres/schema-applier.js";
import {
MIGRATION_BOOKKEEPING_TABLE,
LEGACY_ADOPTION_DRAINED_MARKER,
LEGACY_ADOPTION_DRAINED_MARKER_FUNCTION,
} from "../postgres/schema-applier.js";
import {mkdir, readdir, readFile, stat, writeFile} from "node:fs/promises";
import {join} from "node:path";
import {existsSync, watch, type Dirent} from "node:fs";
@@ -320,6 +324,23 @@ Safety rules:
marker WRITE is warned and swallowed (the next clean drain retries).
- SQLite (non-backend) mode has no bookkeeping table → no marker, sweep always runs.
*/
/*
FNXC:LegacyAdoption 2026-07-21-17:30:
Drizzle wraps Postgres errors as "Failed query: <SQL> params: …" while the real
SQLSTATE lives on err.cause. Walk a short cause chain so drained-marker failures
surface as permission denied / missing function instead of opaque query text.
*/
function describeStoreOpenDbError(error: unknown): string {
const parts: string[] = [];
let current: unknown = error;
for (let depth = 0; current !== undefined && current !== null && depth < 4; depth += 1) {
const message = current instanceof Error ? current.message : String(current);
parts.push(message.length > 400 ? `${message.slice(0, 200)} … ${message.slice(-120)}` : message);
current = current instanceof Error ? current.cause : undefined;
}
return parts.join(" ⇐ ");
}
async function hasLegacyAdoptionDrainedMarker(store: TaskStore): Promise<boolean> {
const db = store.asyncLayer?.db;
if (!db) return false;
@@ -332,7 +353,7 @@ async function hasLegacyAdoptionDrainedMarker(store: TaskStore): Promise<boolean
// Fail-open toward correctness: an unreadable marker means sweep.
storeLog.warn("Legacy-adoption drained-marker read failed — sweeping anyway", {
phase: "init:legacy-adoption",
error: error instanceof Error ? error.message : String(error),
error: describeStoreOpenDbError(error),
});
return false;
}
@@ -342,14 +363,18 @@ async function writeLegacyAdoptionDrainedMarker(store: TaskStore): Promise<void>
const db = store.asyncLayer?.db;
if (!db) return;
try {
await db.execute(
sql`INSERT INTO public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} (version) VALUES (${LEGACY_ADOPTION_DRAINED_MARKER}) ON CONFLICT (version) DO NOTHING`,
);
/*
FNXC:LegacyAdoption 2026-07-21-17:30:
Call the SECURITY DEFINER helper (migration 0032) instead of a raw INSERT.
fusion_runtime has EXECUTE on the function but not unrestricted INSERT on
fusion_schema_migrations, so it cannot stamp arbitrary migration versions.
*/
await db.execute(sql`SELECT public.${sql.identifier(LEGACY_ADOPTION_DRAINED_MARKER_FUNCTION)}()`);
} catch (error) {
// Non-fatal: the next fully-clean drain writes it again.
storeLog.warn("Legacy-adoption drained-marker write failed", {
phase: "init:legacy-adoption",
error: error instanceof Error ? error.message : String(error),
error: describeStoreOpenDbError(error),
});
}
}