diff --git a/AGENTS.md b/AGENTS.md index e45cf9933..54b3d3901 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -264,7 +264,7 @@ All SQL statements processed by SchemaBot **must be parseable by the dialect's r ### Storage Schema (Self-Bootstrapping) -SchemaBot's storage schema is self-bootstrapping via `EnsureSchema` (`pkg/api/ensure_schema.go`), which runs on every server startup before accepting traffic and routes to a per-dialect bootstrapper. On MySQL it reads all embedded SQL files from `pkg/schema/mysql/`, diffs them against the live database using Spirit, and applies any DDL needed — adding a new table or column to `pkg/schema/mysql/` is all that's needed; the next deploy picks it up automatically. On PostgreSQL (`pkg/api/ensure_schema_postgres.go`) it creates missing tables from `pkg/schema/postgres/` and verifies that existing tables contain every expected column and unique index, failing startup when one is missing; a missing non-unique index only logs a startup warning, and it never alters existing tables or rejects extra columns. Apply column changes to already-bootstrapped PostgreSQL databases before deploying schema files that expect them. Keep the two dialect directories in lockstep; the schema parity tests in `pkg/schema` pin this. +SchemaBot's storage schema is self-bootstrapping via `EnsureSchema` (`pkg/api/ensure_schema.go`), which runs on every server startup before accepting traffic and routes to a per-dialect bootstrapper. On MySQL it reads all embedded SQL files from `pkg/schema/mysql/`, diffs them against the live database using Spirit, and applies any DDL needed — adding a new table or column to `pkg/schema/mysql/` is all that's needed; the next deploy picks it up automatically. On PostgreSQL (`pkg/api/ensure_schema_postgres.go`) it transactionally creates missing tables, columns, and indexes under the bootstrap advisory lock. PostgreSQL convergence is additive-only: it tolerates extra objects and checks columns by presence, while a missing `NOT NULL` column without a `DEFAULT` fails startup with a manual-remediation error. Keep the two dialect directories in lockstep; the schema parity tests in `pkg/schema` pin this. ### SQL Schema diff --git a/docs/configuration.md b/docs/configuration.md index 55cda170b..792fc9618 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -762,23 +762,34 @@ on the storage dialect: tables carry a long history, create a newly declared index by hand before rolling out: the startup diff then finds nothing to do, instead of copying the table inside the budget on every pod. -- **PostgreSQL** creates missing tables and verifies that existing tables - contain every column and standalone unique index declared by the embedded - schema. Missing objects fail startup with the affected table and objects - identified; extra columns are tolerated, and a missing non-unique index is - tolerated with a startup warning naming it. Column verification is - presence-only: type, length, and nullability drift - is outside its scope and is not detected. Existing tables are never altered, - and `allow_destructive_schema_changes` has no effect because this flow never - produces destructive DDL. Apply column changes to already-bootstrapped - PostgreSQL databases before deploying schema files that expect them. - - Non-unique indexes work the same way, and the consequence is quieter: an - index added to an embedded schema file reaches newly created databases only, - so an already-bootstrapped database keeps answering the queries that index - was added for — correctly, but without it, and startup warns about the gap - on every deploy until it is closed. Create those by hand. A database - bootstrapped before `idx_plans_created_at` was added to `plans` needs: +- **PostgreSQL** automatically creates missing tables, columns, and standalone + indexes. It discovers drift before taking the bootstrap advisory lock, then + re-checks and applies each table's changes transactionally under that lock. + A missing `NOT NULL` column without a `DEFAULT`, a generated or identity + column, or a column with a constraint shape not explicitly classified as + safe fails startup with instructions for manual remediation. Generated and + identity columns rewrite the populated table under an exclusive lock. + Startup also fails when additive DDL cannot be parsed or executed, or when + re-verification finds unresolved drift. + + Convergence is additive-only: extra columns and indexes remain in place for + binary rollback, and `allow_destructive_schema_changes` has no effect because + this flow never produces destructive DDL. Column verification remains + presence-only, so type, length, and nullability drift is outside its scope and + is not detected. + + Indexes added to an embedded schema file after a database was bootstrapped + converge on the next startup as plain `CREATE INDEX` statements, each in + its own transaction under the bootstrap advisory lock. A plain + `CREATE INDEX` holds a `SHARE` lock on the table for the full build and + blocks writes to it, and the startup budget is the build's only duration + ceiling, so on a deployment whose storage tables carry a long history, + pre-create the index by hand before rolling out — the startup diff then + finds it present and skips the build. The indexes below are the ones a + long-lived database is most likely to be missing. + + A database bootstrapped before `idx_plans_created_at` was added to `plans` + needs: ```sql CREATE INDEX idx_plans_created_at ON plans (created_at); @@ -793,7 +804,7 @@ on the storage dialect: ``` Without it, every driver claim sorts the full claimable set before taking - one row, which slows claiming as apply history grows. And one bootstrapped + one row, which slows claiming as apply history grows. One bootstrapped before refused applies started naming the schema change holding the database needs: diff --git a/pkg/api/ensure_schema_postgres.go b/pkg/api/ensure_schema_postgres.go index 182326a89..812c41d5a 100644 --- a/pkg/api/ensure_schema_postgres.go +++ b/pkg/api/ensure_schema_postgres.go @@ -7,6 +7,7 @@ import ( "fmt" "log/slog" "sort" + "strconv" "strings" "time" @@ -18,30 +19,20 @@ import ( "github.com/block/schemabot/pkg/schema" ) -// ensurePostgresSchema converges SchemaBot's storage schema on PostgreSQL by -// creating every storage table whose embedded schema file has no matching -// table in the current schema. Existing tables are checked for every expected -// column and standalone unique index, but are never altered; extra columns are -// tolerated, and a missing non-unique index is tolerated with a startup -// warning naming it. The column check is presence-only: -// type, length, and nullability drift are outside its scope and are not detected. -// That bound is deliberate — PostgreSQL has -// no in-process diff/apply mechanism here (Spirit is MySQL-only), and -// create-only convergence is sufficient to bootstrap a fresh storage -// database. Evolving an already-bootstrapped PostgreSQL storage schema -// requires a schema diff mechanism, which lands separately. +// ensurePostgresSchema converges additive drift in SchemaBot's PostgreSQL +// storage schema. It creates missing tables, columns, and standalone indexes. +// Extra objects are tolerated for binary rollback, and column comparison is +// presence-only: type, length, and nullability drift are outside its scope. // -// Because the flow only ever creates missing tables, it can never destroy -// existing data, so the destructive-change refusal that guards the MySQL flow -// (WithAllowDestructiveSchemaChanges) does not apply here. +// The flow never destroys or alters existing objects, so the destructive-change +// refusal that guards the MySQL flow does not apply here. // -// Concurrency-safe across pods: checks table existence first without a lock -// (read-only, the common case on 99% of deploys), and returns immediately when -// every table exists. When tables are missing, acquires a PostgreSQL advisory -// lock to serialize creation across pods, then re-checks under the lock — -// another pod may have created the tables while we waited. Each table's file -// executes inside one transaction, so a killed pod leaves either the whole -// table with its indexes or nothing. +// Concurrency-safe across pods: discovers drift without a lock, acquires the +// PostgreSQL advisory lock only when needed, then re-discovers under the lock. +// A change that needs manual remediation aborts the whole convergence before +// any DDL executes. Each transaction bounds its lock wait with lock_timeout. +// Plain CREATE INDEX holds a SHARE lock for the full build, blocking writes; +// EnsureSchemaTimeout is the build's only duration ceiling. func ensurePostgresSchema(dsn string, logger *slog.Logger, locker namedlock.Locker) error { ctx, cancel := context.WithTimeout(context.Background(), EnsureSchemaTimeout) defer cancel() @@ -79,22 +70,35 @@ func ensurePostgresSchema(dsn string, logger *slog.Logger, locker namedlock.Lock ) } - // Fast path: check existence without a lock. If every table exists, - // return immediately. - missing, err := missingPostgresTables(ctx, db, tables) + // Fast path: discover drift without a lock. This is the common case and + // avoids advisory-lock overhead when the schema is already converged. + drift, err := postgresSchemaDriftFor(ctx, db, tables, files) if err != nil { - return fmt.Errorf("check storage tables: %w", err) + return fmt.Errorf("inspect storage schema: %w", err) } - if len(missing) == 0 { + if len(drift) == 0 { if err := verifyAndLogPostgresSchemaShape(ctx, db, tables, files, logger, database, schemaName); err != nil { return fmt.Errorf("validate existing storage tables: %w", err) } logger.Info("storage schema up-to-date", "database", database) return nil } - logger.Info("storage tables missing (pre-lock); acquiring EnsureSchema advisory lock to create them", + // Log what the fast-path scan found before parking on the lock, so the + // last pre-lock line names the work this pod is waiting to do. + for _, table := range tables { + for _, change := range drift[table] { + logger.Info("schema change detected (pre-lock)", + "table", table, + "operation", change.operation, + "object", change.object, + "ddl", change.ddl, + ) + } + } + logger.Info("storage schema drift detected (pre-lock); acquiring EnsureSchema advisory lock to converge it", "database", database, - "tables", missing, + "change_count", drift.changeCount(), + "table_count", len(drift), ) lockConn, err := acquirePostgresEnsureSchemaLock(ctx, dsn, logger, locker) @@ -103,13 +107,13 @@ func ensurePostgresSchema(dsn string, logger *slog.Logger, locker namedlock.Lock } defer utils.CloseAndLog(lockConn) - // Re-check under the lock — another pod may have created the tables - // while we waited. - missing, err = missingPostgresTables(ctx, db, tables) + // Re-check under the lock — another pod may have converged the schema while + // this pod waited. + drift, err = postgresSchemaDriftFor(ctx, db, tables, files) if err != nil { - return fmt.Errorf("check storage tables: %w", err) + return fmt.Errorf("inspect storage schema under lock: %w", err) } - if len(missing) == 0 { + if len(drift) == 0 { if err := verifyAndLogPostgresSchemaShape(ctx, db, tables, files, logger, database, schemaName); err != nil { return fmt.Errorf("validate storage tables after lock: %w", err) } @@ -117,16 +121,29 @@ func ensurePostgresSchema(dsn string, logger *slog.Logger, locker namedlock.Lock return nil } - createStart := time.Now() - for _, table := range missing { - if err := createPostgresTable(ctx, db, table, files[table], logger); err != nil { - return fmt.Errorf("create storage table %q: %w", table, err) + // Gate on manual remediation across the whole drift set before touching + // any table, so a change that cannot run automatically never leaves the + // schema half-converged. + if err := postgresManualRemediation(tables, drift); err != nil { + return err + } + + applyStart := time.Now() + for _, table := range tables { + changes := drift[table] + if len(changes) == 0 { + logger.Debug("storage table already converged", "table", table) + continue + } + if err := applyPostgresTableChanges(ctx, db, table, changes, logger); err != nil { + return fmt.Errorf("converge storage table %q: %w", table, err) } } logger.Info("storage schema applied successfully", "database", database, - "tables_created", len(missing), - "duration", time.Since(createStart), + "change_count", drift.changeCount(), + "table_count", len(drift), + "duration", time.Since(applyStart), ) if err := verifyAndLogPostgresSchemaShape(ctx, db, tables, files, logger, database, schemaName); err != nil { return fmt.Errorf("validate converged storage tables: %w", err) @@ -134,8 +151,176 @@ func ensurePostgresSchema(dsn string, logger *slog.Logger, locker namedlock.Lock return nil } +type postgresSchemaChange struct { + operation string + object string + ddl string + // manualReason is non-empty when the change cannot run automatically and + // names why plus the remediation. Any non-empty reason aborts convergence + // before any DDL executes. + manualReason string +} + +type postgresSchemaDrift map[string][]postgresSchemaChange + +// changeCount returns the total number of planned changes across all tables. +func (d postgresSchemaDrift) changeCount() int { + total := 0 + for _, changes := range d { + total += len(changes) + } + return total +} + +// postgresIndexExpectation is one standalone CREATE INDEX statement's +// expectation: an index under this name must exist, and must be unique when +// unique is set. Indexes are matched by name only — column composition is +// not compared, so a same-named index over different columns reads as +// converged. +type postgresIndexExpectation struct { + name string + unique bool + ddl string +} + +// postgresTableExpectations is the shape one embedded schema file declares +// for its table: the CREATE TABLE statement followed by named standalone +// CREATE INDEX statements. +type postgresTableExpectations struct { + createTable string + columns []string + indexes []postgresIndexExpectation +} + +// postgresExpectationsFor parses one table's embedded schema file into the +// expectations the drift scan and the shape verification both consume. A +// trailing statement that is not a named standalone CREATE INDEX on the +// file's own table fails closed: the additive convergence could neither +// create nor verify it, so a schema file carrying one would silently stop +// being the source of truth for the live schema. +func postgresExpectationsFor(parser ddl.StatementParser, table, file string) (postgresTableExpectations, error) { + statements, err := parser.Split(file) + if err != nil { + return postgresTableExpectations{}, fmt.Errorf("split schema file for table %q: %w", table, err) + } + if len(statements) == 0 { + return postgresTableExpectations{}, fmt.Errorf("schema file for table %q has no statements", table) + } + columns, err := parser.CreateTableColumns(statements[0]) + if err != nil { + return postgresTableExpectations{}, fmt.Errorf("extract expected columns for table %q: %w", table, err) + } + expectations := postgresTableExpectations{createTable: statements[0], columns: columns} + for _, statement := range statements[1:] { + indexName, indexTable, unique, err := parser.CreateIndex(statement) + if err != nil { + return postgresTableExpectations{}, fmt.Errorf("extract expected indexes for table %q: %w", table, err) + } + if indexName == "" { + return postgresTableExpectations{}, fmt.Errorf( + "schema file for table %q contains a statement the additive convergence cannot track: %q; only named standalone CREATE INDEX statements may follow CREATE TABLE", + table, statement) + } + if indexTable != table { + return postgresTableExpectations{}, fmt.Errorf("schema file for table %q declares index %q on table %q", table, indexName, indexTable) + } + expectations.indexes = append(expectations.indexes, postgresIndexExpectation{name: indexName, unique: unique, ddl: statement}) + } + return expectations, nil +} + +func postgresSchemaDriftFor(ctx context.Context, db *sql.DB, tables []string, files map[string]string) (postgresSchemaDrift, error) { + missingTables, err := missingPostgresTables(ctx, db, tables) + if err != nil { + return nil, err + } + missingTable := make(map[string]bool, len(missingTables)) + for _, table := range missingTables { + missingTable[table] = true + } + + parser, err := ddl.ParserForDialect(schema.DialectPostgres) + if err != nil { + return nil, fmt.Errorf("select PostgreSQL statement parser: %w", err) + } + drift := make(postgresSchemaDrift) + for _, table := range tables { + expected, err := postgresExpectationsFor(parser, table, files[table]) + if err != nil { + return nil, err + } + if missingTable[table] { + drift[table] = []postgresSchemaChange{{operation: "create_table", object: table, ddl: files[table]}} + continue + } + + existingColumns, err := postgresTableColumns(ctx, db, table) + if err != nil { + return nil, err + } + for _, column := range expected.columns { + if existingColumns[column] { + continue + } + statement, err := parser.SynthesizeAddColumn(expected.createTable, column) + if err != nil { + return nil, fmt.Errorf("synthesize ADD COLUMN for %q.%q: %w", table, column, err) + } + manualReason, err := ddl.PostgresAddColumnManualReason(expected.createTable, column) + if err != nil { + return nil, fmt.Errorf("classify ADD COLUMN safety for %q.%q: %w", table, column, err) + } + drift[table] = append(drift[table], postgresSchemaChange{ + operation: "add_column", + object: column, + ddl: statement, + manualReason: manualReason, + }) + } + + existingIndexes, err := postgresTableIndexes(ctx, db, table) + if err != nil { + return nil, err + } + for _, index := range expected.indexes { + existingUnique, present := existingIndexes[index.name] + if present && (!index.unique || existingUnique) { + continue + } + // A non-unique live index cannot satisfy a unique expectation. CREATE + // INDEX would collide by name, so fail closed rather than altering it. + if present { + return nil, fmt.Errorf("storage table %q has non-unique index %q where the embedded schema requires a unique index; replace it manually", table, index.name) + } + drift[table] = append(drift[table], postgresSchemaChange{operation: "create_index", object: index.name, ddl: index.ddl}) + } + } + return drift, nil +} + +// postgresManualRemediation returns an error naming every planned change that +// needs manual remediation, or nil when all planned changes can run +// automatically. It scans the whole drift set so the gate fires before any +// table's DDL executes — an operator sees every problem at once rather than +// one per crashloop restart. +func postgresManualRemediation(tables []string, drift postgresSchemaDrift) error { + var problems []string + for _, table := range tables { + for _, change := range drift[table] { + if change.manualReason == "" { + continue + } + problems = append(problems, fmt.Sprintf("storage table %q is missing column %q whose %s", table, change.object, change.manualReason)) + } + } + if len(problems) == 0 { + return nil + } + return errors.New(strings.Join(problems, "; ")) +} + func verifyAndLogPostgresSchemaShape(ctx context.Context, db *sql.DB, tables []string, files map[string]string, logger *slog.Logger, database, schemaName string) error { - if err := verifyPostgresSchemaShape(ctx, db, tables, files, logger); err != nil { + if err := verifyPostgresSchemaShape(ctx, db, tables, files); err != nil { logger.Error("PostgreSQL storage schema shape check failed", "dialect", schema.DialectPostgres, "database", database, @@ -148,28 +333,18 @@ func verifyAndLogPostgresSchemaShape(ctx context.Context, db *sql.DB, tables []s return nil } -// verifyPostgresSchemaShape checks expected columns by presence only; it does -// not detect type, length, or nullability drift. It requires standalone -// unique indexes because losing their constraints can change write semantics. -// A missing non-unique index never alters results, so it does not fail -// startup — but it is warned about by name, because the queries it serves run -// unindexed until an operator creates it by hand (see docs/configuration.md). -func verifyPostgresSchemaShape(ctx context.Context, db *sql.DB, tables []string, files map[string]string, logger *slog.Logger) error { +// verifyPostgresSchemaShape checks the additive convergence result. Columns +// remain presence-only; type, length, and nullability drift are not detected. +// Indexes are matched by name and uniqueness only, not column composition. +func verifyPostgresSchemaShape(ctx context.Context, db *sql.DB, tables []string, files map[string]string) error { parser, err := ddl.ParserForDialect(schema.DialectPostgres) if err != nil { return fmt.Errorf("select PostgreSQL statement parser: %w", err) } for _, table := range tables { - statements, err := parser.Split(files[table]) + expected, err := postgresExpectationsFor(parser, table, files[table]) if err != nil { - return fmt.Errorf("split schema file for table %q: %w", table, err) - } - if len(statements) == 0 { - return fmt.Errorf("schema file for table %q has no statements", table) - } - expected, err := parser.CreateTableColumns(statements[0]) - if err != nil { - return fmt.Errorf("extract expected columns for table %q: %w", table, err) + return err } existing, err := postgresTableColumns(ctx, db, table) if err != nil { @@ -177,7 +352,7 @@ func verifyPostgresSchemaShape(ctx context.Context, db *sql.DB, tables []string, } var missing []string - for _, column := range expected { + for _, column := range expected.columns { if !existing[column] { missing = append(missing, column) } @@ -186,52 +361,18 @@ func verifyPostgresSchemaShape(ctx context.Context, db *sql.DB, tables []string, return fmt.Errorf("storage table %q is missing expected columns: %s", table, strings.Join(missing, ", ")) } - expectedUnique := make([]string, 0) - expectedNonUnique := make([]string, 0) - for _, statement := range statements[1:] { - indexName, indexTable, unique, err := parser.CreateIndex(statement) - if err != nil { - return fmt.Errorf("extract expected indexes for table %q: %w", table, err) - } - if indexName == "" { - // Not a standalone CREATE INDEX statement, so it declares no - // index expectation. - continue - } - if indexTable != table { - return fmt.Errorf("schema file for table %q declares index %q on table %q", table, indexName, indexTable) - } - if unique { - expectedUnique = append(expectedUnique, indexName) - } else { - expectedNonUnique = append(expectedNonUnique, indexName) - } - } existingIndexes, err := postgresTableIndexes(ctx, db, table) if err != nil { return err } - missing = nil - for _, indexName := range expectedUnique { - if !existingIndexes[indexName] { - missing = append(missing, indexName) - } - } - if len(missing) > 0 { - return fmt.Errorf("storage table %q is missing expected unique indexes: %s", table, strings.Join(missing, ", ")) - } - var missingNonUnique []string - for _, indexName := range expectedNonUnique { - if _, present := existingIndexes[indexName]; !present { - missingNonUnique = append(missingNonUnique, indexName) + var missingUnique []string + for _, index := range expected.indexes { + if index.unique && !existingIndexes[index.name] { + missingUnique = append(missingUnique, index.name) } } - if len(missingNonUnique) > 0 { - logger.Warn("storage table is missing non-unique indexes the embedded schema declares; the queries they serve run unindexed until an operator creates them by hand (see docs/configuration.md)", - "dialect", schema.DialectPostgres, - "table", table, - "indexes", strings.Join(missingNonUnique, ", "), - ) + if len(missingUnique) > 0 { + return fmt.Errorf("storage table %q is missing expected unique indexes: %s", table, strings.Join(missingUnique, ", ")) } } return nil @@ -354,16 +495,55 @@ func missingPostgresTables(ctx context.Context, db *sql.DB, want []string) ([]st return missing, nil } -// createPostgresTable executes one embedded schema file — a CREATE TABLE -// followed by its CREATE INDEX statements — inside a single transaction. -// PostgreSQL DDL is transactional, so the table appears with all of its -// indexes or not at all; an interrupted bootstrap never leaves a -// partially-indexed table behind. -// -// The file executes whole in one Exec: pgx uses the simple query protocol for -// zero-argument Execs, and the simple protocol runs a multi-statement string -// natively, so no client-side statement splitting is needed. -func createPostgresTable(ctx context.Context, db *sql.DB, table, content string, logger *slog.Logger) error { +// postgresDDLLockTimeout bounds how long a convergence DDL statement waits +// for its table lock. Without it, one long-running reader queues the ALTER +// TABLE's AccessExclusiveLock request indefinitely, and every later reader +// queues behind that request — a table-wide stall. With it, the statement +// fails, the transaction rolls back, and the startup attempt retries or +// fails visibly instead. +const postgresDDLLockTimeout = 10 * time.Second + +// applyPostgresTableChanges executes one table's additive changes. A CREATE +// TABLE change contains its complete embedded schema file, including indexes, +// executed as one transaction; pgx's simple query protocol executes that +// multi-statement string. For an existing table, column changes share one +// transaction — each is metadata-only after the manual-remediation gate — and +// each index builds in its own transaction. Plain CREATE INDEX holds a SHARE +// lock for the full build and blocks writes; lock_timeout bounds only the wait +// to acquire that lock, while EnsureSchemaTimeout bounds the build itself. +// CREATE INDEX CONCURRENTLY cannot run inside these transactions, so the write +// block is the accepted cost. Cross-transaction atomicity is unnecessary: a +// startup killed between transactions leaves additive drift the next run +// re-discovers and converges. +func applyPostgresTableChanges(ctx context.Context, db *sql.DB, table string, changes []postgresSchemaChange, logger *slog.Logger) error { + if changes[0].operation == "create_table" { + return execPostgresChanges(ctx, db, table, changes, logger) + } + var columnChanges []postgresSchemaChange + var indexChanges []postgresSchemaChange + for _, change := range changes { + if change.operation == "create_index" { + indexChanges = append(indexChanges, change) + } else { + columnChanges = append(columnChanges, change) + } + } + if len(columnChanges) > 0 { + if err := execPostgresChanges(ctx, db, table, columnChanges, logger); err != nil { + return err + } + } + for _, index := range indexChanges { + if err := execPostgresChanges(ctx, db, table, []postgresSchemaChange{index}, logger); err != nil { + return err + } + } + return nil +} + +// execPostgresChanges executes one batch of changes in a single transaction +// with a bounded lock wait. +func execPostgresChanges(ctx context.Context, db *sql.DB, table string, changes []postgresSchemaChange, logger *slog.Logger) error { tx, err := db.BeginTx(ctx, nil) if err != nil { return fmt.Errorf("begin transaction: %w", err) @@ -376,13 +556,20 @@ func createPostgresTable(ctx context.Context, db *sql.DB, table, content string, } }() - logger.Info("schema change", - "table", table, - "operation", "create", - "ddl", content, - ) - if _, err := tx.ExecContext(ctx, content); err != nil { - return fmt.Errorf("execute schema file: %w", err) + if _, err := tx.ExecContext(ctx, "SELECT set_config('lock_timeout', $1, true)", + strconv.FormatInt(postgresDDLLockTimeout.Milliseconds(), 10)); err != nil { + return fmt.Errorf("set lock_timeout for table %q: %w", table, err) + } + for _, change := range changes { + logger.Info("schema change", + "table", table, + "operation", change.operation, + "object", change.object, + "ddl", change.ddl, + ) + if _, err := tx.ExecContext(ctx, change.ddl); err != nil { + return fmt.Errorf("execute %s for %q: %w", change.operation, change.object, err) + } } if err := tx.Commit(); err != nil { return fmt.Errorf("commit: %w", err) diff --git a/pkg/api/ensure_schema_postgres_integration_test.go b/pkg/api/ensure_schema_postgres_integration_test.go index 319b9a48d..8483c0259 100644 --- a/pkg/api/ensure_schema_postgres_integration_test.go +++ b/pkg/api/ensure_schema_postgres_integration_test.go @@ -65,19 +65,22 @@ func TestEnsureSchemaPostgres_Idempotent(t *testing.T) { requireStorageTables(t, db) } -// Startup refuses an existing storage table that is missing an expected -// column and identifies the exact table shape operators need to restore. -func TestEnsureSchemaPostgres_RejectsMissingColumn(t *testing.T) { +// Startup restores an additive column from the embedded CREATE TABLE and a +// second startup observes the converged shape without executing more DDL. +func TestEnsureSchemaPostgres_ConvergesMissingColumn(t *testing.T) { ctx := t.Context() dsn, db := startPostgresStorage(t) logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelDebug})) require.NoError(t, EnsureSchema(dsn, logger, WithDialect(schema.DialectPostgres))) - _, err := db.ExecContext(ctx, "ALTER TABLE settings DROP COLUMN setting_value") + _, err := db.ExecContext(ctx, "ALTER TABLE applies DROP COLUMN caller") require.NoError(t, err) - err = EnsureSchema(dsn, logger, WithDialect(schema.DialectPostgres)) - require.ErrorContains(t, err, `storage table "settings" is missing expected columns: setting_value`) + require.NoError(t, EnsureSchema(dsn, logger, WithDialect(schema.DialectPostgres))) + require.NoError(t, EnsureSchema(dsn, logger, WithDialect(schema.DialectPostgres))) + columns, err := postgresTableColumns(ctx, db, "applies") + require.NoError(t, err) + assert.True(t, columns["caller"]) } // Startup tolerates columns unknown to the running binary so an older binary @@ -90,15 +93,18 @@ func TestEnsureSchemaPostgres_AllowsExtraColumn(t *testing.T) { require.NoError(t, EnsureSchema(dsn, logger, WithDialect(schema.DialectPostgres))) _, err := db.ExecContext(ctx, "ALTER TABLE settings ADD COLUMN future_value text") require.NoError(t, err) + _, err = db.ExecContext(ctx, "CREATE INDEX idx_settings_future_value ON settings (future_value)") + require.NoError(t, err) require.NoError(t, EnsureSchema(dsn, logger, WithDialect(schema.DialectPostgres))) + indexes, err := postgresTableIndexes(ctx, db, "settings") + require.NoError(t, err) + assert.Contains(t, indexes, "idx_settings_future_value") } -// Startup tolerates a missing non-unique index because it affects query -// performance rather than the storage model's write semantics — but it warns -// with the table and index named, so an operator learns the index must be -// created by hand instead of discovering unindexed queries later. -func TestEnsureSchemaPostgres_AllowsMissingIndex(t *testing.T) { +// Startup recreates a missing non-unique index from the schema file's own +// CREATE INDEX statement. +func TestEnsureSchemaPostgres_ConvergesMissingNonUniqueIndex(t *testing.T) { ctx := t.Context() dsn, db := startPostgresStorage(t) logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelDebug})) @@ -108,16 +114,17 @@ func TestEnsureSchemaPostgres_AllowsMissingIndex(t *testing.T) { require.NoError(t, err) var logs bytes.Buffer - warnLogger := slog.New(slog.NewTextHandler(&logs, nil)) - require.NoError(t, EnsureSchema(dsn, warnLogger, WithDialect(schema.DialectPostgres))) - assert.Contains(t, logs.String(), "missing non-unique indexes") - assert.Contains(t, logs.String(), "idx_apply_logs_level") - assert.Contains(t, logs.String(), "apply_logs") + convergeLogger := slog.New(slog.NewTextHandler(&logs, nil)) + require.NoError(t, EnsureSchema(dsn, convergeLogger, WithDialect(schema.DialectPostgres))) + indexes, err := postgresTableIndexes(ctx, db, "apply_logs") + require.NoError(t, err) + assert.Contains(t, indexes, "idx_apply_logs_level") + assert.Contains(t, logs.String(), "CREATE INDEX idx_apply_logs_level ON apply_logs (level)") } -// Startup refuses a missing unique index and identifies the exact table and -// index operators must restore before writes can safely resume. -func TestEnsureSchemaPostgres_RejectsMissingUniqueIndex(t *testing.T) { +// Startup recreates a missing unique index transactionally before accepting +// traffic that depends on its write constraint. +func TestEnsureSchemaPostgres_ConvergesMissingUniqueIndex(t *testing.T) { ctx := t.Context() dsn, db := startPostgresStorage(t) logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelDebug})) @@ -126,8 +133,71 @@ func TestEnsureSchemaPostgres_RejectsMissingUniqueIndex(t *testing.T) { _, err := db.ExecContext(ctx, "DROP INDEX idx_settings_setting_key") require.NoError(t, err) + require.NoError(t, EnsureSchema(dsn, logger, WithDialect(schema.DialectPostgres))) + indexes, err := postgresTableIndexes(ctx, db, "settings") + require.NoError(t, err) + assert.True(t, indexes["idx_settings_setting_key"]) +} + +// Startup refuses automatic convergence when the desired missing column is +// NOT NULL without a DEFAULT and gives the operator a safe remediation. +func TestEnsureSchemaPostgres_RejectsMissingNotNullColumnWithoutDefault(t *testing.T) { + ctx := t.Context() + dsn, db := startPostgresStorage(t) + logger := slog.New(slog.NewTextHandler(os.Stdout, nil)) + + require.NoError(t, EnsureSchema(dsn, logger, WithDialect(schema.DialectPostgres))) + _, err := db.ExecContext(ctx, "ALTER TABLE settings DROP COLUMN setting_value") + require.NoError(t, err) + + err = EnsureSchema(dsn, logger, WithDialect(schema.DialectPostgres)) + require.ErrorContains(t, err, `storage table "settings" is missing column "setting_value" whose definition is NOT NULL without a DEFAULT`) + require.ErrorContains(t, err, "add it manually or ship the column with a DEFAULT") +} + +// A change that needs manual remediation aborts the whole convergence before +// any DDL executes: automatic drift on another table must stay untouched +// rather than being half-applied ahead of the failure, so a crashloop never +// repeats a partial convergence. +func TestEnsureSchemaPostgres_ManualRemediationBlocksAllDDL(t *testing.T) { + ctx := t.Context() + dsn, db := startPostgresStorage(t) + logger := slog.New(slog.NewTextHandler(os.Stdout, nil)) + + require.NoError(t, EnsureSchema(dsn, logger, WithDialect(schema.DialectPostgres))) + // "applies" sorts before "settings": without the whole-set gate its + // automatic change would commit before the settings failure surfaced. + _, err := db.ExecContext(ctx, "ALTER TABLE applies DROP COLUMN caller") + require.NoError(t, err) + _, err = db.ExecContext(ctx, "ALTER TABLE settings DROP COLUMN setting_value") + require.NoError(t, err) + + err = EnsureSchema(dsn, logger, WithDialect(schema.DialectPostgres)) + require.ErrorContains(t, err, `storage table "settings" is missing column "setting_value"`) + + columns, err := postgresTableColumns(ctx, db, "applies") + require.NoError(t, err) + assert.False(t, columns["caller"], "no DDL may run when any change needs manual remediation") +} + +// A live non-unique index under a name the embedded schema requires to be +// unique cannot be converged automatically — CREATE UNIQUE INDEX would +// collide by name — so startup fails closed with a manual remediation rather +// than silently accepting the weaker index. +func TestEnsureSchemaPostgres_RejectsNonUniqueIndexWhereUniqueRequired(t *testing.T) { + ctx := t.Context() + dsn, db := startPostgresStorage(t) + logger := slog.New(slog.NewTextHandler(os.Stdout, nil)) + + require.NoError(t, EnsureSchema(dsn, logger, WithDialect(schema.DialectPostgres))) + _, err := db.ExecContext(ctx, "DROP INDEX idx_settings_setting_key") + require.NoError(t, err) + _, err = db.ExecContext(ctx, "CREATE INDEX idx_settings_setting_key ON settings (setting_key)") + require.NoError(t, err) + err = EnsureSchema(dsn, logger, WithDialect(schema.DialectPostgres)) - require.ErrorContains(t, err, `storage table "settings" is missing expected unique indexes: idx_settings_setting_key`) + require.ErrorContains(t, err, `storage table "settings" has non-unique index "idx_settings_setting_key" where the embedded schema requires a unique index`) + require.ErrorContains(t, err, "replace it manually") } // A storage database missing a subset of tables converges back to the full diff --git a/pkg/api/ensure_schema_postgres_test.go b/pkg/api/ensure_schema_postgres_test.go index 5b5c8ad61..b1d928060 100644 --- a/pkg/api/ensure_schema_postgres_test.go +++ b/pkg/api/ensure_schema_postgres_test.go @@ -77,3 +77,89 @@ func TestPostgresCreateTableColumns_EmbeddedFiles(t *testing.T) { assert.NotEmpty(t, columns, "table %s", table) } } + +// The manual-remediation gate must name every problem across every table in +// one error, so an operator fixes them all in one pass instead of one per +// startup attempt. An all-automatic drift set passes the gate untouched. +func TestPostgresManualRemediation(t *testing.T) { + t.Parallel() + + tables := []string{"applies", "settings"} + automatic := postgresSchemaDrift{ + "applies": {{operation: "add_column", object: "caller", ddl: "ALTER TABLE applies ADD COLUMN caller text"}}, + } + require.NoError(t, postgresManualRemediation(tables, automatic)) + + mixed := postgresSchemaDrift{ + "applies": { + {operation: "add_column", object: "caller", ddl: "ALTER TABLE applies ADD COLUMN caller text"}, + {operation: "add_column", object: "lock_id", manualReason: "definition is NOT NULL without a DEFAULT; add it manually or ship the column with a DEFAULT"}, + }, + "settings": { + {operation: "add_column", object: "setting_value", manualReason: "definition is NOT NULL without a DEFAULT; add it manually or ship the column with a DEFAULT"}, + }, + } + err := postgresManualRemediation(tables, mixed) + require.Error(t, err) + assert.Contains(t, err.Error(), `storage table "applies" is missing column "lock_id"`) + assert.Contains(t, err.Error(), `storage table "settings" is missing column "setting_value"`) + assert.Contains(t, err.Error(), "add it manually or ship the column with a DEFAULT") +} + +// The expectations parser fails closed on schema-file statements the additive +// convergence cannot create or verify — an unnamed index or a non-index +// trailing statement — so a schema file can never silently stop being the +// source of truth for the live schema. +func TestPostgresExpectationsFor_RejectsUntrackableStatements(t *testing.T) { + t.Parallel() + + parser, err := ddl.ParserForDialect(schema.DialectPostgres) + require.NoError(t, err) + + tests := []struct { + name string + file string + want string + }{ + { + name: "unnamed index", + file: "CREATE TABLE settings (id bigint);\nCREATE INDEX ON settings (id);", + want: "cannot track", + }, + { + name: "non-index trailing statement", + file: "CREATE TABLE settings (id bigint);\nCOMMENT ON TABLE settings IS 'x';", + want: "cannot track", + }, + { + name: "index on another table", + file: "CREATE TABLE settings (id bigint);\nCREATE INDEX idx_other ON other (id);", + want: `declares index "idx_other" on table "other"`, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + _, err := postgresExpectationsFor(parser, "settings", tt.file) + require.Error(t, err) + assert.Contains(t, err.Error(), tt.want) + }) + } +} + +// Every embedded PostgreSQL schema file must parse into trackable +// expectations: a CREATE TABLE followed only by named CREATE INDEX statements +// on the file's own table. +func TestPostgresExpectationsFor_EmbeddedFiles(t *testing.T) { + t.Parallel() + + tables, files, err := readEmbeddedPostgresSchemaFiles() + require.NoError(t, err) + parser, err := ddl.ParserForDialect(schema.DialectPostgres) + require.NoError(t, err) + for _, table := range tables { + expected, err := postgresExpectationsFor(parser, table, files[table]) + require.NoError(t, err, "table %s", table) + assert.NotEmpty(t, expected.columns, "table %s", table) + } +} diff --git a/pkg/ddl/postgres_parser.go b/pkg/ddl/postgres_parser.go index fcb74523c..159b4bb16 100644 --- a/pkg/ddl/postgres_parser.go +++ b/pkg/ddl/postgres_parser.go @@ -116,36 +116,41 @@ func (postgresStatementParser) CreateTableColumns(stmt string) ([]string, error) return columns, nil } -// SynthesizeAddColumn implements StatementParser. It grafts the column's -// ColumnDef parse node — type and column-level constraints intact — from the -// CREATE TABLE tree into a fresh ALTER TABLE ... ADD COLUMN tree and deparses -// it, so the output is libpg_query's normalized rendering rather than a -// textual slice of the input. Table-level constraints are separate TableElts -// nodes, not part of the ColumnDef, and are not carried. -func (postgresStatementParser) SynthesizeAddColumn(createTableDDL, columnName string) (string, error) { +// postgresCreateTableColumn parses exactly one PostgreSQL CREATE TABLE +// statement and returns its parse result, the CREATE TABLE node, and the +// declaration node of the named column. +func postgresCreateTableColumn(createTableDDL, columnName string) (*pgproto.ParseResult, *pgproto.Node_CreateStmt, *pgproto.Node, error) { result, err := pgquery.Parse(createTableDDL) if err != nil { - return "", fmt.Errorf("parse CREATE TABLE %q: %w", statementPreview(createTableDDL), err) + return nil, nil, nil, fmt.Errorf("parse CREATE TABLE %q: %w", statementPreview(createTableDDL), err) } stmts := result.GetStmts() if len(stmts) != 1 { - return "", fmt.Errorf("expected one CREATE TABLE statement, got %d", len(stmts)) + return nil, nil, nil, fmt.Errorf("expected one CREATE TABLE statement, got %d", len(stmts)) } createNode, ok := stmts[0].GetStmt().GetNode().(*pgproto.Node_CreateStmt) if !ok { - return "", fmt.Errorf("expected CREATE TABLE statement") + return nil, nil, nil, fmt.Errorf("expected CREATE TABLE statement") } - - var columnNode *pgproto.Node for _, element := range createNode.CreateStmt.GetTableElts() { column, ok := element.GetNode().(*pgproto.Node_ColumnDef) if ok && column.ColumnDef.GetColname() == columnName { - columnNode = element - break + return result, createNode, element, nil } } - if columnNode == nil { - return "", fmt.Errorf("column %q not found in CREATE TABLE statement", columnName) + return nil, nil, nil, fmt.Errorf("column %q not found in CREATE TABLE statement", columnName) +} + +// SynthesizeAddColumn implements StatementParser. It grafts the column's +// ColumnDef parse node — type and column-level constraints intact — from the +// CREATE TABLE tree into a fresh ALTER TABLE ... ADD COLUMN tree and deparses +// it, so the output is libpg_query's normalized rendering rather than a +// textual slice of the input. Table-level constraints are separate TableElts +// nodes, not part of the ColumnDef, and are not carried. +func (postgresStatementParser) SynthesizeAddColumn(createTableDDL, columnName string) (string, error) { + result, createNode, columnNode, err := postgresCreateTableColumn(createTableDDL, columnName) + if err != nil { + return "", err } result.Stmts = []*pgproto.RawStmt{{ @@ -165,6 +170,74 @@ func (postgresStatementParser) SynthesizeAddColumn(createTableDDL, columnName st return ddl, nil } +// PostgresAddColumnManualReason reports why adding the named column from a +// CREATE TABLE declaration to a populated table needs manual remediation, +// or "" when the synthesized ADD COLUMN is safe to run automatically. The +// decision reads the column's parsed constraint list: +// +// - Generated and identity columns rewrite the whole table under an +// exclusive lock while PostgreSQL computes values for existing rows. +// - NOT NULL without a DEFAULT needs a backfill — the server would reject +// the ADD COLUMN outright on a populated table. +// - A DEFAULT whose expression is not provably non-volatile (a constant, +// a cast of a constant, or a SQL value function such as +// CURRENT_TIMESTAMP) fails closed: the parse tree cannot see function +// volatility, and a volatile default rewrites the whole table under an +// exclusive lock. +// - Constraint shapes not explicitly known to be safe fail closed. +func PostgresAddColumnManualReason(createTableDDL, columnName string) (string, error) { + _, _, columnNode, err := postgresCreateTableColumn(createTableDDL, columnName) + if err != nil { + return "", err + } + column := columnNode.GetColumnDef() + + var notNull, hasDefault, constantDefault bool + for _, node := range column.GetConstraints() { + constraint := node.GetConstraint() + switch constraint.GetContype() { + case pgproto.ConstrType_CONSTR_NOTNULL: + notNull = true + case pgproto.ConstrType_CONSTR_DEFAULT: + hasDefault = true + constantDefault = postgresNonVolatileExpression(constraint.GetRawExpr()) + case pgproto.ConstrType_CONSTR_GENERATED, pgproto.ConstrType_CONSTR_IDENTITY: + return "definition is generated or identity, which rewrites the whole table under an exclusive lock; add it manually", nil + case pgproto.ConstrType_CONSTR_NULL, pgproto.ConstrType_CONSTR_UNIQUE, pgproto.ConstrType_CONSTR_FOREIGN: + // These constraints are safe on a nullable new column. + default: + return fmt.Sprintf("definition has constraint %s, which is not safe for automatic convergence; add it manually", constraint.GetContype().String()), nil + } + } + if hasDefault && !constantDefault { + return "definition has a DEFAULT expression whose volatility cannot be proven from the statement alone, and a volatile default rewrites the whole table under an exclusive lock; add it manually or ship the column with a constant DEFAULT", nil + } + if notNull && !hasDefault { + return "definition is NOT NULL without a DEFAULT; add it manually or ship the column with a DEFAULT", nil + } + return "", nil +} + +// postgresNonVolatileExpression reports whether a DEFAULT expression is +// provably non-volatile from its parse tree: a constant, a cast whose +// argument is itself provably non-volatile, or a SQL value function +// (CURRENT_TIMESTAMP and friends, which PostgreSQL defines as STABLE). +// Function calls report false — the parse tree carries no volatility +// information, so even a stable function like now() cannot be proven safe +// without catalog access. +func postgresNonVolatileExpression(expr *pgproto.Node) bool { + switch x := expr.GetNode().(type) { + case *pgproto.Node_AConst: + return true + case *pgproto.Node_TypeCast: + return postgresNonVolatileExpression(x.TypeCast.GetArg()) + case *pgproto.Node_SqlvalueFunction: + return true + default: + return false + } +} + // CreateIndex implements StatementParser using the parsed IndexStmt. Any // standalone CREATE INDEX statement reports its index and table names, with // unique carrying the UNIQUE declaration; other parsed statement types return diff --git a/pkg/ddl/postgres_parser_test.go b/pkg/ddl/postgres_parser_test.go index 14ac06109..0cf0e15f2 100644 --- a/pkg/ddl/postgres_parser_test.go +++ b/pkg/ddl/postgres_parser_test.go @@ -376,6 +376,108 @@ func clearParseLocations(m protoreflect.Message) { }) } +// The manual-remediation classifier reads the column's parsed constraints. +// Generated, identity, and unrecognized constraint shapes fail closed, quoted +// identifiers cannot mask a missing DEFAULT, and a function-call DEFAULT fails +// closed because its volatility cannot be proven from the statement alone. +func TestPostgresAddColumnManualReason(t *testing.T) { + tests := []struct { + name string + createDDL string + columnName string + wantReason string + }{ + { + name: "not null without default", + createDDL: "CREATE TABLE settings (id bigint, setting_value text NOT NULL)", + columnName: "setting_value", + wantReason: "NOT NULL without a DEFAULT", + }, + { + name: "not null with constant default", + createDDL: "CREATE TABLE applies (id bigint, caller varchar(255) DEFAULT '' NOT NULL)", + columnName: "caller", + }, + { + name: "nullable", + createDDL: "CREATE TABLE applies (id bigint, expected_operation_keys jsonb)", + columnName: "expected_operation_keys", + }, + { + name: "generated stored not null", + createDDL: "CREATE TABLE metrics (a bigint, doubled bigint GENERATED ALWAYS AS (a * 2) STORED NOT NULL)", + columnName: "doubled", + wantReason: "generated or identity, which rewrites the whole table under an exclusive lock", + }, + { + name: "identity not null", + createDDL: "CREATE TABLE metrics (a bigint, seq bigint GENERATED ALWAYS AS IDENTITY NOT NULL)", + columnName: "seq", + wantReason: "generated or identity, which rewrites the whole table under an exclusive lock", + }, + { + name: "primary key", + createDDL: "CREATE TABLE metrics (a bigint, seq bigint PRIMARY KEY)", + columnName: "seq", + wantReason: "constraint CONSTR_PRIMARY", + }, + { + name: "nullable unique", + createDDL: "CREATE TABLE metrics (a bigint, external_id bigint UNIQUE)", + columnName: "external_id", + }, + { + name: "nullable references", + createDDL: "CREATE TABLE metrics (a bigint, parent_id bigint REFERENCES parents (id))", + columnName: "parent_id", + }, + { + name: "quoted identifier containing default", + createDDL: `CREATE TABLE odd (id bigint, " default " text NOT NULL)`, + columnName: " default ", + wantReason: "NOT NULL without a DEFAULT", + }, + { + name: "sql value function default", + createDDL: "CREATE TABLE applies (id bigint, created_at timestamp DEFAULT CURRENT_TIMESTAMP NOT NULL)", + columnName: "created_at", + }, + { + name: "typecast constant default", + createDDL: "CREATE TABLE applies (id bigint, options jsonb DEFAULT '{}'::jsonb NOT NULL)", + columnName: "options", + }, + { + name: "volatile function default", + createDDL: "CREATE TABLE applies (id bigint, external_id uuid DEFAULT gen_random_uuid() NOT NULL)", + columnName: "external_id", + wantReason: "volatility cannot be proven", + }, + { + name: "nullable volatile function default", + createDDL: "CREATE TABLE applies (id bigint, external_id uuid DEFAULT gen_random_uuid())", + columnName: "external_id", + wantReason: "volatility cannot be proven", + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + reason, err := PostgresAddColumnManualReason(tc.createDDL, tc.columnName) + require.NoError(t, err) + if tc.wantReason == "" { + assert.Empty(t, reason) + } else { + assert.Contains(t, reason, tc.wantReason) + } + }) + } + + t.Run("column not found", func(t *testing.T) { + _, err := PostgresAddColumnManualReason("CREATE TABLE users (id bigint)", "email") + require.ErrorContains(t, err, `column "email" not found`) + }) +} + func TestPostgresParserCreateIndex(t *testing.T) { p := postgresStatementParser{}