fix(api): refuse unsplittable storage ALTERs whole, not fail startup

pkg/api/ensure_schema.go

@@ -236,21 +236,9 @@ func ensureMySQLSchema(dsn string, logger *slog.Logger, o ensureSchemaOptions, l
 			return fmt.Errorf("classify storage schema changes: %w", err)
 		}
 		for _, r := range refused {
-			attrs := []any{
-				"database", "schemabot",
-				"table", r.change.Table,
-				"operation", ddl.StatementTypeToOp(r.change.Operation),
-				"reason", r.reason,
-				"ddl", r.change.DDL,
-			}
-			if r.splitFrom != "" {
-				logger.Warn("refusing destructive clauses of a mixed storage-schema ALTER; the destructive clauses will not run, the safe clauses still execute, and startup continues — set storage.allow_destructive_schema_changes: true to allow them",
-					append(attrs, "split_from_ddl", r.splitFrom)...)
-			} else {
-				logger.Warn("refusing destructive storage-schema change; the statement will not run and startup continues — set storage.allow_destructive_schema_changes: true to allow it",
-					attrs...)
-			}
-			metrics.RecordStorageSchemaDestructiveRefusal(ctx, r.change.Table, ddl.StatementTypeToOp(r.change.Operation))
+			scope, message, attrs := r.refusalTelemetry()
+			logger.Warn(message, attrs...)
+			metrics.RecordStorageSchemaDestructiveRefusal(ctx, r.change.Table, ddl.StatementTypeToOp(r.change.Operation), scope)
 		}
 		if len(allowed) == 0 {
 			logger.Warn("all planned storage schema changes are destructive and refused; storage schema left unchanged",
@@ -357,6 +345,39 @@ type refusedStorageChange struct {
 	// splitFrom is the combined ALTER TABLE statement the refused clauses
 	// were split out of; empty when the whole statement was refused.
 	splitFrom string
+	// splitErr is the error that prevented partitioning an unsafe ALTER into
+	// safe and destructive clauses; when set, the statement was refused whole
+	// so no clause of it executed.
+	splitErr error
+}
+
+// refusalTelemetry returns the operator-facing telemetry for one refusal: the
+// metrics scope saying whether any of the statement still ran, the warning to
+// log, and its structured attributes. A split refusal carries the combined
+// ALTER its destructive clauses were split out of; a whole refusal of an
+// unsplittable ALTER carries the error that prevented the split.
+func (r refusedStorageChange) refusalTelemetry() (scope, message string, attrs []any) {
+	attrs = []any{
+		"database", "schemabot",
+		"table", r.change.Table,
+		"operation", ddl.StatementTypeToOp(r.change.Operation),
+		"reason", r.reason,
+		"ddl", r.change.DDL,
+	}
+	switch {
+	case r.splitErr != nil:
+		return metrics.StorageSchemaRefusalWhole,
+			"refusing an unsafe storage-schema ALTER whole because its clauses could not be partitioned; no clause of it will run and startup continues — set storage.allow_destructive_schema_changes: true to allow it",
+			append(attrs, "split_error", r.splitErr)
+	case r.splitFrom != "":
+		return metrics.StorageSchemaRefusalSplit,
+			"refusing destructive clauses of a mixed storage-schema ALTER; the destructive clauses will not run, the safe clauses still execute, and startup continues — set storage.allow_destructive_schema_changes: true to allow them",
+			append(attrs, "split_from_ddl", r.splitFrom)
+	default:
+		return metrics.StorageSchemaRefusalWhole,
+			"refusing destructive storage-schema change; the statement will not run and startup continues — set storage.allow_destructive_schema_changes: true to allow it",
+			attrs
+	}
 }
 
 // partitionDestructiveChanges splits planned storage-schema changes into the
@@ -367,8 +388,10 @@ type refusedStorageChange struct {
 // truncating or coalescing partitions, discarding a tablespace — is refused,
 // while structural statements that lose nothing (DROP INDEX, renames) are
 // allowed. A statement Spirit's parser cannot classify fails startup rather
-// than executing unclassified — classification uncertainty must never widen
-// what the bootstrap will execute. The Spirit diff emits an
+// than executing unclassified: a classification failure can land on a
+// statement the starting binary needs — an additive ALTER in a syntax a
+// bumped parser trips on — and skipping it would trade a loud startup
+// failure for a missing column at query time. The Spirit diff emits an
 // unsafe statement when the live storage database holds a table or column the
 // starting binary's embedded schema does not declare — during a rolling
 // deploy or rollback that surplus state usually belongs to a newer binary,
@@ -378,9 +401,15 @@ type refusedStorageChange struct {
 // still execute, and only the destructive clauses are refused. Clauses that
 // cannot run without a refused clause (the ADD PRIMARY KEY half of a
 // primary-key change) are refused with it, so the executed remainder is
-// always independently runnable. A split that cannot be performed fails
-// startup — classification or partitioning uncertainty must never widen what
-// the bootstrap will execute.
+// always independently runnable. A split that cannot be performed falls back
+// to refusing the statement whole rather than failing startup — the opposite
+// disposition from a classification failure, because a split failure only
+// ever happens on a statement already classified unsafe: the starting binary
+// demonstrably does not need it, so refusing it whole is the established
+// answer, and it executes strictly less than any split would, so the failed
+// split cannot widen what the bootstrap executes. Startup survives it, where
+// failing would crash-loop every pod whose pending ALTER the splitter cannot
+// partition.
 func partitionDestructiveChanges(changes []engine.SchemaChange) (allowed []engine.SchemaChange, refused []refusedStorageChange, err error) {
 	for _, sc := range changes {
 		kept := sc
@@ -395,9 +424,15 @@ func partitionDestructiveChanges(changes []engine.SchemaChange) (allowed []engin
 				continue
 			}
 			if tc.Operation == ddl.StatementAlterTable {
-				safeDDL, unsafeDDL, err := ddl.SplitUnsafeAlter(tc.DDL)
-				if err != nil {
-					return nil, nil, fmt.Errorf("split unsafe storage schema change for table %q (%s): %w", tc.Table, tc.DDL, err)
+				safeDDL, unsafeDDL, splitErr := ddl.SplitUnsafeAlter(tc.DDL)
+				if splitErr != nil {
+					// Refusing the statement whole executes strictly less
+					// than any split would, so the failed split cannot widen
+					// what the bootstrap executes — and startup proceeds,
+					// which is the reason this path exists. The caller logs
+					// the fallback with the split error.
+					refused = append(refused, refusedStorageChange{change: tc, reason: reason, splitErr: splitErr})
+					continue
 				}
 				if safeDDL != "" {
 					// A mixed ALTER: execute the clauses that lose nothing and

pkg/api/ensure_schema_test.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/block/schemabot/pkg/ddl"
 	"github.com/block/schemabot/pkg/engine"
+	"github.com/block/schemabot/pkg/metrics"
 	"github.com/block/schemabot/pkg/schema"
 )
 
@@ -103,6 +104,8 @@ func TestPartitionDestructiveChangesPinsUnsafeVocabulary(t *testing.T) {
 			require.Len(t, refused, 1)
 			assert.Equal(t, tt.ddl, refused[0].change.DDL)
 			assert.NotEmpty(t, refused[0].reason)
+			scope, _, _ := refused[0].refusalTelemetry()
+			assert.Equal(t, metrics.StorageSchemaRefusalWhole, scope)
 		})
 	}
 
@@ -138,6 +141,9 @@ func TestPartitionDestructiveChangesPinsUnsafeVocabulary(t *testing.T) {
 		assert.Contains(t, refused[0].reason, "DROP COLUMN")
 		assert.Contains(t, refused[0].reason, "lease_owner")
 		assert.Equal(t, "ALTER TABLE `applies` ADD COLUMN `caller` VARCHAR(64), DROP COLUMN `lease_owner`", refused[0].splitFrom)
+		scope, _, attrs := refused[0].refusalTelemetry()
+		assert.Equal(t, metrics.StorageSchemaRefusalSplit, scope)
+		assert.Contains(t, attrs, "split_from_ddl")
 	})
 
 	t.Run("a primary-key change is refused whole because its ADD half cannot run alone", func(t *testing.T) {
@@ -150,6 +156,33 @@ func TestPartitionDestructiveChangesPinsUnsafeVocabulary(t *testing.T) {
 		assert.Equal(t, pkChange, refused[0].change.DDL)
 		assert.NotEmpty(t, refused[0].reason)
 		assert.Empty(t, refused[0].splitFrom)
+		scope, _, _ := refused[0].refusalTelemetry()
+		assert.Equal(t, metrics.StorageSchemaRefusalWhole, scope)
+	})
+
+	t.Run("an unsafe ALTER whose clauses cannot be partitioned is refused whole", func(t *testing.T) {
+		t.Parallel()
+		// The Operation is set directly: the DDL carries two statements,
+		// which the splitter rejects, standing in for any split failure on
+		// an unsafe ALTER — for example a future linter rule with
+		// cross-clause reasoning tripping the safe-partition re-check. The
+		// fallback must refuse the statement whole, so nothing in it
+		// executes and the bootstrap still succeeds rather than
+		// crash-looping every starting pod.
+		multi := "ALTER TABLE `applies` DROP COLUMN `caller`; ALTER TABLE `applies` DROP COLUMN `lease_owner`"
+		changes := []engine.SchemaChange{{TableChanges: []engine.TableChange{{Table: "applies", Operation: ddl.StatementAlterTable, DDL: multi}}}}
+		allowed, refused, err := partitionDestructiveChanges(changes)
+		require.NoError(t, err)
+		assert.Empty(t, allowed)
+		require.Len(t, refused, 1)
+		assert.Equal(t, multi, refused[0].change.DDL)
+		assert.NotEmpty(t, refused[0].reason)
+		assert.Empty(t, refused[0].splitFrom)
+		require.Error(t, refused[0].splitErr)
+		scope, message, attrs := refused[0].refusalTelemetry()
+		assert.Equal(t, metrics.StorageSchemaRefusalWhole, scope)
+		assert.Contains(t, message, "could not be partitioned")
+		assert.Contains(t, attrs, "split_error")
 	})
 
 	t.Run("a statement Spirit cannot classify fails the bootstrap", func(t *testing.T) {

pkg/metrics/README.md

@@ -55,6 +55,7 @@ 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.storage_schema.destructive_refusals_total` | Counter | table, operation, scope, environment | Destructive storage-schema DDL statements the startup bootstrap (`EnsureSchema`) refused to execute. `scope` says whether the safe clauses of the statement still ran. A nonzero rate means a starting binary's embedded schema no longer declares a table or column that exists in the storage database — expected briefly from older pods during a rolling deploy or rollback. `environment` is always `unknown`: the bootstrap precedes any schema-change environment |
 | `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 |
@@ -135,6 +136,10 @@ available, such as `repository`, `github_app`, and `installation_id`.
 
 **reason** (operator resume failures): `missing_deployment`, `no_client`, `resume_error`, `lease_lost`, `retry_budget_exhausted`, `recovery_window_expired`
 
+**operation** (storage schema refusals): `alter`, `drop` — the statement types Spirit's unsafe vocabulary can flag
+
+**scope** (storage schema refusals): `split` (a mixed ALTER executed its safe clauses and refused only the destructive remainder), `whole` (nothing in the statement ran — either the whole statement was destructive or its clauses could not be partitioned)
+
 ### Webhook Ownership Rejections
 
 A webhook whose signing App does not own the target repository is rejected with

pkg/metrics/metrics.go

@@ -430,19 +430,35 @@ func RecordSourcePolicyBlock(ctx context.Context, operation, database, environme
 	)
 }
 
+// Scope values for RecordStorageSchemaDestructiveRefusal: whether the whole
+// statement was refused or only the destructive clauses split out of a mixed
+// ALTER (whose safe clauses still executed).
+const (
+	StorageSchemaRefusalWhole = "whole"
+	StorageSchemaRefusalSplit = "split"
+)
+
 // RecordStorageSchemaDestructiveRefusal increments the counter for destructive
 // storage-schema DDL statements EnsureSchema refused to execute at startup.
 // A nonzero rate means a starting binary's embedded schema no longer declares
 // a table or column that exists in the storage database — expected briefly
-// from older pods during a rolling deploy or rollback. Operator action: if the
-// removal is intended and every pod runs a binary without the table or column,
-// set storage.allow_destructive_schema_changes to true for one deploy;
-// otherwise investigate which binary is starting against newer storage state.
-func RecordStorageSchemaDestructiveRefusal(ctx context.Context, table, operation string) {
+// from older pods during a rolling deploy or rollback. The scope attribute
+// says whether the safe clauses of the statement still ran: "split" means a
+// mixed ALTER executed its safe clauses and refused only the destructive
+// remainder; "whole" means nothing in the statement ran. Operator action: if
+// the removal is intended and every pod runs a binary without the table or
+// column, set storage.allow_destructive_schema_changes to true for one
+// deploy; otherwise investigate which binary is starting against newer
+// storage state.
+func RecordStorageSchemaDestructiveRefusal(ctx context.Context, table, operation, scope string) {
 	addCounter(ctx, "schemabot.storage_schema.destructive_refusals_total",
 		"Total destructive storage-schema DDL statements refused by EnsureSchema", "{statement}",
 		attribute.String("table", table),
 		attribute.String("operation", operation),
+		attribute.String("scope", scope),
+		// The storage-schema bootstrap precedes any schema-change
+		// environment, so the counter carries the canonical unknown value.
+		EnvironmentAttribute(""),
 	)
 }