feat(observability): report why a process runs without a pending drops cleaner

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.
 

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.

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,28 +109,34 @@ 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.
 
 ## Observability
 
 | 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

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

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")

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

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)