diff --git a/docs/architecture.md b/docs/architecture.md index 33f35df6a..067f1b0f7 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -209,10 +209,11 @@ Unsafe operations that produce error-severity violations: - `MODIFY COLUMN` (type change) — may truncate or lose data if the new type is narrower - `DROP INDEX` without first making it invisible — may cause query performance regression -For MySQL/Spirit databases, an applied `DROP TABLE` is additionally quarantined -instead of executed: the table is renamed into a `_pending_drops` database and -stays recoverable until a background cleaner drops it after the retention -period. See [Pending Drops](pending-drops.md). +For MySQL/Spirit databases, an applied `DROP TABLE` can additionally be +quarantined instead of executed: the table is renamed into a `_pending_drops` +database and stays recoverable until a background cleaner drops it after the +retention period. The quarantine is opt-in per deployment and off by default. +See [Pending Drops](pending-drops.md). `HasErrors()` on the plan result checks if any lint warning has error severity. The CLI, webhook check runs, and PR comments all use this to gate applies and surface warnings to reviewers. diff --git a/docs/configuration.md b/docs/configuration.md index 73a6942cd..0b7fdef8a 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -339,25 +339,31 @@ The listener binds `:9102` by default; set `metrics_port` to move it. The API po ## Pending Drops -For MySQL databases executed by the Spirit engine, `DROP TABLE` statements are -quarantined by default: the table is renamed into a `_pending_drops` database -on the target server with a timestamp prefix instead of being dropped. The data -stays recoverable until a background cleaner permanently drops the table after +For MySQL databases executed by the Spirit engine, `DROP TABLE` statements can +be quarantined instead of executed: the table is renamed into a +`_pending_drops` database on the target server with a timestamp prefix, and the +data stays recoverable until a background cleaner permanently drops it after the retention period. See [Pending Drops](pending-drops.md) for the full lifecycle and recovery steps. +The quarantine is **off by default**. With no `pending_drops` block, `DROP +TABLE` executes as written. + ```yaml pending_drops: - enabled: true # default: true + enabled: true # default: false (the quarantine is opt-in) cleanup_enabled: true # default: true when enabled is true retention: 168h # default: 168h (7 days) dry_run: false # default: false — set true to log instead of dropping ``` -Set `enabled: false` to execute `DROP TABLE` directly with no quarantine and no -cleaner. Set `cleanup_enabled: false` to keep quarantining drops while this -server process does not run the background cleaner; use that for frequently -redeployed executors when a stable operator deployment owns cleanup. +Set `enabled: true` only on a deployment whose cleaner can reach its own +targets: the quarantine leaves tables on the target server that only the +cleaner removes, so enabling one without the other grows the target's disk with +nothing scheduled to reclaim it. Set `cleanup_enabled: false` to keep +quarantining drops while this server process does not run the background +cleaner; use that for frequently redeployed executors when a stable operator +deployment owns cleanup for the same targets. `dry_run: true` keeps the quarantine active but makes the cleaner log the tables it would drop without dropping them, which is useful when previewing a retention change. diff --git a/docs/pending-drops.md b/docs/pending-drops.md index 2317b9bbf..ef7bd0427 100644 --- a/docs/pending-drops.md +++ b/docs/pending-drops.md @@ -20,6 +20,18 @@ server. The table keeps all of its data and can be restored with a single `RENAME TABLE` until a background cleaner permanently drops it after the retention period (7 days by default). +**Pending drops is opt-in and off by default.** With no `pending_drops` block, +`DROP TABLE` executes as written and the table is gone, recoverable only from +the target's own backups. + +Enable it only on a deployment whose cleaner can reach the targets it executes +against. The quarantine and the cleaner are two halves of one feature: the +quarantine writes tables into `_pending_drops` on the target server, and only +the cleaner removes them. A deployment that quarantines without reaping leaves +tables on its targets that nothing will ever delete, which costs the target +owner disk indefinitely and is worse than the direct drop it replaced. See +[Cleanup coordination](#cleanup-coordination) for which deployments can reap. + Pending drops applies to MySQL databases executed by the Spirit engine. Vitess online DDL has its own native table lifecycle that holds dropped tables before purging them, so the PlanetScale engine does not need a SchemaBot-side @@ -97,17 +109,21 @@ pass; one bad target never blocks cleanup for the others. ```yaml pending_drops: - enabled: true # default: true + enabled: true # default: false (the quarantine is opt-in) cleanup_enabled: true # default: true when enabled is true retention: 168h # default: 168h (7 days) dry_run: false # default: false ``` +`enabled: true` is the whole opt-in: it turns on both the quarantine and this +process's cleaner. Leaving the block out entirely executes `DROP TABLE` +directly. + Set `cleanup_enabled: false` on frequently redeployed executors when a stable operator deployment owns cleanup. Quarantine remains active as long as -`enabled` is true. Set `dry_run: true` to keep quarantine active while making -the cleaner log the expired tables it would drop without permanently dropping -them. +`enabled` is true, so use this only when another deployment reaps the same +targets. Set `dry_run: true` to keep quarantine active while making the cleaner +log the expired tables it would drop without permanently dropping them. See [Configuration](configuration.md#pending-drops) for field semantics. @@ -115,10 +131,12 @@ See [Configuration](configuration.md#pending-drops) for field semantics. | Signal | Meaning | | --- | --- | +| `schemabot.drop_table.already_absent_total` | DROP TABLE targets that were already absent when the apply reached them, by database and environment. Expected after a stopped apply resumes, because the DROP phase replays from its first statement. Outside that, the schema files and the target have diverged. | | `schemabot.pending_drops.tables_moved_total` | Tables quarantined instead of dropped, by database and environment. | | `schemabot.pending_drops.cleanup_dropped_total` | Expired quarantined tables permanently dropped by the cleaner. | | `schemabot.pending_drops.cleanup_skipped_total` | Quarantined tables skipped because their names carry no valid timestamp prefix. A sustained nonzero rate means tables are accumulating that an operator must inspect and remove manually. | | `schemabot.pending_drops.cleanup_lock_skipped_total` | Cleanup target passes skipped because another instance held the per-target advisory lock. | +| `schemabot.pending_drops.cleaner_not_started_total` | Processes that started without a cleaner, by `reason`. `quarantine_disabled` and `no_local_targets` are the expected states for a process that drops directly and for a control plane that routes every target elsewhere. `cleanup_disabled_for_process` is only safe while another deployment reaps the same targets. `invalid_retention` is a config bug that blocks reaping until it is fixed. | | `schemabot.pending_drops.cleanup_errors_total` | Cleanup failures by reason (`dsn_resolution_error`, `locker_missing`, `target_error`, or `drop_error`). Failed work retries on the next pass, but `locker_missing` is a producer wiring bug that fails every pass until fixed. | The apply log for a schema change that drops a table records the quarantine diff --git a/pkg/api/config.go b/pkg/api/config.go index 0ccb17580..e59885c3e 100644 --- a/pkg/api/config.go +++ b/pkg/api/config.go @@ -164,10 +164,12 @@ type ServerConfig struct { RespondToUnscoped *bool `yaml:"respond_to_unscoped"` // PendingDrops configures the pending drops quarantine for MySQL/Spirit - // databases. When enabled (the default), DROP TABLE statements rename the - // table into the _pending_drops database instead of dropping it. The - // background cleaner can be run by this process or disabled so another - // deployment owns permanent cleanup after the retention period. + // databases. When enabled, DROP TABLE statements rename the table into the + // _pending_drops database instead of dropping it, and the background cleaner + // removes it once the retention period expires. The quarantine is off by + // default: enable it only on a deployment whose cleaner can reach its own + // targets, because a quarantine no cleaner reaps grows on the target server + // forever. PendingDrops PendingDropsConfig `yaml:"pending_drops,omitempty"` // Spirit overrides the Spirit engine's default run settings for every @@ -183,7 +185,8 @@ type ServerConfig struct { // databases. type PendingDropsConfig struct { // Enabled controls the quarantine. - // Defaults to true when not configured (nil = enabled). + // Defaults to false when not configured (nil = disabled), so a DROP TABLE + // executes as written. Enabled *bool `yaml:"enabled"` // CleanupEnabled controls whether this server process starts the background @@ -214,9 +217,18 @@ func (c SupportChannelConfig) Enabled() bool { } // PendingDropsEnabled reports whether the pending drops quarantine is enabled. -// Defaults to true when not configured. +// Defaults to false when not configured. +// +// Quarantine and reaping are one feature, and a deployment gets both or +// neither. Defaulting the quarantine on made every deployment rename dropped +// tables into a schema on the target server, while reaping them required a +// cleaner reaching that same server, so a deployment that quarantined but could +// not reap accumulated tables on its targets with nothing scheduled to remove +// them. Off by default keeps the two halves from being enabled independently: +// turning the quarantine on is a deliberate statement that this deployment +// reaps its own targets. func (c *ServerConfig) PendingDropsEnabled() bool { - return c.PendingDrops.Enabled == nil || *c.PendingDrops.Enabled + return c.PendingDrops.Enabled != nil && *c.PendingDrops.Enabled } // PendingDropsCleanupEnabled reports whether this process should run the diff --git a/pkg/api/config_test.go b/pkg/api/config_test.go index e7d2c3b0e..9bb9d01df 100644 --- a/pkg/api/config_test.go +++ b/pkg/api/config_test.go @@ -3616,9 +3616,16 @@ repos: func TestPendingDropsConfig(t *testing.T) { boolPtr := func(b bool) *bool { return &b } - t.Run("enabled by default", func(t *testing.T) { + t.Run("disabled by default", func(t *testing.T) { cfg := ServerConfig{} + assert.False(t, cfg.PendingDropsEnabled()) + assert.False(t, cfg.PendingDropsCleanupEnabled()) + }) + + t.Run("explicit enable turns on the quarantine and its cleaner together", func(t *testing.T) { + cfg := ServerConfig{PendingDrops: PendingDropsConfig{Enabled: boolPtr(true)}} assert.True(t, cfg.PendingDropsEnabled()) + assert.True(t, cfg.PendingDropsCleanupEnabled()) }) t.Run("explicit disable", func(t *testing.T) { @@ -3627,17 +3634,21 @@ func TestPendingDropsConfig(t *testing.T) { assert.False(t, cfg.PendingDropsCleanupEnabled()) }) - t.Run("cleanup enabled by default", func(t *testing.T) { - cfg := ServerConfig{} - assert.True(t, cfg.PendingDropsCleanupEnabled()) - }) - t.Run("cleanup can be disabled without disabling quarantine", func(t *testing.T) { - cfg := ServerConfig{PendingDrops: PendingDropsConfig{CleanupEnabled: boolPtr(false)}} + cfg := ServerConfig{PendingDrops: PendingDropsConfig{Enabled: boolPtr(true), CleanupEnabled: boolPtr(false)}} assert.True(t, cfg.PendingDropsEnabled()) assert.False(t, cfg.PendingDropsCleanupEnabled()) }) + t.Run("cleanup stays off when only the cleaner is enabled", func(t *testing.T) { + // Enabling the cleaner without the quarantine would start a reaper for a + // deployment that writes no quarantined tables of its own, which is how + // one deployment ends up sweeping another's targets. + cfg := ServerConfig{PendingDrops: PendingDropsConfig{CleanupEnabled: boolPtr(true)}} + assert.False(t, cfg.PendingDropsEnabled()) + assert.False(t, cfg.PendingDropsCleanupEnabled()) + }) + t.Run("default retention", func(t *testing.T) { cfg := ServerConfig{} retention, err := cfg.PendingDropsRetention() @@ -3674,7 +3685,7 @@ func TestPendingDropsConfig(t *testing.T) { }, }, }, - PendingDrops: PendingDropsConfig{Retention: "not-a-duration"}, + PendingDrops: PendingDropsConfig{Enabled: boolPtr(true), Retention: "not-a-duration"}, } err := cfg.Validate() assert.ErrorContains(t, err, "pending_drops.retention") diff --git a/pkg/api/pending_drops_cleaner.go b/pkg/api/pending_drops_cleaner.go index 0d61867b0..bb7b33225 100644 --- a/pkg/api/pending_drops_cleaner.go +++ b/pkg/api/pending_drops_cleaner.go @@ -23,8 +23,21 @@ const PendingDropsCleanupInterval = 6 * time.Hour // targets are configured (gRPC-mode targets are cleaned by the deployment that // executes the schema changes). func (s *Service) StartPendingDropsCleaner(ctx context.Context) { + // The ways cleanup can be off mean different things to an operator and need + // different responses, so they are reported separately rather than through + // the single PendingDropsCleanupEnabled predicate. Each decline is counted + // by reason: quarantining without reaping is what makes tables accumulate + // on a target, so the reason a process is not reaping is the thing an + // operator needs to be able to alert on. + if !s.config.PendingDropsEnabled() { + s.logger.Info("pending drops cleaner not started because the quarantine is disabled; DROP TABLE executes as written, so there is nothing to reap") + metrics.RecordPendingDropsCleanerNotStarted(ctx, "quarantine_disabled") + return + } + if !s.config.PendingDropsCleanupEnabled() { - s.logger.Info("pending drops cleaner not started because cleanup is disabled for this process") + s.logger.Info("pending drops cleaner not started because cleanup is disabled for this process; another deployment must reap this deployment's targets or quarantined tables will accumulate on them") + metrics.RecordPendingDropsCleanerNotStarted(ctx, "cleanup_disabled_for_process") return } @@ -33,11 +46,21 @@ func (s *Service) StartPendingDropsCleaner(ctx context.Context) { // Validate() rejects invalid retention before the server starts, so // this guards direct embedders that skip config validation. s.logger.Error("pending drops cleaner not started because retention is invalid; quarantined tables will accumulate until the config is fixed", "error", err) + metrics.RecordPendingDropsCleanerNotStarted(ctx, "invalid_retention") return } - if !s.hasPendingDropsLocalTargets() { - s.logger.Info("pending drops cleaner not started because no local MySQL database targets are configured") + local, remote := s.pendingDropsTargetCounts() + if local == 0 { + // A control plane that routes every target over gRPC executes nothing + // itself, so it has nothing to reap and this is its expected state. + // The counts separate that from a process that executes against + // targets it never configured, which quarantines tables no cleaner + // will reach. + s.logger.Info("pending drops cleaner not started because no local MySQL database targets are configured; the deployment that executes against each target reaps it", + "remote_mysql_targets", remote, + ) + metrics.RecordPendingDropsCleanerNotStarted(ctx, "no_local_targets") return } @@ -121,18 +144,26 @@ func (s *Service) runPendingDropsCleanupPass(ctx context.Context, retention time return nil } -func (s *Service) hasPendingDropsLocalTargets() bool { +// pendingDropsTargetCounts returns how many configured MySQL database +// environments this process executes against itself (local) and how many it +// routes to another deployment (remote). Only local targets are reapable from +// here; the remote count tells an operator whether a process with no local +// targets is a control plane whose deployments reap their own targets, or a +// process with no MySQL topology at all. +func (s *Service) pendingDropsTargetCounts() (local, remote int) { for _, dbConfig := range s.config.Databases { if dbConfig.Type != storage.DatabaseTypeMySQL { continue } for _, envConfig := range dbConfig.Environments { if envConfig.HasLocalDSN() { - return true + local++ + continue } + remote++ } } - return false + return local, remote } // pendingDropsTargets resolves the local-mode MySQL databases the cleaner diff --git a/pkg/api/pending_drops_cleaner_integration_test.go b/pkg/api/pending_drops_cleaner_integration_test.go index 9a1ee530d..a9b707069 100644 --- a/pkg/api/pending_drops_cleaner_integration_test.go +++ b/pkg/api/pending_drops_cleaner_integration_test.go @@ -22,6 +22,10 @@ import ( "github.com/block/schemabot/pkg/testutil" ) +// pendingDropsEnabled opts these tests into the quarantine, which is off by +// default so a deployment only quarantines when it also reaps its own targets. +var pendingDropsEnabled = true + // The service-level pending drops cleaner starts a scheduled background loop // and runs an immediate cleanup pass, so expired quarantined tables are dropped // without waiting for the next interval. @@ -64,7 +68,7 @@ func TestStartPendingDropsCleanerDropsExpiredTable(t *testing.T) { }, }, }, - PendingDrops: PendingDropsConfig{Retention: "24h"}, + PendingDrops: PendingDropsConfig{Enabled: &pendingDropsEnabled, Retention: "24h"}, }, nil, slog.New(slog.NewTextHandler(io.Discard, nil))) cleanerCtx, cancel := context.WithCancel(ctx) diff --git a/pkg/api/pending_drops_cleaner_test.go b/pkg/api/pending_drops_cleaner_test.go index dcad8c660..ef7cbdf6c 100644 --- a/pkg/api/pending_drops_cleaner_test.go +++ b/pkg/api/pending_drops_cleaner_test.go @@ -1,6 +1,7 @@ package api import ( + "bytes" "io" "log/slog" "maps" @@ -69,10 +70,11 @@ func TestPendingDropsTargetsIncludeOnlyLocalMySQLDatabases(t *testing.T) { "non-MySQL databases must be skipped by policy, not counted as resolution failures") } -// hasPendingDropsLocalTargets gates whether the cleaner loop starts at all: a -// deployment whose local databases are all non-MySQL has no quarantine to -// clean, so the loop must not start for them. -func TestHasPendingDropsLocalTargetsIgnoresNonMySQLDatabases(t *testing.T) { +// The local count gates whether the cleaner loop starts at all, and the remote +// count tells an operator whether a process with no local targets is a control +// plane whose deployments reap their own targets. Only MySQL databases are +// counted: no other family has a quarantine to reap. +func TestPendingDropsTargetCountsIgnoresNonMySQLDatabases(t *testing.T) { t.Parallel() nonMySQL := map[string]DatabaseConfig{ "postgres_local": { @@ -88,8 +90,9 @@ func TestHasPendingDropsLocalTargetsIgnoresNonMySQLDatabases(t *testing.T) { }, }, } - assert.False(t, newPendingDropsTestService(t, nonMySQL).hasPendingDropsLocalTargets(), - "non-MySQL local databases must not enable the pending drops cleaner") + local, remote := newPendingDropsTestService(t, nonMySQL).pendingDropsTargetCounts() + assert.Zero(t, local, "non-MySQL local databases must not enable the pending drops cleaner") + assert.Zero(t, remote, "non-MySQL databases must not be counted as remote MySQL targets") withMySQL := maps.Clone(nonMySQL) withMySQL["mysql_local"] = DatabaseConfig{ @@ -98,6 +101,98 @@ func TestHasPendingDropsLocalTargetsIgnoresNonMySQLDatabases(t *testing.T) { "staging": {DSN: "user:pass@tcp(127.0.0.1:3306)/mysql_local"}, }, } - assert.True(t, newPendingDropsTestService(t, withMySQL).hasPendingDropsLocalTargets(), - "a mysql-typed local database enables the pending drops cleaner") + local, remote = newPendingDropsTestService(t, withMySQL).pendingDropsTargetCounts() + assert.Equal(t, 1, local, "a mysql-typed local database enables the pending drops cleaner") + assert.Zero(t, remote) +} + +// A control plane routes every MySQL target to the deployment that executes +// against it, so it has no local target to reap and its remote count is what +// tells an operator the targets are covered elsewhere. +func TestPendingDropsTargetCountsSeparatesRoutedTargets(t *testing.T) { + t.Parallel() + routed := map[string]DatabaseConfig{ + "mysql_routed": { + Type: storage.DatabaseTypeMySQL, + Environments: map[string]EnvironmentConfig{ + "staging": {Target: "shard-1", Deployment: "executor"}, + "production": {Target: "shard-2", Deployment: "executor"}, + }, + }, + "mysql_local": { + Type: storage.DatabaseTypeMySQL, + Environments: map[string]EnvironmentConfig{ + "staging": {DSN: "user:pass@tcp(127.0.0.1:3306)/mysql_local"}, + }, + }, + } + local, remote := newPendingDropsTestService(t, routed).pendingDropsTargetCounts() + assert.Equal(t, 1, local, "only the database with a local DSN is reapable from this process") + assert.Equal(t, 2, remote, "each routed environment is reaped by the deployment that executes it") +} + +// The cleaner declines to start for reasons that mean different things to an +// operator, and the message must say which one applies: a process with the +// quarantine off has nothing to reap, while a process that quarantines but +// leaves reaping to another deployment accumulates tables on its targets until +// that deployment runs. +func TestStartPendingDropsCleanerReportsWhyItDeclined(t *testing.T) { + localMySQL := map[string]DatabaseConfig{ + "mysql_local": { + Type: storage.DatabaseTypeMySQL, + Environments: map[string]EnvironmentConfig{ + "staging": {DSN: "user:pass@tcp(127.0.0.1:3306)/mysql_local"}, + }, + }, + } + enabled, disabled := true, false + + tests := []struct { + name string + config PendingDropsConfig + databases map[string]DatabaseConfig + wantLog string + }{ + { + name: "quarantine disabled", + config: PendingDropsConfig{}, + databases: localMySQL, + wantLog: "the quarantine is disabled", + }, + { + name: "cleanup disabled for this process", + config: PendingDropsConfig{Enabled: &enabled, CleanupEnabled: &disabled}, + databases: localMySQL, + wantLog: "another deployment must reap", + }, + { + name: "retention invalid", + config: PendingDropsConfig{Enabled: &enabled, Retention: "not-a-duration"}, + databases: localMySQL, + wantLog: "retention is invalid", + }, + { + name: "no local targets", + config: PendingDropsConfig{Enabled: &enabled}, + databases: map[string]DatabaseConfig{}, + wantLog: "no local MySQL database targets are configured", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var logs bytes.Buffer + svc := New(nil, &ServerConfig{Databases: tt.databases, PendingDrops: tt.config}, + nil, slog.New(slog.NewTextHandler(&logs, nil))) + + svc.StartPendingDropsCleaner(t.Context()) + t.Cleanup(svc.StopPendingDropsCleaner) + + svc.pendingDropsMu.Lock() + started := svc.pendingDropsCancel != nil + svc.pendingDropsMu.Unlock() + assert.False(t, started, "the cleaner loop must not run") + assert.Contains(t, logs.String(), tt.wantLog) + }) + } } diff --git a/pkg/api/service.go b/pkg/api/service.go index 6e10b4a0c..9d6f6238c 100644 --- a/pkg/api/service.go +++ b/pkg/api/service.go @@ -11,6 +11,7 @@ import ( "maps" "net/http" "os" + "strconv" "strings" "sync" "sync/atomic" @@ -597,9 +598,11 @@ func (s *Service) newLocalTernClient(key, database, dbType string, envConfig Env if envConfig.APIURL != "" { metadata["api_url"] = envConfig.APIURL } - if !s.config.PendingDropsEnabled() { - metadata["pending_drops"] = "false" - } + // Stated either way rather than only when disabled: a data plane that + // predates the opt-in default reads an absent key as "quarantine", so + // leaving it out during a rolling deploy would quarantine on a deployment + // that has turned the quarantine off. + metadata["pending_drops"] = strconv.FormatBool(s.config.PendingDropsEnabled()) spiritMetadata, err := s.config.SpiritMetadata() if err != nil { return nil, fmt.Errorf("resolve spirit config for %s: %w", key, err) diff --git a/pkg/engine/spirit/drop.go b/pkg/engine/spirit/drop.go new file mode 100644 index 000000000..a978dfe22 --- /dev/null +++ b/pkg/engine/spirit/drop.go @@ -0,0 +1,122 @@ +// drop.go parses DROP TABLE statements into their targets and executes them +// directly against the target, for the deployments that drop tables outright +// rather than quarantining them in the pending drops database. +package spirit + +import ( + "context" + "fmt" + + "github.com/block/spirit/pkg/dbconn/sqlescape" + "github.com/block/spirit/pkg/parser/ast" + "github.com/block/spirit/pkg/statement" + "github.com/block/spirit/pkg/utils" + + "github.com/block/schemabot/pkg/metrics" + "github.com/block/schemabot/pkg/mysqlconn" +) + +// dropTarget is one table named by a DROP TABLE statement, with the statement's +// database filled in when the name is unqualified. +type dropTarget struct { + schema string + table string +} + +// parseDropTableStatement parses a single DROP TABLE statement into its AST +// node. DROP VIEW and DROP TEMPORARY TABLE parse into the same node; callers +// distinguish them with isNonTableDrop. +func parseDropTableStatement(stmt string) (*ast.DropTableStmt, error) { + parsed, err := statement.New(stmt) + if err != nil { + return nil, fmt.Errorf("parse DROP TABLE statement: %w", err) + } + if len(parsed) != 1 { + return nil, fmt.Errorf("expected exactly 1 parsed DROP TABLE statement, got %d", len(parsed)) + } + dropStmt, ok := (*parsed[0].StmtNode).(*ast.DropTableStmt) + if !ok { + return nil, fmt.Errorf("statement is not DROP TABLE: %s", stmt) + } + return dropStmt, nil +} + +// isNonTableDrop reports whether the statement drops a view or a temporary +// table rather than a base table. Neither holds recoverable table data, and +// neither participates in the pending drops quarantine or the existence check +// on the direct path, so both are executed as written. +func isNonTableDrop(dropStmt *ast.DropTableStmt) bool { + return dropStmt.IsView || dropStmt.TemporaryKeyword != ast.TemporaryNone +} + +// dropTableTargets returns the tables the statement names, qualifying any +// unqualified name with the database the statement runs against. +func dropTableTargets(dropStmt *ast.DropTableStmt, database string) []dropTarget { + targets := make([]dropTarget, 0, len(dropStmt.Tables)) + for _, table := range dropStmt.Tables { + schema := table.Schema.String() + if schema == "" { + schema = database + } + targets = append(targets, dropTarget{schema: schema, table: table.Name.String()}) + } + return targets +} + +// executeDropDirectly drops each table the statement names, skipping the ones +// that are already absent. +// +// The declarative differ emits a bare DROP TABLE, and the DROP phase re-runs +// from its first statement every time an apply resumes, so a phase that dropped +// some of its tables before being stopped would otherwise fail on the second +// attempt with "unknown table" and never reach the tables still standing. +// Skipping an absent table converges on exactly the state the plan asked for: +// the table is gone. The skip is logged because on a first attempt it means +// something outside this apply removed the table. +func (e *Engine) executeDropDirectly(ctx context.Context, host, username, password, database, stmt string) error { + dropStmt, err := parseDropTableStatement(stmt) + if err != nil { + return err + } + // A view or temporary table has no base-table row in information_schema. + // tables to check, and MySQL's own IF EXISTS already tolerates a missing + // target, so both go straight through as written. + if isNonTableDrop(dropStmt) || dropStmt.IfExists { + return e.executeSingleStatement(ctx, host, username, password, database, stmt) + } + + db, err := mysqlconn.Open(targetDSN(host, username, password, database)) + if err != nil { + return fmt.Errorf("open database %s: %w", database, err) + } + defer utils.CloseAndLog(db) + if err := db.PingContext(ctx); err != nil { + return fmt.Errorf("ping database %s: %w", database, err) + } + + for _, target := range dropTableTargets(dropStmt, database) { + exists, err := tableExistsInSchema(ctx, db, target.schema, target.table) + if err != nil { + return fmt.Errorf("check table `%s`.`%s` exists: %w", target.schema, target.table, err) + } + if !exists { + e.logger.Warn("DROP TABLE target is already absent; nothing to drop", + "database", target.schema, + "table", target.table, + ) + e.emitTableLog(target.table, "table was already absent; nothing to drop") + metrics.RecordDropTableAlreadyAbsent(ctx, target.schema) + continue + } + // IF EXISTS carries the convergence through the window between the + // check above and the drop itself: a table another actor removes in + // that window has already reached the state the plan asked for, and + // the check is what decides whether to report the table as absent. + drop := fmt.Sprintf("DROP TABLE IF EXISTS %s.%s", + sqlescape.EscapeIdentifier(target.schema), sqlescape.EscapeIdentifier(target.table)) + if err := e.executeSingleStatement(ctx, host, username, password, database, drop); err != nil { + return fmt.Errorf("drop table `%s`.`%s`: %w", target.schema, target.table, err) + } + } + return nil +} diff --git a/pkg/engine/spirit/execution.go b/pkg/engine/spirit/execution.go index 8dfd7dbe0..3b8dec37d 100644 --- a/pkg/engine/spirit/execution.go +++ b/pkg/engine/spirit/execution.go @@ -79,9 +79,10 @@ func (e *Engine) executeSchemaChange(ctx context.Context, host, username, passwo } } - // Execute DROP TABLE statements last. By default each table is quarantined - // in the pending drops database instead of dropped; when pending drops is - // disabled the DROP runs directly through Spirit. + // Execute DROP TABLE statements last. Each table is dropped directly + // through Spirit unless the deployment enabled the pending drops + // quarantine, in which case it is renamed into the pending drops database + // instead. if !e.executeDropStatements(ctx, host, username, password, database, phases.drops) { return } @@ -262,24 +263,25 @@ func (e *Engine) executeAlterPhase(ctx context.Context, host, username, password return e.executeSpiritMigration(ctx, host, username, password, database, combinedStatement, deferCutover) == nil } -// executeDropStatements runs the DROP TABLE phase. By default each table is -// quarantined in the pending drops database so its data stays recoverable until -// the retention period expires; when pending drops is disabled the DROP runs -// directly through Spirit. Both the initial-apply DROP phase and the resume DROP -// phase call this helper, so a resumed DROP quarantines exactly like an initial -// one. It returns false when execution should stop: a cancelled context leaves -// the state Stopped, a genuine failure transitions to StateFailed. +// executeDropStatements runs the DROP TABLE phase. Each table is dropped +// directly through Spirit unless the deployment enabled the pending drops +// quarantine, in which case it is renamed into the pending drops database so +// its data stays recoverable until the retention period expires. Both the +// initial-apply DROP phase and the resume DROP phase call this helper, so a +// resumed DROP behaves exactly like an initial one. It returns false when +// execution should stop: a cancelled context leaves the state Stopped, a +// genuine failure transitions to StateFailed. func (e *Engine) executeDropStatements(ctx context.Context, host, username, password, database string, drops []string) bool { return e.runStatementPhase(ctx, database, "DROP TABLE", "DROP TABLE phase", drops, func(ctx context.Context, stmt string) error { return e.executeDropStatement(ctx, host, username, password, database, stmt) }) } -// executeDropStatement runs a single DROP TABLE statement, quarantining the -// table by default and dropping it directly only when pending drops is disabled. +// executeDropStatement runs a single DROP TABLE statement, dropping the table +// outright unless the pending drops quarantine is enabled for this deployment. func (e *Engine) executeDropStatement(ctx context.Context, host, username, password, database, stmt string) error { if e.disablePendingDrops { - if err := e.executeSingleStatement(ctx, host, username, password, database, stmt); err != nil { + if err := e.executeDropDirectly(ctx, host, username, password, database, stmt); err != nil { return fmt.Errorf("drop table directly: %w", err) } return nil diff --git a/pkg/engine/spirit/pending_drops.go b/pkg/engine/spirit/pending_drops.go index 6e84d05df..c34b88e5b 100644 --- a/pkg/engine/spirit/pending_drops.go +++ b/pkg/engine/spirit/pending_drops.go @@ -9,8 +9,6 @@ import ( "fmt" "time" - "github.com/block/spirit/pkg/parser/ast" - "github.com/block/spirit/pkg/statement" "github.com/block/spirit/pkg/utils" "github.com/block/schemabot/pkg/metrics" @@ -23,21 +21,14 @@ import ( // database instead of being dropped. IF EXISTS semantics are preserved — // missing tables are skipped when the statement allows it. func (e *Engine) quarantineDroppedTables(ctx context.Context, host, username, password, database, stmt string) error { - parsed, err := statement.New(stmt) + dropStmt, err := parseDropTableStatement(stmt) if err != nil { - return fmt.Errorf("parse DROP TABLE statement: %w", err) - } - if len(parsed) != 1 { - return fmt.Errorf("expected exactly 1 parsed DROP TABLE statement, got %d", len(parsed)) - } - dropStmt, ok := (*parsed[0].StmtNode).(*ast.DropTableStmt) - if !ok { - return fmt.Errorf("statement is not DROP TABLE: %s", stmt) + return err } // DROP VIEW and DROP TEMPORARY TABLE also parse as DropTableStmt. Neither // holds recoverable table data, and neither can be renamed into the pending // drops database with table semantics, so execute them as written. - if dropStmt.IsView || dropStmt.TemporaryKeyword != ast.TemporaryNone { + if isNonTableDrop(dropStmt) { e.logger.Info("executing non-table drop directly without pending drops quarantine", "database", database, "statement", stmt, @@ -59,28 +50,23 @@ func (e *Engine) quarantineDroppedTables(ctx context.Context, host, username, pa return fmt.Errorf("ping database %s: %w", database, err) } - tables := make([]pendingdrops.TableMove, 0, len(dropStmt.Tables)) - for _, table := range dropStmt.Tables { - tableName := table.Name.String() - schemaName := table.Schema.String() - if schemaName == "" { - schemaName = database - } - + targets := dropTableTargets(dropStmt, database) + tables := make([]pendingdrops.TableMove, 0, len(targets)) + for _, target := range targets { if dropStmt.IfExists { - exists, err := tableExistsInSchema(ctx, db, schemaName, tableName) + exists, err := tableExistsInSchema(ctx, db, target.schema, target.table) if err != nil { - return fmt.Errorf("check table `%s`.`%s` exists: %w", schemaName, tableName, err) + return fmt.Errorf("check table `%s`.`%s` exists: %w", target.schema, target.table, err) } if !exists { e.logger.Info("DROP TABLE IF EXISTS target does not exist, skipping quarantine", - "database", schemaName, - "table", tableName, + "database", target.schema, + "table", target.table, ) continue } } - tables = append(tables, pendingdrops.TableMove{SchemaName: schemaName, TableName: tableName}) + tables = append(tables, pendingdrops.TableMove{SchemaName: target.schema, TableName: target.table}) } moved, err := pendingdrops.MoveTables(ctx, db, tables, time.Now()) diff --git a/pkg/engine/spirit/pending_drops_integration_test.go b/pkg/engine/spirit/pending_drops_integration_test.go index 5890b6b5d..bd11e5eb4 100644 --- a/pkg/engine/spirit/pending_drops_integration_test.go +++ b/pkg/engine/spirit/pending_drops_integration_test.go @@ -255,3 +255,127 @@ func TestEngine_ExecuteSchemaChange_DropTemporaryTableExecutesDirectly(t *testin quarantined := listQuarantinedTables(t, db) assert.Empty(t, quarantined, "temporary tables must not create quarantine entries") } + +// The DROP phase re-runs from its first statement whenever an apply resumes, +// so on a deployment that drops tables outright the phase must tolerate the +// tables an earlier attempt already dropped and still drop the ones that +// remain, rather than failing the apply on the first absent table. +func TestEngine_ExecuteSchemaChange_DirectDropSkipsAlreadyAbsentTables(t *testing.T) { + dsn, db := setupTestMySQL(t) + cleanupTables(t, db) + cleanupPendingDropsDB(t, db) + + for _, name := range []string{"resumed_first", "resumed_second"} { + _, err := db.ExecContext(t.Context(), + fmt.Sprintf("CREATE TABLE `%s` (id INT PRIMARY KEY AUTO_INCREMENT)", name)) + require.NoError(t, err, "create table %s", name) + } + + // Stand in for the earlier attempt: the first table is already dropped + // when the phase restarts from the top. + _, err := db.ExecContext(t.Context(), "DROP TABLE `resumed_first`") + require.NoError(t, err, "pre-drop resumed_first") + + logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelDebug})) + eng := New(Config{Logger: logger, DisablePendingDrops: true}) + + state := runDDLApply(t, eng, dsn, []string{ + "DROP TABLE `resumed_first`", + "DROP TABLE `resumed_second`", + }) + assert.Equal(t, engine.StateCompleted, state) + + var count int + err = db.QueryRowContext(t.Context(), + "SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = 'testdb' AND table_name = 'resumed_second'").Scan(&count) + require.NoError(t, err) + assert.Equal(t, 0, count, "the statement after the absent table must still run") + + quarantined := listQuarantinedTables(t, db) + assert.Empty(t, quarantined, "the direct path must not quarantine anything") +} + +// A multi-table DROP on the direct path drops the tables that are still there +// and leaves the already-absent ones alone, so a stop partway through a +// multi-table statement still converges on the planned end state. +func TestEngine_ExecuteSchemaChange_DirectDropMultiTablePartiallyAbsent(t *testing.T) { + dsn, db := setupTestMySQL(t) + cleanupTables(t, db) + cleanupPendingDropsDB(t, db) + + _, err := db.ExecContext(t.Context(), + "CREATE TABLE `partial_present` (id INT PRIMARY KEY AUTO_INCREMENT)") + require.NoError(t, err, "create table") + + logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelDebug})) + eng := New(Config{Logger: logger, DisablePendingDrops: true}) + + state := runDDLApply(t, eng, dsn, []string{"DROP TABLE `partial_absent`, `partial_present`"}) + assert.Equal(t, engine.StateCompleted, state) + + var count int + err = db.QueryRowContext(t.Context(), + "SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = 'testdb' AND table_name = 'partial_present'").Scan(&count) + require.NoError(t, err) + assert.Equal(t, 0, count, "the table that was still present must be dropped") +} + +// The direct path forwards drops that carry no base-table data or that already +// tolerate a missing target: views, temporary tables, and an explicit IF +// EXISTS. Each must execute as written rather than going through the existence +// check, which only understands base tables. +func TestEngine_ExecuteSchemaChange_DirectDropForwardsBypassedStatements(t *testing.T) { + tests := []struct { + name string + setup string + ddl string + absent string + }{ + { + name: "view", + setup: "CREATE VIEW `direct_view` AS SELECT 1 AS one", + ddl: "DROP VIEW `direct_view`", + absent: "direct_view", + }, + { + name: "if exists on a present table", + setup: "CREATE TABLE `direct_if_exists` (id INT PRIMARY KEY AUTO_INCREMENT)", + ddl: "DROP TABLE IF EXISTS `direct_if_exists`", + absent: "direct_if_exists", + }, + { + name: "if exists on a missing table", + ddl: "DROP TABLE IF EXISTS `direct_never_existed`", + absent: "direct_never_existed", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + dsn, db := setupTestMySQL(t) + cleanupTables(t, db) + cleanupPendingDropsDB(t, db) + + if tt.setup != "" { + _, err := db.ExecContext(t.Context(), tt.setup) + require.NoError(t, err, "setup") + } + + logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelDebug})) + eng := New(Config{Logger: logger, DisablePendingDrops: true}) + + state := runDDLApply(t, eng, dsn, []string{tt.ddl}) + assert.Equal(t, engine.StateCompleted, state) + + var count int + err := db.QueryRowContext(t.Context(), + "SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = 'testdb' AND table_name = ?", + tt.absent).Scan(&count) + require.NoError(t, err) + assert.Equal(t, 0, count, "%s should not exist in testdb", tt.absent) + + quarantined := listQuarantinedTables(t, db) + assert.Empty(t, quarantined, "the direct path must not quarantine anything") + }) + } +} diff --git a/pkg/metrics/README.md b/pkg/metrics/README.md index 37915f81b..a93f8b055 100644 --- a/pkg/metrics/README.md +++ b/pkg/metrics/README.md @@ -53,10 +53,12 @@ available, such as `repository`, `github_app`, and `installation_id`. | `schemabot.operator.stuck_pending_applies` | Gauge | environment | Pending applies past the stuck threshold that a driver should have claimed (sampled; capped at 500, so a value of 500 means "at least 500") | | `schemabot.operator.stuck_pending_scan_failures` | Counter | environment | Failed stuck-pending apply scans (liveness signal for the gauge above) | | `schemabot.operator.stranded_operations_reaped_total` | Counter | database, deployment, environment, parent_state | Pending apply operations the reaper settled from an already-settled parent apply. `deployment` is the reaped operation's own. A one-time burst is the historical backlog draining; a climbing rate means a producer is terminalizing parents without settling their children | +| `schemabot.drop_table.already_absent_total` | Counter | database, environment | DROP TABLE targets that were already absent when the apply reached them | | `schemabot.pending_drops.tables_moved_total` | Counter | database, environment | Dropped tables quarantined into the pending drops database | | `schemabot.pending_drops.cleanup_dropped_total` | Counter | database, environment | Expired quarantined tables permanently dropped by the cleaner | | `schemabot.pending_drops.cleanup_skipped_total` | Counter | database, environment | Quarantined tables skipped by the cleaner due to unparseable names | | `schemabot.pending_drops.cleanup_lock_skipped_total` | Counter | database, environment | Cleanup target passes skipped because another instance held the per-target advisory lock | +| `schemabot.pending_drops.cleaner_not_started_total` | Counter | reason | Processes that started without a pending drops cleaner, by why it declined | | `schemabot.pending_drops.cleanup_errors_total` | Counter | database, environment, reason | Pending drops cleanup failures (retried on the next pass) | > **Deprecated aliases:** the `schemabot.scheduler.*` series (`resumed_total`, `resume_failures_total`, `claim_failures_total`, `claim_duration_seconds`) is still emitted alongside the `schemabot.operator.*` series for one release so dashboards and alerts can migrate. The scheduler-named series will be removed afterward. diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go index a19daefab..b8cd678a0 100644 --- a/pkg/metrics/metrics.go +++ b/pkg/metrics/metrics.go @@ -2025,6 +2025,20 @@ func RecordPendingDropMoved(ctx context.Context, database string) { ) } +// RecordDropTableAlreadyAbsent increments the counter for a DROP TABLE target +// that was already gone when the apply reached it. The DROP phase replays from +// its first statement on resume, so a stopped and resumed apply produces these +// for the tables its earlier attempt dropped. Outside that, it means something +// other than the apply removed the table, and the schema files and the target +// have diverged. +func RecordDropTableAlreadyAbsent(ctx context.Context, database string) { + addCounter(ctx, "schemabot.drop_table.already_absent_total", + "Total number of DROP TABLE targets that were already absent when the apply reached them", "{table}", + attribute.String("database", database), + EnvironmentAttribute(""), + ) +} + // knownDirectExecutionOutcomes limits metric cardinality to the outcomes the // direct execution path can produce. Executed statements terminate as // completed, failed, or stopped; refused statements the policy does not route @@ -2091,6 +2105,35 @@ func RecordPendingDropsCleanupLockSkipped(ctx context.Context, database, environ ) } +// knownPendingDropsCleanerDeclines limits metric cardinality to the reasons the +// cleaner can decline to start. +var knownPendingDropsCleanerDeclines = map[string]bool{ + "quarantine_disabled": true, + "cleanup_disabled_for_process": true, + "invalid_retention": true, + "no_local_targets": true, +} + +// RecordPendingDropsCleanerNotStarted increments the counter for a process that +// started without a pending drops cleaner, by reason. Quarantining without +// reaping is what leaves tables on a target forever, so this is the signal that +// a deployment is running the half of the feature that costs disk without the +// half that reclaims it. quarantine_disabled and no_local_targets are the +// expected states for a process that drops directly and for a control plane +// that routes every target elsewhere; cleanup_disabled_for_process is only safe +// while another deployment reaps the same targets, and invalid_retention is a +// config bug that blocks reaping until it is fixed. +func RecordPendingDropsCleanerNotStarted(ctx context.Context, reason string) { + if !knownPendingDropsCleanerDeclines[reason] { + reason = "unknown" + } + addCounter(ctx, "schemabot.pending_drops.cleaner_not_started_total", + "Total number of processes that started without a pending drops cleaner", "{process}", + attribute.String("reason", reason), + EnvironmentAttribute(""), + ) +} + // RecordPendingDropsCleanupError increments the counter for pending drops // cleanup failures. Failed targets and tables are retried on the next cleanup // pass. diff --git a/pkg/serve/serve.go b/pkg/serve/serve.go index 9373a8b8b..e793e00e6 100644 --- a/pkg/serve/serve.go +++ b/pkg/serve/serve.go @@ -734,20 +734,20 @@ func buildGRPCTernClient(ctx context.Context, config *api.ServerConfig, st stora // LocalClient the data plane builds, so the router and single-database paths // share identical execution semantics and can resolve custom database types. func grpcLocalClientFactory(config *api.ServerConfig, wakeOperator func(applyIdentifier, database, environment string), engineFactories map[string]tern.EngineFactory) tern.LocalClientFactory { - pendingDropsDisabled := !config.PendingDropsEnabled() + pendingDrops := strconv.FormatBool(config.PendingDropsEnabled()) return func(cfg tern.LocalConfig, st storage.Storage, logger *slog.Logger) (tern.Client, error) { spiritMetadata, err := config.SpiritMetadata() if err != nil { return nil, fmt.Errorf("resolve spirit config for database %q: %w", cfg.Database, err) } - if pendingDropsDisabled || len(spiritMetadata) > 0 { - if cfg.Metadata == nil { - cfg.Metadata = map[string]string{} - } - } - if pendingDropsDisabled { - cfg.Metadata["pending_drops"] = "false" + if cfg.Metadata == nil { + cfg.Metadata = map[string]string{} } + // Stated either way rather than only when disabled: a data plane that + // predates the opt-in default reads an absent key as "quarantine", so + // leaving it out during a rolling deploy would quarantine on a + // deployment that has turned the quarantine off. + cfg.Metadata["pending_drops"] = pendingDrops // Server-level spirit overrides are defaults; a database's own // metadata entry for the same key wins. for key, value := range spiritMetadata { diff --git a/pkg/tern/local_client.go b/pkg/tern/local_client.go index 6da4d86eb..e03215528 100644 --- a/pkg/tern/local_client.go +++ b/pkg/tern/local_client.go @@ -300,10 +300,8 @@ func NewLocalClient(cfg LocalConfig, stor storage.Storage, logger *slog.Logger) config: cfg, storage: stor, spiritEngine: spirit.New(spirit.Config{ - Logger: logger, - // Pending drops quarantine is on by default; deployments opt out - // via the pending_drops metadata key. - DisablePendingDrops: cfg.Metadata["pending_drops"] == "false", + Logger: logger, + DisablePendingDrops: pendingDropsDisabled(cfg.Metadata), Settings: spiritSettings, }), planetscaleEngine: psEngine, @@ -3170,3 +3168,15 @@ func dsnLogAttrs(dsn string) []any { "target_db", cfg.DBName, } } + +// pendingDropsDisabled reports whether this client drops tables outright rather +// than quarantining them in the pending drops database. +// +// The quarantine is opt-in: a deployment turns it on with the pending_drops +// metadata key, and anything else, including an absent key, drops the table +// outright. Quarantining is only safe for a deployment that also reaps its own +// targets, because a quarantine no cleaner reaches grows on the target server +// forever, so an embedder that never states the intent must not inherit it. +func pendingDropsDisabled(metadata map[string]string) bool { + return metadata["pending_drops"] != "true" +} diff --git a/pkg/tern/local_client_pending_drops_test.go b/pkg/tern/local_client_pending_drops_test.go new file mode 100644 index 000000000..c3077c5e6 --- /dev/null +++ b/pkg/tern/local_client_pending_drops_test.go @@ -0,0 +1,33 @@ +package tern + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +// The pending drops quarantine is opt-in, and the metadata key is how a server +// states that intent to the data plane. An embedder that builds a LocalConfig +// itself never states it, so the absent key must drop tables outright: a +// deployment that quarantines without reaping leaves tables on its targets that +// nothing will ever remove. +func TestPendingDropsDisabledRequiresAnExplicitOptIn(t *testing.T) { + t.Parallel() + tests := []struct { + name string + metadata map[string]string + want bool + }{ + {name: "nil metadata", metadata: nil, want: true}, + {name: "key absent", metadata: map[string]string{"organization": "acme"}, want: true}, + {name: "explicitly disabled", metadata: map[string]string{"pending_drops": "false"}, want: true}, + {name: "unrecognized value", metadata: map[string]string{"pending_drops": "yes"}, want: true}, + {name: "explicitly enabled", metadata: map[string]string{"pending_drops": "true"}, want: false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + assert.Equal(t, tt.want, pendingDropsDisabled(tt.metadata)) + }) + } +}