From 923baeedcd881ae8741d84b08ba65818ffcf4f0b Mon Sep 17 00:00:00 2001 From: Sheng Kun Chang Date: Sat, 13 Jun 2026 23:54:46 +0800 Subject: [PATCH] fix: pin dev app-schema migrations to their schema and self-heal drift Dev per-app schema migrations could leak DDL into public when the context-based search_path was unset, and the (scope, version) tracking PK collided when two modules shared one app schema (a second module's migrations were skipped as already-applied). This hardens the SDK dev migration path: - Tracking table PK becomes (module_id, scope, version); a legacy table is upgraded in place (ADD COLUMN + DROP/ADD PK, idempotent). Two modules can now share one app schema without colliding, and each app's history stays independent. - pinnedRunTx wraps the lifecycle TxRunner with SET LOCAL search_path to the target schema for the duration of each migration transaction. This is defense-in-depth on top of db.WithSchema: even an empty migration context lands the DDL in the app schema instead of public (the exact failure that leaked module tables into public and broke platform login). - Drift self-heal: a version recorded as applied but whose tables are absent from the schema (the prior public leak) is forgotten and re-applied into the correct schema; module migrations are CREATE TABLE IF NOT EXISTS so re-apply is clean. Drift is detected by the module's table-name prefix, so it generalizes to any module/app schema. Adds dev_migrate_drift_test.go covering two modules sharing one app schema (no public leak) and the drop-table -> re-provision self-heal path. Design: docs-temp/dev-app-isolation/design.md Co-Authored-By: Claude Opus 4.8 (1M context) --- internal/core/dev_migrate.go | 173 +++++++++++++++++++--- internal/core/dev_migrate_drift_test.go | 186 ++++++++++++++++++++++++ 2 files changed, 336 insertions(+), 23 deletions(-) create mode 100644 internal/core/dev_migrate_drift_test.go diff --git a/internal/core/dev_migrate.go b/internal/core/dev_migrate.go index ada4d11..aaf7a07 100644 --- a/internal/core/dev_migrate.go +++ b/internal/core/dev_migrate.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "os" + "strings" "sync" "github.com/jackc/pgx/v5" @@ -100,9 +101,11 @@ func (m *Module) provisionDevAppSchema(ctx context.Context, schema string) error // ensureSchemaMigrated creates the target schema and its per-schema migration // tracking table, then applies any not-yet-recorded migrations for the given // scope. Each schema owns its own schema_migrations table (keyed by -// scope+version) so every app's app-scope history is independent and a second -// app installing the same module re-runs the migrations into its own schema -// instead of being skipped as "already applied". +// module_id+scope+version) so every app's app-scope history is independent +// AND two modules sharing one app schema do not collide on the (scope,version) +// key — a second app, or a second module in the same app, re-runs its own +// migrations into the target schema instead of being skipped as "already +// applied". func (m *Module) ensureSchemaMigrated(ctx context.Context, scope migration.Scope, schema string) error { pool, release, err := m.resolvePool(ctx) if err != nil { @@ -116,23 +119,75 @@ func (m *Module) ensureSchemaMigrated(ctx context.Context, scope migration.Scope } trackingTable := pgx.Identifier{schema, "schema_migrations"}.Sanitize() + if err := ensureTrackingTable(ctx, pool, trackingTable); err != nil { + return fmt.Errorf("dev migrate: tracking table in %s: %w", schema, err) + } + + return m.applyDevScope(ctx, pool, trackingTable, schema, scope) +} + +// ensureTrackingTable creates the per-schema schema_migrations table and, for +// schemas created by an older SDK, upgrades the legacy (scope, version) layout +// to the module-scoped (module_id, scope, version) layout. +// +// Legacy rows are backfilled with module_id set to the empty string (the +// "unknown owner" marker): no live module ever records under the empty id, so +// on the next dev run every module's applied-set comes up empty for those +// versions, drift detection finds its tables missing, and the migrations +// re-apply into the schema (cleanly, because module migrations use CREATE +// TABLE IF NOT EXISTS). This is how the recorded-but-never-applied schemas +// self-heal — see applyDevScope. +func ensureTrackingTable(ctx context.Context, pool *pgxpool.Pool, trackingTable string) error { if _, err := pool.Exec(ctx, fmt.Sprintf(` CREATE TABLE IF NOT EXISTS %s ( + module_id text NOT NULL DEFAULT '', scope text NOT NULL, version text NOT NULL, applied_at timestamptz NOT NULL DEFAULT now(), - PRIMARY KEY (scope, version) + PRIMARY KEY (module_id, scope, version) )`, trackingTable)); err != nil { - return fmt.Errorf("dev migrate: create tracking table in %s: %w", schema, err) + return fmt.Errorf("create: %w", err) } - - return m.applyDevScope(ctx, pool, trackingTable, schema, scope) + // Upgrade a legacy table (PK was (scope, version), no module_id column). + if _, err := pool.Exec(ctx, fmt.Sprintf( + `ALTER TABLE %s ADD COLUMN IF NOT EXISTS module_id text NOT NULL DEFAULT ''`, + trackingTable)); err != nil { + return fmt.Errorf("add module_id column: %w", err) + } + // Rebuild the PK to include module_id when the table predates it. The + // constraint name is schema-local and stable, so DROP-then-ADD is safe and + // idempotent across runs. + if _, err := pool.Exec(ctx, fmt.Sprintf( + `DO $$ + DECLARE pk text; + BEGIN + SELECT conname INTO pk FROM pg_constraint + WHERE conrelid = %s::regclass AND contype = 'p'; + IF pk IS NOT NULL AND NOT EXISTS ( + SELECT 1 FROM pg_attribute a + JOIN pg_constraint c ON c.conrelid = a.attrelid AND a.attnum = ANY (c.conkey) + WHERE c.conname = pk AND a.attname = 'module_id' + ) THEN + EXECUTE 'ALTER TABLE %s DROP CONSTRAINT ' || quote_ident(pk); + EXECUTE 'ALTER TABLE %s ADD PRIMARY KEY (module_id, scope, version)'; + END IF; + END $$`, + quoteLiteral(trackingTable), trackingTable, trackingTable)); err != nil { + return fmt.Errorf("rebuild primary key: %w", err) + } + return nil } // applyDevScope filters the embedded sql// migrations down to those not -// already recorded in the schema's tracking table, then applies the remainder -// via the same TxRunner the production install handler uses, pinned to the -// target schema. +// already recorded in the schema's tracking table FOR THIS MODULE, then applies +// the remainder via the same TxRunner the production install handler uses, +// pinned to the target schema. +// +// Self-heal: a version recorded as applied but whose effect is missing from the +// schema (no table carrying this module's id prefix exists) is drift — the +// historical search_path bug recorded the row while the DDL leaked into public. +// Such recordings are discarded so the migrations re-apply into the right +// schema on this run. func (m *Module) applyDevScope(ctx context.Context, pool *pgxpool.Pool, trackingTable, schema string, scope migration.Scope) error { all, err := migration.List(m.config.SQL, scope) if err != nil { @@ -142,11 +197,29 @@ func (m *Module) applyDevScope(ctx context.Context, pool *pgxpool.Pool, tracking return nil } - applied, err := loadAppliedVersions(ctx, pool, trackingTable, scope) + applied, err := m.loadAppliedVersions(ctx, pool, trackingTable, scope) if err != nil { return fmt.Errorf("dev migrate %s: load applied: %w", scope, err) } + // Drift detection: if anything is recorded for this module but none of its + // tables exist in the schema, the recorded DDL never landed here (the public + // leak). Forget the recordings so every migration re-applies; clear the stale + // rows so the tracking table reflects reality after the re-apply. + if len(applied) > 0 { + present, err := m.moduleHasTables(ctx, pool, schema) + if err != nil { + return fmt.Errorf("dev migrate %s: drift check: %w", scope, err) + } + if !present { + m.logger.Printf("dev: %s schema %s has tracking rows for module %s but no tables — re-applying (search_path drift self-heal)", scope, schema, m.config.ID) + if err := m.clearAppliedVersions(ctx, pool, trackingTable, scope); err != nil { + return fmt.Errorf("dev migrate %s: clear drift: %w", scope, err) + } + applied = map[string]bool{} + } + } + pending := make([]migration.Migration, 0, len(all)) for _, mig := range all { if !applied[mig.Version] { @@ -157,15 +230,17 @@ func (m *Module) applyDevScope(ctx context.Context, pool *pgxpool.Pool, tracking return nil } - // Pin the whole sequence to the target schema so every migration lands in - // the same place. App scope runs via Module.Tx, which reads this schema from - // the context; module scope runs via Module.ModuleTx, which overlays - // mod_ itself (identical to schema here) — setting it is harmless. + // Pin the whole sequence to the target schema two ways: db.WithSchema sets it + // in context (the path Module.Tx/ModuleTx read), and pinnedRunTx forces a + // SET LOCAL search_path inside each migration's transaction. The explicit pin + // is defense-in-depth: it guarantees the DDL lands in the app schema even if + // the context-based path ever regresses — which is the exact failure that + // once leaked module tables into public. migrateCtx := db.WithSchema(ctx, schema) - runTx := m.lifecycleTxRunner(scope) + runTx := pinnedRunTx(m.lifecycleTxRunner(scope), schema) ran, err := migration.Apply(migrateCtx, runTx, m.config.SQL, pending) // Record whatever ran before the failure (if any) so retries skip them. - if recErr := recordAppliedVersions(ctx, pool, trackingTable, scope, ran); recErr != nil && err == nil { + if recErr := m.recordAppliedVersions(ctx, pool, trackingTable, scope, ran); recErr != nil && err == nil { return fmt.Errorf("dev migrate %s: record applied: %w", scope, recErr) } if err != nil { @@ -175,10 +250,43 @@ func (m *Module) applyDevScope(ctx context.Context, pool *pgxpool.Pool, tracking return nil } -func loadAppliedVersions(ctx context.Context, pool *pgxpool.Pool, trackingTable string, scope migration.Scope) (map[string]bool, error) { +// pinnedRunTx wraps a TxRunner so the wrapped fn runs with search_path pinned +// to schema for the duration of the transaction. SET LOCAL is cleared on +// COMMIT/ROLLBACK, so it cannot leak to a later use of the pooled connection. +// The pin runs inside the same transaction as the DDL, so even an empty +// migration context (no db.WithSchema) still lands the DDL in the app schema. +func pinnedRunTx(inner migration.TxRunner, schema string) migration.TxRunner { + pin := "SET LOCAL search_path TO " + pgx.Identifier{schema}.Sanitize() + return func(ctx context.Context, fn func(q db.Querier) error) error { + return inner(ctx, func(q db.Querier) error { + if _, err := q.Exec(ctx, pin); err != nil { + return fmt.Errorf("pin search_path to %s: %w", schema, err) + } + return fn(q) + }) + } +} + +// moduleHasTables reports whether the schema contains at least one ordinary +// table whose name starts with this module's id prefix (m_...). Used by the +// drift self-heal: a module that recorded migrations but owns no tables in the +// schema never actually ran its DDL there. +func (m *Module) moduleHasTables(ctx context.Context, pool *pgxpool.Pool, schema string) (bool, error) { + var exists bool + err := pool.QueryRow(ctx, ` + SELECT EXISTS ( + SELECT 1 FROM information_schema.tables + WHERE table_schema = $1 + AND table_type = 'BASE TABLE' + AND table_name LIKE $2 + )`, schema, m.config.ID+"\\_%").Scan(&exists) + return exists, err +} + +func (m *Module) loadAppliedVersions(ctx context.Context, pool *pgxpool.Pool, trackingTable string, scope migration.Scope) (map[string]bool, error) { rows, err := pool.Query(ctx, - fmt.Sprintf(`SELECT version FROM %s WHERE scope = $1`, trackingTable), - string(scope)) + fmt.Sprintf(`SELECT version FROM %s WHERE module_id = $1 AND scope = $2`, trackingTable), + m.config.ID, string(scope)) if err != nil { return nil, err } @@ -194,14 +302,33 @@ func loadAppliedVersions(ctx context.Context, pool *pgxpool.Pool, trackingTable return out, rows.Err() } -func recordAppliedVersions(ctx context.Context, pool *pgxpool.Pool, trackingTable string, scope migration.Scope, versions []string) error { +func (m *Module) recordAppliedVersions(ctx context.Context, pool *pgxpool.Pool, trackingTable string, scope migration.Scope, versions []string) error { for _, v := range versions { if _, err := pool.Exec(ctx, - fmt.Sprintf(`INSERT INTO %s (scope, version) VALUES ($1, $2) + fmt.Sprintf(`INSERT INTO %s (module_id, scope, version) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING`, trackingTable), - string(scope), v); err != nil { + m.config.ID, string(scope), v); err != nil { return err } } return nil } + +// clearAppliedVersions removes this module's recorded rows for the scope. Used +// by the drift self-heal before a full re-apply so the tracking table is +// rewritten to match what actually runs this time. Legacy rows under the empty +// "unknown owner" id for the same versions are also cleared so the re-applied +// rows are not shadowed by a duplicate version under a different module_id. +func (m *Module) clearAppliedVersions(ctx context.Context, pool *pgxpool.Pool, trackingTable string, scope migration.Scope) error { + _, err := pool.Exec(ctx, + fmt.Sprintf(`DELETE FROM %s WHERE scope = $1 AND module_id IN ($2, '')`, trackingTable), + string(scope), m.config.ID) + return err +} + +// quoteLiteral wraps s as a single-quoted SQL string literal (doubling embedded +// quotes) for use where a value — not an identifier — is interpolated into DDL, +// such as the regclass cast inside the tracking-table PK upgrade. +func quoteLiteral(s string) string { + return "'" + strings.ReplaceAll(s, "'", "''") + "'" +} diff --git a/internal/core/dev_migrate_drift_test.go b/internal/core/dev_migrate_drift_test.go new file mode 100644 index 0000000..04cc9b7 --- /dev/null +++ b/internal/core/dev_migrate_drift_test.go @@ -0,0 +1,186 @@ +package core + +import ( + "context" + "fmt" + "os" + "testing" + "testing/fstest" + + "github.com/jackc/pgx/v5" +) + +// driftTestDBURL returns the dev module-Postgres URL these integration tests +// connect to. MS_LOCAL_DB_URL wins (the var `mirrorstack dev` exports); the +// fallback matches the live ms-app-modules compose Postgres reachable over the +// OrbStack container network. Tests skip cleanly when neither is reachable. +func driftTestDBURL() string { + if u := os.Getenv("MS_LOCAL_DB_URL"); u != "" { + return u + } + return "postgres://mirrorstack:mirrorstack@192.168.107.3:5432/ms_app_modules?sslmode=disable" +} + +func qualifiedTables(t *testing.T, pool *pgx.Conn, schema, prefix string) int { + t.Helper() + var n int + if err := pool.QueryRow(context.Background(), + `SELECT count(*) FROM information_schema.tables + WHERE table_schema=$1 AND table_type='BASE TABLE' AND table_name LIKE $2`, + schema, prefix+"\\_%").Scan(&n); err != nil { + t.Fatalf("count tables %s.%s_*: %v", schema, prefix, err) + } + return n +} + +// TestDevAppSchema_TwoModulesShareSchema_Integration is the regression guard for +// the cross-module tracking collision: two modules provisioning the SAME app +// schema must each get their own tables. Before the per-module tracking key, +// the second module saw the first module's (scope,version) rows, decided its +// migrations were "already applied", and skipped them — leaving its tables +// missing while a tracking row claimed success. +func TestDevAppSchema_TwoModulesShareSchema_Integration(t *testing.T) { + if testing.Short() { + t.Skip("skipping integration test in short mode") + } + resetDefault(t) + t.Setenv(devMigrateEnvVar, driftTestDBURL()) + + coreSQL := fstest.MapFS{ + "sql/app/0001_init.up.sql": &fstest.MapFile{ + Data: []byte(`CREATE TABLE IF NOT EXISTS mcore_items (id text PRIMARY KEY)`), + }, + } + googSQL := fstest.MapFS{ + "sql/app/0001_init.up.sql": &fstest.MapFile{ + // Same version "0001" as mcore on purpose — proves the keys don't collide. + Data: []byte(`CREATE TABLE IF NOT EXISTS mgoog_nonces (nonce text PRIMARY KEY)`), + }, + } + + mcore, err := New(Config{ID: "mcore", Name: "Core", SQL: coreSQL}) + if err != nil { + t.Fatalf("New mcore: %v", err) + } + mgoog, err := New(Config{ID: "mgoog", Name: "Goog", SQL: googSQL}) + if err != nil { + t.Fatalf("New mgoog: %v", err) + } + + ctx := context.Background() + pool, release, err := mcore.resolvePool(ctx) + if err != nil { + t.Skipf("skipping (no dev postgres): %v", err) + } + defer release() + if err := pool.Ping(ctx); err != nil { + t.Skipf("skipping (no dev postgres): %v", err) + } + + schema, _ := devAppSchemaName("aaaaaaaa-0000-0000-0000-000000000001") + ident := func(s string) string { return pgx.Identifier{s}.Sanitize() } + cleanup := func() { + pool.Exec(ctx, `DROP SCHEMA IF EXISTS `+ident(schema)+` CASCADE`) + pool.Exec(ctx, `DROP TABLE IF EXISTS public.mcore_items`) + pool.Exec(ctx, `DROP TABLE IF EXISTS public.mgoog_nonces`) + } + cleanup() + t.Cleanup(cleanup) + + if err := mcore.ensureDevAppSchema(ctx, schema); err != nil { + t.Fatalf("provision mcore: %v", err) + } + if err := mgoog.ensureDevAppSchema(ctx, schema); err != nil { + t.Fatalf("provision mgoog: %v", err) + } + + conn, err := pool.Acquire(ctx) + if err != nil { + t.Fatalf("acquire: %v", err) + } + defer conn.Release() + raw := conn.Conn() + + if got := qualifiedTables(t, raw, schema, "mcore"); got != 1 { + t.Errorf("mcore tables in %s = %d, want 1", schema, got) + } + if got := qualifiedTables(t, raw, schema, "mgoog"); got != 1 { + t.Errorf("mgoog tables in %s = %d, want 1 — cross-module tracking collision", schema, got) + } + // Neither module's DDL leaked into public. + if got := qualifiedTables(t, raw, "public", "mgoog"); got != 0 { + t.Errorf("mgoog tables leaked into public = %d, want 0 — search_path not pinned", got) + } +} + +// TestDevAppSchema_DriftSelfHeal_Integration is the regression guard for the +// search_path drift bug: a migration recorded as applied but whose table is +// missing from the app schema (it leaked into public) must be re-applied on the +// next dev run. +func TestDevAppSchema_DriftSelfHeal_Integration(t *testing.T) { + if testing.Short() { + t.Skip("skipping integration test in short mode") + } + resetDefault(t) + t.Setenv(devMigrateEnvVar, driftTestDBURL()) + + sqlFS := fstest.MapFS{ + "sql/app/0001_init.up.sql": &fstest.MapFile{ + Data: []byte(`CREATE TABLE IF NOT EXISTS mheal_items (id text PRIMARY KEY)`), + }, + } + m, err := New(Config{ID: "mheal", Name: "Heal", SQL: sqlFS}) + if err != nil { + t.Fatalf("New: %v", err) + } + + ctx := context.Background() + pool, release, err := m.resolvePool(ctx) + if err != nil { + t.Skipf("skipping (no dev postgres): %v", err) + } + defer release() + if err := pool.Ping(ctx); err != nil { + t.Skipf("skipping (no dev postgres): %v", err) + } + + schema, _ := devAppSchemaName("bbbbbbbb-0000-0000-0000-000000000002") + ident := func(s string) string { return pgx.Identifier{s}.Sanitize() } + cleanup := func() { pool.Exec(ctx, `DROP SCHEMA IF EXISTS `+ident(schema)+` CASCADE`) } + cleanup() + t.Cleanup(cleanup) + + if err := m.ensureDevAppSchema(ctx, schema); err != nil { + t.Fatalf("provision: %v", err) + } + + // Simulate the drift: the migration is recorded, but its table is gone from + // the schema (the historical bug landed it in public instead). + if _, err := pool.Exec(ctx, fmt.Sprintf(`DROP TABLE %s.mheal_items`, ident(schema))); err != nil { + t.Fatalf("simulate drift drop: %v", err) + } + // Tracking row still claims 0001 is applied. + var rows int + pool.QueryRow(ctx, fmt.Sprintf( + `SELECT count(*) FROM %s.schema_migrations WHERE module_id='mheal' AND scope='app' AND version='0001'`, + ident(schema))).Scan(&rows) + if rows != 1 { + t.Fatalf("expected tracking row pre-drop, got %d", rows) + } + + // Re-provision must detect drift and re-apply (the sync.Once is per-module + // instance; a fresh instance models the next dev run). + m2, _ := New(Config{ID: "mheal", Name: "Heal", SQL: sqlFS}) + if err := m2.ensureDevAppSchema(ctx, schema); err != nil { + t.Fatalf("re-provision: %v", err) + } + + conn, err := pool.Acquire(ctx) + if err != nil { + t.Fatalf("acquire: %v", err) + } + defer conn.Release() + if got := qualifiedTables(t, conn.Conn(), schema, "mheal"); got != 1 { + t.Errorf("after self-heal, mheal tables in %s = %d, want 1", schema, got) + } +}