diff --git a/docs/configuration.md b/docs/configuration.md index 6523d3592..55c38675e 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -18,6 +18,7 @@ - [Pending Drops](#pending-drops) - [Direct Execution](#direct-execution) - [Storage Dialect](#storage-dialect) + - [Resyncing PostgreSQL identity sequences](#resyncing-postgresql-identity-sequences) - [Storage Connection Pool](#storage-connection-pool) - [Spirit Run Settings](#spirit-run-settings) - [PlanetScale mTLS](#planetscale-mtls) @@ -503,6 +504,29 @@ supports both `mysql` and `postgres`.) See [Storage Schema Changes](#storage-schema-changes) for how schema bootstrapping differs between the two dialects. +### Resyncing PostgreSQL identity sequences + +After an explicit-ID bulk load into PostgreSQL storage, advance the identity +sequences before SchemaBot resumes default inserts. Run the command only after +the load has fully committed. It advances sequences when needed, never rewinds +them, and is idempotent and safe to rerun. + +Pass the storage DSN directly: + +```shell +schemabot storage resync-identity-sequences --dsn "$STORAGE_DSN" +``` + +Or resolve it from the server configuration: + +```shell +schemabot storage resync-identity-sequences --config /etc/schemabot/config.yaml +``` + +The command refuses to run when none of SchemaBot's storage tables exist in +the target database. Confirm the target and complete the resync before +restarting the server or otherwise allowing default inserts. + ## Storage Connection Pool SchemaBot tunes the `database/sql` pool for its internal storage database with diff --git a/pkg/api/resync_postgres_identity.go b/pkg/api/resync_postgres_identity.go index 32f5f8cbe..3c96b1e04 100644 --- a/pkg/api/resync_postgres_identity.go +++ b/pkg/api/resync_postgres_identity.go @@ -36,33 +36,48 @@ import ( func ResyncPostgresIdentitySequences(ctx context.Context, db *sql.DB, logger *slog.Logger) error { tables, _, err := readEmbeddedPostgresSchemaFiles() if err != nil { - return err + return fmt.Errorf("read embedded PostgreSQL storage schema: %w", err) + } + missing, err := missingPostgresTables(ctx, db, tables) + if err != nil { + return fmt.Errorf("check target for storage tables: %w", err) + } + if len(missing) == len(tables) { + return fmt.Errorf("none of the %d storage tables exist in the target database; it does not look like SchemaBot's storage database", len(tables)) } columns, err := postgresIdentityColumns(ctx, db, tables) if err != nil { - return err + return fmt.Errorf("find identity columns on storage tables: %w", err) } + advanced := 0 + skipped := 0 for _, col := range columns { newValue, outcome, err := advancePostgresIdentitySequence(ctx, db, col) if err != nil { - return err + return fmt.Errorf("resync identity column %s.%s: %w", col.table, col.column, err) } switch outcome { case sequenceAdvanced: + advanced++ logger.Info("advanced identity sequence past stored maximum", "table", col.table, "column", col.column, "sequence_value", newValue) case sequenceSkippedEmptyTable: + skipped++ logger.Debug("identity column has no stored rows; sequence left untouched", "table", col.table, "column", col.column) case sequenceSkippedAlreadyAhead: + skipped++ logger.Debug("identity sequence already at or past stored maximum; left untouched", "table", col.table, "column", col.column) case sequenceSkippedDescending: + skipped++ logger.Warn("identity sequence is descending; resync skipped it — a stored-maximum resync only applies to ascending sequences", "table", col.table, "column", col.column) } } + logger.Info("identity sequence resync summary", + "tables", len(tables), "examined", len(columns), "advanced", advanced, "skipped", skipped) return nil } diff --git a/pkg/api/resync_postgres_identity_integration_test.go b/pkg/api/resync_postgres_identity_integration_test.go index 53522ef46..dac9c985b 100644 --- a/pkg/api/resync_postgres_identity_integration_test.go +++ b/pkg/api/resync_postgres_identity_integration_test.go @@ -102,6 +102,16 @@ func resyncLogger(t *testing.T) (*slog.Logger, *recordingLogHandler) { return slog.New(handler), handler } +// A target without any SchemaBot storage tables is rejected before sequence +// discovery, preventing an unrelated PostgreSQL database from appearing to +// resync successfully. +func TestResyncPostgresIdentitySequences_RejectsTargetWithoutStorageTables(t *testing.T) { + _, db := startPostgresStorage(t) + err := ResyncPostgresIdentitySequences(t.Context(), db, slog.New(slog.DiscardHandler)) + require.ErrorContains(t, err, "none of the") + require.ErrorContains(t, err, "does not look like SchemaBot's storage database") +} + // After an explicit-id bulk load, the identity sequences still point below // the loaded rows — GENERATED BY DEFAULT AS IDENTITY accepts explicit values // without advancing the sequence — so the next default insert collides with diff --git a/pkg/cmd/commands/storage.go b/pkg/cmd/commands/storage.go new file mode 100644 index 000000000..ccfdab322 --- /dev/null +++ b/pkg/cmd/commands/storage.go @@ -0,0 +1,142 @@ +package commands + +import ( + "context" + "fmt" + "log/slog" + "os" + "strings" + "time" + + "github.com/block/spirit/pkg/utils" + + "github.com/block/schemabot/pkg/api" + "github.com/block/schemabot/pkg/postgresconn" + "github.com/block/schemabot/pkg/schema" +) + +// StorageCmd groups operator commands that act directly on SchemaBot's own +// storage database. Unlike the API-client commands, these connect to storage +// themselves and work while the server is down — they exist for maintenance +// windows such as a cross-dialect data move or a restore from a dump. +type StorageCmd struct { + ResyncIdentitySequences ResyncIdentitySequencesCmd `cmd:"" name:"resync-identity-sequences" help:"Advance PostgreSQL identity sequences on storage tables past their columns' stored maxima after an explicit-id bulk load; run after the load has fully committed and before default inserts resume — advance-only and safe to rerun."` +} + +// ResyncIdentitySequencesCmd resyncs the identity sequences of SchemaBot's +// PostgreSQL storage tables after an explicit-id bulk load — a data move +// that preserves ids, or a restore from a dump without sequence state — +// so default inserts resume above the loaded ids instead of colliding with +// them. Run it after the load has fully committed and before the server +// resumes default inserts. The resync is advance-only and idempotent, so +// rerunning it is safe. +// +// The storage DSN comes from --dsn directly, or from the server config +// (--config, falling back to $SCHEMABOT_CONFIG_FILE) whose storage dialect +// must be postgres. +type ResyncIdentitySequencesCmd struct { + DSN string `help:"PostgreSQL DSN of the storage database to resync; bypasses the server config"` + Config string `help:"Server config file to resolve the storage DSN from; defaults to $SCHEMABOT_CONFIG_FILE when neither flag is set"` +} + +// storagePingTimeout bounds the connection check so an unreachable storage +// database fails the command promptly instead of hanging it. +const storagePingTimeout = 10 * time.Second + +func (cmd *ResyncIdentitySequencesCmd) Run(ctx context.Context, g *Globals) error { + // A text handler, deliberately: this is a one-shot operator command read + // at a terminal during a maintenance window, not a long-running server + // whose stdout feeds a JSON log collector. Diagnostics go to stderr so + // stdout stays free for machine-readable output. + logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{ + Level: logLevel(), + })).With("schemabot_version", g.Version) + + dsn, source, err := cmd.resolveStorageDSN() + if err != nil { + return err + } + logger.Info("resolved storage DSN", "source", source) + + db, err := postgresconn.Open(dsn) + if err != nil { + return fmt.Errorf("open storage database: %w", err) + } + defer utils.CloseAndLog(db) + pingCtx, cancel := context.WithTimeout(ctx, storagePingTimeout) + defer cancel() + if err := db.PingContext(pingCtx); err != nil { + return fmt.Errorf("ping storage database: %w", err) + } + + if err := api.ResyncPostgresIdentitySequences(ctx, db, logger); err != nil { + return fmt.Errorf("resync identity sequences on storage tables: %w", err) + } + logger.Info("identity sequence resync complete") + return nil +} + +// resolveStorageDSN returns the storage DSN and a loggable description of +// where it came from. A direct --dsn is used as-is; otherwise the server +// config (--config, then $SCHEMABOT_CONFIG_FILE) is loaded and its resolved +// storage DSN is used, failing closed when the configured storage dialect is +// not postgres. The source never contains the DSN itself, which may embed +// credentials. +func (cmd *ResyncIdentitySequencesCmd) resolveStorageDSN() (string, string, error) { + directDSN := strings.TrimSpace(cmd.DSN) + if directDSN != "" && cmd.Config != "" { + return "", "", fmt.Errorf("--dsn and --config are mutually exclusive; pass the storage DSN directly or resolve it from a server config, not both") + } + if cmd.DSN != "" { + if directDSN == "" { + return "", "", fmt.Errorf("storage DSN not configured: --dsn contains only whitespace") + } + return directDSN, "--dsn flag", nil + } + + configPath := cmd.Config + source := fmt.Sprintf("server config %s", configPath) + if configPath == "" { + configPath = os.Getenv("SCHEMABOT_CONFIG_FILE") + if configPath == "" { + return "", "", fmt.Errorf("no storage DSN source: set --dsn, --config, or the SCHEMABOT_CONFIG_FILE environment variable") + } + source = fmt.Sprintf("server config %s ($SCHEMABOT_CONFIG_FILE)", configPath) + } + + var cfg *api.ServerConfig + var err error + if cmd.Config == "" { + cfg, err = api.LoadServerConfig() + } else { + cfg, err = api.LoadServerConfigFromFile(configPath) + } + if err != nil { + return "", "", fmt.Errorf("load %s: %w", source, err) + } + + dialect, err := cfg.Storage.ResolveDialect() + if err != nil { + return "", "", fmt.Errorf("resolve storage dialect from %s: %w", source, err) + } + if dialect != schema.DialectPostgres { + return "", "", fmt.Errorf("storage dialect in %s is %q; the identity sequence resync only applies to %q storage", source, dialect, schema.DialectPostgres) + } + + dsn, err := cfg.StorageDSN() + if err != nil { + return "", "", fmt.Errorf("resolve storage DSN from %s: %w", source, err) + } + dsn = strings.TrimSpace(dsn) + if dsn == "" { + return "", "", fmt.Errorf("storage DSN not configured (set --dsn, config storage.dsn or storage.dsn_from, STORAGE_DSN, or MYSQL_DSN)") + } + if cfg.Storage.DSN == "" && cfg.Storage.DSNFrom == nil { + if strings.TrimSpace(os.Getenv("STORAGE_DSN")) != "" { + source = "STORAGE_DSN environment variable" + } else if strings.TrimSpace(os.Getenv("MYSQL_DSN")) != "" { + source = "MYSQL_DSN environment variable" + } + } + return dsn, source, nil +} diff --git a/pkg/cmd/commands/storage_integration_test.go b/pkg/cmd/commands/storage_integration_test.go new file mode 100644 index 000000000..565003be8 --- /dev/null +++ b/pkg/cmd/commands/storage_integration_test.go @@ -0,0 +1,87 @@ +//go:build integration + +package commands + +import ( + "database/sql" + "errors" + "log/slog" + "testing" + + "github.com/jackc/pgx/v5/pgconn" + "github.com/stretchr/testify/require" + + "github.com/block/schemabot/pkg/api" + "github.com/block/schemabot/pkg/schema" + "github.com/block/schemabot/pkg/testutil" +) + +// startResyncStorage boots a PostgreSQL storage database with the full +// storage schema and seeds the settings table with explicit-id rows, the way +// an id-preserving bulk load writes them. The seeded rows leave the identity +// sequence behind the stored maximum, so a default insert collides until the +// resync runs. +func startResyncStorage(t *testing.T) (string, *sql.DB) { + t.Helper() + dsn, db := testutil.StartPostgres(t, "schemabot") + require.NoError(t, api.EnsureSchema(dsn, slog.New(slog.DiscardHandler), api.WithDialect(schema.DialectPostgres))) + for id, key := range map[int64]string{1: "loaded-1", 2: "loaded-2", 3: "loaded-3"} { + _, err := db.ExecContext(t.Context(), + `INSERT INTO settings (id, setting_key, setting_value) VALUES ($1, $2, '')`, id, key) + require.NoError(t, err) + } + return dsn, db +} + +// requireDefaultInsertResumes asserts that a default insert into settings +// draws the id above the loaded maximum, proving the resync advanced the +// sequence past the explicit-id rows. +func requireDefaultInsertResumes(t *testing.T, db *sql.DB) { + t.Helper() + var id int64 + err := db.QueryRowContext(t.Context(), + `INSERT INTO settings (setting_key, setting_value) VALUES ('after-resync', '') RETURNING id`).Scan(&id) + require.NoError(t, err, "default insert must succeed after the resync") + require.Equal(t, int64(4), id, "the first default insert after the resync draws max+1") +} + +func requireDefaultInsertCollides(t *testing.T, db *sql.DB) { + t.Helper() + _, err := db.ExecContext(t.Context(), + `INSERT INTO settings (setting_key, setting_value) VALUES ('before-resync', '')`) + require.Error(t, err) + var pgErr *pgconn.PgError + require.True(t, errors.As(err, &pgErr), "default insert must return a PostgreSQL error") + require.Equal(t, "23505", pgErr.Code, "default insert must collide before the resync") +} + +// After an explicit-id bulk load, the operator runs the resync subcommand +// with the storage DSN passed directly; afterwards default inserts resume +// above the loaded ids instead of colliding with them. +func TestResyncIdentitySequencesCmd_DSNFlag(t *testing.T) { + dsn, db := startResyncStorage(t) + requireDefaultInsertCollides(t, db) + + cmd := &ResyncIdentitySequencesCmd{DSN: dsn} + require.NoError(t, cmd.Run(t.Context(), &Globals{Version: "test"})) + + requireDefaultInsertResumes(t, db) +} + +// In a deployed pod the operator points the resync subcommand at the server +// config instead of hand-building a DSN; the command resolves the storage +// DSN from the config's storage section and resyncs the same way. +func TestResyncIdentitySequencesCmd_ConfigFile(t *testing.T) { + dsn, db := startResyncStorage(t) + requireDefaultInsertCollides(t, db) + + path := writeStorageTestConfig(t, ` +storage: + dialect: postgres + dsn: `+dsn+` +`) + cmd := &ResyncIdentitySequencesCmd{Config: path} + require.NoError(t, cmd.Run(t.Context(), &Globals{Version: "test"})) + + requireDefaultInsertResumes(t, db) +} diff --git a/pkg/cmd/commands/storage_test.go b/pkg/cmd/commands/storage_test.go new file mode 100644 index 000000000..fa1ebeb88 --- /dev/null +++ b/pkg/cmd/commands/storage_test.go @@ -0,0 +1,127 @@ +package commands + +import ( + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// writeStorageTestConfig writes a minimal valid server config with the given +// storage section and returns its path. +func writeStorageTestConfig(t *testing.T, storageSection string) string { + t.Helper() + content := storageSection + ` +databases: + testapp: + type: mysql + environments: + production: + target: testapp-production + deployment: default +tern_deployments: + default: + production: "tern-prod:9090" +` + path := filepath.Join(t.TempDir(), "config.yaml") + require.NoError(t, os.WriteFile(path, []byte(content), 0o644)) + return path +} + +func TestResolveStorageDSN_DirectDSNBypassesConfig(t *testing.T) { + cmd := &ResyncIdentitySequencesCmd{DSN: "postgres://schemabot:test@localhost:5432/schemabot"} + dsn, source, err := cmd.resolveStorageDSN() + require.NoError(t, err) + assert.Equal(t, cmd.DSN, dsn) + assert.Equal(t, "--dsn flag", source) +} + +func TestResolveStorageDSN_DSNAndConfigAreMutuallyExclusive(t *testing.T) { + cmd := &ResyncIdentitySequencesCmd{DSN: "postgres://localhost/schemabot", Config: "/etc/schemabot/config.yaml"} + _, _, err := cmd.resolveStorageDSN() + require.ErrorContains(t, err, "mutually exclusive") +} + +func TestResolveStorageDSN_NoSourceConfigured(t *testing.T) { + t.Setenv("SCHEMABOT_CONFIG_FILE", "") + cmd := &ResyncIdentitySequencesCmd{} + _, _, err := cmd.resolveStorageDSN() + require.ErrorContains(t, err, "no storage DSN source") +} + +func TestResolveStorageDSN_ConfigResolvesPostgresStorageDSN(t *testing.T) { + path := writeStorageTestConfig(t, ` +storage: + dialect: postgres + dsn: postgres://schemabot:hunter2-distinctive@storage-host:5432/schemabot +`) + cmd := &ResyncIdentitySequencesCmd{Config: path} + dsn, source, err := cmd.resolveStorageDSN() + require.NoError(t, err) + assert.Equal(t, "postgres://schemabot:hunter2-distinctive@storage-host:5432/schemabot", dsn) + assert.Contains(t, source, path) + assert.NotContains(t, source, "hunter2-distinctive", "the loggable source must not leak DSN credentials") +} + +func TestResolveStorageDSN_ConfigFromEnvFallback(t *testing.T) { + path := writeStorageTestConfig(t, ` +storage: + dialect: postgres + dsn: postgres://schemabot:test@storage-host:5432/schemabot +`) + t.Setenv("SCHEMABOT_CONFIG_FILE", path) + cmd := &ResyncIdentitySequencesCmd{} + dsn, source, err := cmd.resolveStorageDSN() + require.NoError(t, err) + assert.Equal(t, "postgres://schemabot:test@storage-host:5432/schemabot", dsn) + assert.Contains(t, source, "$SCHEMABOT_CONFIG_FILE") +} + +func TestResolveStorageDSN_RejectsNonPostgresStorageDialect(t *testing.T) { + path := writeStorageTestConfig(t, ` +storage: + dsn: user:pass@tcp(storage-host:3306)/schemabot +`) + cmd := &ResyncIdentitySequencesCmd{Config: path} + _, _, err := cmd.resolveStorageDSN() + require.ErrorContains(t, err, `only applies to "postgres" storage`) +} + +func TestResolveStorageDSN_EmptyConfigDSN(t *testing.T) { + t.Setenv("STORAGE_DSN", "") + t.Setenv("MYSQL_DSN", "") + path := writeStorageTestConfig(t, ` +storage: + dialect: postgres +`) + cmd := &ResyncIdentitySequencesCmd{Config: path} + _, _, err := cmd.resolveStorageDSN() + require.ErrorContains(t, err, "storage DSN not configured") +} + +func TestResolveStorageDSN_WhitespaceDirectDSN(t *testing.T) { + cmd := &ResyncIdentitySequencesCmd{DSN: " "} + _, _, err := cmd.resolveStorageDSN() + require.ErrorContains(t, err, "storage DSN not configured") +} + +func TestResolveStorageDSN_ReportsEnvironmentSource(t *testing.T) { + path := writeStorageTestConfig(t, ` +storage: + dialect: postgres +`) + t.Setenv("STORAGE_DSN", "postgres://schemabot@storage-host:5432/schemabot") + t.Setenv("MYSQL_DSN", "postgres://legacy@storage-host:5432/schemabot") + cmd := &ResyncIdentitySequencesCmd{Config: path} + _, source, err := cmd.resolveStorageDSN() + require.NoError(t, err) + assert.Equal(t, "STORAGE_DSN environment variable", source) +} + +func TestResyncIdentitySequencesCmd_PingFailure(t *testing.T) { + cmd := &ResyncIdentitySequencesCmd{DSN: "postgres://user@127.0.0.1:1/db?sslmode=disable"} + err := cmd.Run(t.Context(), &Globals{Version: "test"}) + require.ErrorContains(t, err, "ping storage database:") +} diff --git a/pkg/cmd/main.go b/pkg/cmd/main.go index 55e59071b..9138c2a9e 100644 --- a/pkg/cmd/main.go +++ b/pkg/cmd/main.go @@ -57,6 +57,7 @@ type CLI struct { Settings commands.SettingsCmd `cmd:"" help:"View or update schema change settings"` Webhooks commands.WebhooksCmd `cmd:"" help:"Manage GitHub App webhook deliveries"` Checks commands.ChecksCmd `cmd:"" help:"Manage SchemaBot Check Runs on PRs"` + Storage commands.StorageCmd `cmd:"" help:"Operate directly on SchemaBot's storage database"` Serve commands.ServeCmd `cmd:"" help:"Start the SchemaBot HTTP API server"` } diff --git a/pkg/cmd/main_test.go b/pkg/cmd/main_test.go index a2fc77c23..922159295 100644 --- a/pkg/cmd/main_test.go +++ b/pkg/cmd/main_test.go @@ -23,3 +23,16 @@ func TestRollbackRequiresEnvironmentFlag(t *testing.T) { require.Error(t, err) assert.Contains(t, err.Error(), "-e") } + +func TestStorageResyncIdentitySequencesIsInvocable(t *testing.T) { + var cli CLI + parser, err := kong.New(&cli, + kong.Name("schemabot"), + kong.Writers(io.Discard, io.Discard), + kong.Vars{"cli_name": "schemabot"}, + ) + require.NoError(t, err) + + _, err = parser.Parse([]string{"storage", "resync-identity-sequences", "--dsn", "postgres://user@localhost:5432/db"}) + require.NoError(t, err) +}