feat(storage): add the pending drops quarantine ledger

pkg/api/handlers_test.go

@@ -46,6 +46,7 @@ func (m *mockStorage) ApplyOperations() storage.ApplyOperationStore { return nil
 func (m *mockStorage) Checks() storage.CheckStore                   { return nil }
 func (m *mockStorage) Settings() storage.SettingsStore              { return nil }
 func (m *mockStorage) WebhookEvents() storage.WebhookEventStore     { return m.webhookEvents }
+func (m *mockStorage) PendingDrops() storage.PendingDropStore       { return nil }
 func (m *mockStorage) Ping(ctx context.Context) error               { return m.pingErr }
 func (m *mockStorage) Close() error                                 { return nil }
 

pkg/schema/mysql/pending_drops.sql

@@ -0,0 +1,20 @@
+CREATE TABLE `pending_drops` (
+  `id` bigint unsigned NOT NULL AUTO_INCREMENT,
+  `target` varchar(255) NOT NULL,
+  `environment` varchar(50) NOT NULL,
+  `database_name` varchar(255) NOT NULL DEFAULT '',
+  `original_table` varchar(64) NOT NULL DEFAULT '',
+  `quarantined_name` varchar(64) NOT NULL,
+  `quarantined_at` datetime(6) NOT NULL,
+  `run_id` varchar(255) NOT NULL DEFAULT '',
+  `engine` varchar(50) NOT NULL,
+  `state` varchar(20) NOT NULL,
+  `arrival_target` varchar(255) NOT NULL DEFAULT '',
+  `metadata` json NOT NULL,
+  `created_at` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  `updated_at` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+  PRIMARY KEY (`id`),
+  UNIQUE KEY `idx_pending_drops_quarantined_name` (`target`,`environment`,`quarantined_name`),
+  KEY `idx_pending_drops_expiry` (`state`,`quarantined_at`),
+  KEY `idx_pending_drops_origin` (`target`,`environment`,`database_name`,`original_table`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci

pkg/schema/postgres/pending_drops.sql

@@ -0,0 +1,20 @@
+CREATE TABLE pending_drops (
+  id bigint GENERATED BY DEFAULT AS IDENTITY,
+  target varchar(255) NOT NULL,
+  environment varchar(50) NOT NULL,
+  database_name varchar(255) NOT NULL DEFAULT '',
+  original_table varchar(64) NOT NULL DEFAULT '',
+  quarantined_name varchar(64) NOT NULL,
+  quarantined_at timestamp NOT NULL,
+  run_id varchar(255) NOT NULL DEFAULT '',
+  engine varchar(50) NOT NULL,
+  state varchar(20) NOT NULL,
+  arrival_target varchar(255) NOT NULL DEFAULT '',
+  metadata jsonb NOT NULL,
+  created_at timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  updated_at timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  PRIMARY KEY (id)
+);
+CREATE UNIQUE INDEX idx_pending_drops_quarantined_name ON pending_drops (target, environment, quarantined_name);
+CREATE INDEX idx_pending_drops_expiry ON pending_drops (state, quarantined_at);
+CREATE INDEX idx_pending_drops_origin ON pending_drops (target, environment, database_name, original_table);

pkg/storage/internal/sqlstore/pending_drops.go

@@ -0,0 +1,212 @@
+package sqlstore
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/block/spirit/pkg/utils"
+
+	"github.com/block/schemabot/pkg/storage"
+)
+
+const pendingDropColumns = `id, target, environment, database_name, original_table,
+	quarantined_name, quarantined_at, run_id, engine, state, arrival_target,
+	metadata, created_at, updated_at`
+
+// pendingDropConflictColumns is the unique key that identifies one quarantined
+// table within a deployment's ledger. A target cannot hold two tables under the
+// same quarantined name, so a conflict always means the row is already recorded.
+var pendingDropConflictColumns = []string{"target", "environment", "quarantined_name"}
+
+type pendingDropStore struct {
+	db      *rebindDB
+	dialect Dialect
+}
+
+func (s *pendingDropStore) Record(ctx context.Context, drops []*storage.PendingDrop) error {
+	if len(drops) == 0 {
+		return nil
+	}
+
+	syntax := s.dialect.InsertIfAbsent(pendingDropConflictColumns)
+	query := `INSERT` + syntax.Modifier + ` INTO pending_drops (
+		target, environment, database_name, original_table,
+		quarantined_name, quarantined_at, run_id, engine, state, arrival_target, metadata
+	) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)` + syntax.Suffix
+
+	// One statement per row rather than a multi-row VALUES list: the rows in a
+	// single call are the tables of one RENAME, so the count is small, and
+	// per-row statements keep an insert-if-absent conflict on one table from
+	// deciding the outcome of its siblings.
+	tx, err := s.db.BeginTx(ctx, nil)
+	if err != nil {
+		return fmt.Errorf("begin pending drops ledger transaction for %d table(s): %w", len(drops), err)
+	}
+	defer rollbackTx(ctx, tx, "record pending drops")
+
+	for _, drop := range drops {
+		state := drop.State
+		if state == "" {
+			state = storage.PendingDropQuarantined
+		}
+		_, err := tx.ExecContext(ctx, query,
+			drop.Target, drop.Environment, drop.DatabaseName, drop.OriginalTable,
+			drop.QuarantinedName, drop.QuarantinedAt.UTC(), drop.RunID, drop.Engine,
+			state, drop.ArrivalTarget, nullJSON(drop.Metadata),
+		)
+		if err != nil {
+			return fmt.Errorf("record pending drop %s.%s as `%s` on target %s/%s: %w",
+				drop.DatabaseName, drop.OriginalTable, drop.QuarantinedName,
+				drop.Target, drop.Environment, err)
+		}
+	}
+
+	if err := tx.Commit(); err != nil {
+		return fmt.Errorf("commit pending drops ledger transaction for %d table(s): %w", len(drops), err)
+	}
+	return nil
+}
+
+func (s *pendingDropStore) LatestForTable(ctx context.Context, target, environment, databaseName, originalTable string) (*storage.PendingDrop, error) {
+	row := s.db.QueryRowContext(ctx, `
+		SELECT `+pendingDropColumns+`
+		FROM pending_drops
+		WHERE target = ? AND environment = ? AND database_name = ? AND original_table = ?
+		ORDER BY quarantined_at DESC, id DESC
+		LIMIT 1
+	`, target, environment, databaseName, originalTable)
+	drop, err := scanPendingDrop(row)
+	if err != nil {
+		return nil, fmt.Errorf("get latest pending drop for %s.%s on target %s/%s: %w",
+			databaseName, originalTable, target, environment, err)
+	}
+	return drop, nil
+}
+
+func (s *pendingDropStore) ListExpired(ctx context.Context, cutoff time.Time, limit int) ([]*storage.PendingDrop, error) {
+	rows, err := s.db.QueryContext(ctx, `
+		SELECT `+pendingDropColumns+`
+		FROM pending_drops
+		WHERE state = ? AND quarantined_at <= ?
+		ORDER BY quarantined_at ASC, id ASC
+		LIMIT ?
+	`, storage.PendingDropQuarantined, cutoff.UTC(), limit)
+	if err != nil {
+		return nil, fmt.Errorf("list pending drops expired before %s: %w", cutoff.UTC().Format(time.RFC3339), err)
+	}
+	drops, err := scanPendingDrops(rows)
+	if err != nil {
+		return nil, fmt.Errorf("list pending drops expired before %s: %w", cutoff.UTC().Format(time.RFC3339), err)
+	}
+	return drops, nil
+}
+
+func (s *pendingDropStore) ListQuarantined(ctx context.Context, target, environment string) ([]*storage.PendingDrop, error) {
+	rows, err := s.db.QueryContext(ctx, `
+		SELECT `+pendingDropColumns+`
+		FROM pending_drops
+		WHERE target = ? AND environment = ? AND state = ?
+		ORDER BY quarantined_at ASC, id ASC
+	`, target, environment, storage.PendingDropQuarantined)
+	if err != nil {
+		return nil, fmt.Errorf("list quarantined pending drops on target %s/%s: %w", target, environment, err)
+	}
+	drops, err := scanPendingDrops(rows)
+	if err != nil {
+		return nil, fmt.Errorf("list quarantined pending drops on target %s/%s: %w", target, environment, err)
+	}
+	return drops, nil
+}
+
+func (s *pendingDropStore) SetState(ctx context.Context, ids []int64, state storage.PendingDropState) error {
+	if len(ids) == 0 {
+		return nil
+	}
+	placeholders, args := int64List(ids)
+	args = append([]any{string(state)}, args...)
+	_, err := s.db.ExecContext(ctx, `
+		UPDATE pending_drops
+		SET state = ?, updated_at = `+s.dialect.CurrentTimestamp(TimestampPrecisionDefault)+`
+		WHERE id IN (`+placeholders+`)
+	`, args...)
+	if err != nil {
+		return fmt.Errorf("set %d pending drop row(s) to state %s: %w", len(ids), state, err)
+	}
+	return nil
+}
+
+func (s *pendingDropStore) Prune(ctx context.Context, cutoff time.Time, limit int) (int64, error) {
+	// Terminal rows only: a quarantined row is still the proof an interrupted
+	// DROP phase converges on, and deleting it would turn a completed change
+	// into a fail-closed error on re-run.
+	//
+	// The bounded victim set is selected through a derived table rather than a
+	// correlated subquery because MySQL refuses to read the delete's own target
+	// table directly, and it is selected at all so one pass cannot lock an
+	// unbounded number of rows.
+	result, err := s.db.ExecContext(ctx, `
+		DELETE FROM pending_drops
+		WHERE id IN (
+			SELECT id FROM (
+				SELECT id FROM pending_drops
+				WHERE state <> ? AND updated_at <= ?
+				ORDER BY updated_at ASC, id ASC
+				LIMIT ?
+			) victims
+		)
+	`, storage.PendingDropQuarantined, cutoff.UTC(), limit)
+	if err != nil {
+		return 0, fmt.Errorf("prune terminal pending drop rows older than %s: %w", cutoff.UTC().Format(time.RFC3339), err)
+	}
+	pruned, err := result.RowsAffected()
+	if err != nil {
+		return 0, fmt.Errorf("count pruned terminal pending drop rows older than %s: %w", cutoff.UTC().Format(time.RFC3339), err)
+	}
+	return pruned, nil
+}
+
+// int64List renders a placeholder list and its arguments for an IN clause.
+func int64List(ids []int64) (string, []any) {
+	args := make([]any, len(ids))
+	for i, id := range ids {
+		args[i] = id
+	}
+	return strings.TrimSuffix(strings.Repeat("?, ", len(ids)), ", "), args
+}
+
+func scanPendingDrops(rows *sql.Rows) ([]*storage.PendingDrop, error) {
+	defer utils.CloseAndLog(rows)
+
+	var drops []*storage.PendingDrop
+	for rows.Next() {
+		drop, err := scanPendingDrop(rows)
+		if err != nil {
+			return nil, err
+		}
+		drops = append(drops, drop)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("iterate pending drop rows: %w", err)
+	}
+	return drops, nil
+}
+
+func scanPendingDrop(s scanner) (*storage.PendingDrop, error) {
+	var drop storage.PendingDrop
+	err := s.Scan(
+		&drop.ID, &drop.Target, &drop.Environment, &drop.DatabaseName, &drop.OriginalTable,
+		&drop.QuarantinedName, &drop.QuarantinedAt, &drop.RunID, &drop.Engine, &drop.State,
+		&drop.ArrivalTarget, &drop.Metadata, &drop.CreatedAt, &drop.UpdatedAt,
+	)
+	if err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return nil, nil
+		}
+		return nil, err
+	}
+	return &drop, nil
+}

pkg/storage/internal/sqlstore/storage.go

@@ -28,6 +28,7 @@ type Storage struct {
 	checks          *checkStore
 	settings        *settingsStore
 	webhookEvents   *webhookEventStore
+	pendingDrops    *pendingDropStore
 }
 
 var _ storage.Storage = (*Storage)(nil)
@@ -86,6 +87,7 @@ func NewWithDependencies(deps Dependencies) *Storage {
 		checks:          &checkStore{db: rdb, dialect: deps.Dialect, classifier: deps.Classifier},
 		settings:        &settingsStore{db: rdb, dialect: deps.Dialect},
 		webhookEvents:   &webhookEventStore{db: rdb, dialect: deps.Dialect, identity: deps.Identity, classifier: deps.Classifier},
+		pendingDrops:    &pendingDropStore{db: rdb, dialect: deps.Dialect},
 	}
 }
 
@@ -149,6 +151,11 @@ func (s *Storage) WebhookEvents() storage.WebhookEventStore {
 	return s.webhookEvents
 }
 
+// PendingDrops returns the pending-drops quarantine ledger store.
+func (s *Storage) PendingDrops() storage.PendingDropStore {
+	return s.pendingDrops
+}
+
 // Ping verifies the database connection is alive.
 func (s *Storage) Ping(ctx context.Context) error {
 	return s.db.PingContext(ctx)

pkg/storage/storage.go

@@ -59,6 +59,9 @@ type Storage interface {
 	// WebhookEvents returns the durable webhook event inbox store.
 	WebhookEvents() WebhookEventStore
 
+	// PendingDrops returns the pending-drops quarantine ledger store.
+	PendingDrops() PendingDropStore
+
 	// Ping verifies the database connection is alive.
 	Ping(ctx context.Context) error
 
@@ -1118,3 +1121,62 @@ type ControlRequestStore interface {
 	// successfully. It reports whether the stored row changed.
 	ClearRemoteFailure(ctx context.Context, applyID int64, operation ControlOperation) (bool, error)
 }
+
+// PendingDropStore records the tables this deployment moved into an engine's
+// pending-drops quarantine, and tracks them until the retention period expires
+// and they are permanently removed.
+//
+// The ledger is a derived index over the targets this deployment executes
+// against, never an authority: the server holding the quarantine schema is the
+// truth, and the reaper re-syncs from it on every visit. Two things depend on
+// the rows. Discovery reads them to learn which servers hold expired
+// quarantines, so cleanup cost scales with drops rather than with the number of
+// registered databases. Re-run convergence reads them as proof that an
+// interrupted DROP phase already executed.
+//
+// Support is expressed by whether rows exist rather than by configuration. A
+// deployment that only dispatches to remote data planes never quarantines, so
+// it writes no rows, so every query here returns nothing and its cleanup pass
+// is a no-op by construction. The same holds for an engine with no quarantine
+// implementation.
+type PendingDropStore interface {
+	// Record inserts ledger rows for tables that are about to be quarantined.
+	// Callers must write before the engine performs the move: an interruption
+	// between the two then orphans a row, which the reaper resolves to a no-op,
+	// rather than a quarantined table no deployment has any record of.
+	//
+	// Rows already present for the same target, environment, and quarantined
+	// name are left untouched, so a retried write and an adoption of a table
+	// this deployment already recorded are both idempotent.
+	Record(ctx context.Context, drops []*PendingDrop) error
+
+	// LatestForTable returns the most recently recorded row for a source table
+	// on a target, or nil when this deployment has no record of quarantining it.
+	//
+	// Callers compare RunID themselves rather than passing it in. A row written
+	// by a different run holds that run's data, and the quarantined names of two
+	// applies dropping the same table differ only by timestamp, so a caller that
+	// matched on run identity inside the query could not tell "no record at all"
+	// apart from "an earlier apply's copy" — two cases that fail closed for
+	// different reasons.
+	LatestForTable(ctx context.Context, target, environment, databaseName, originalTable string) (*PendingDrop, error)
+
+	// ListExpired returns quarantined rows whose quarantine time is at or before
+	// cutoff, oldest first, capped at limit. These are the candidates a cleanup
+	// pass groups by target and sweeps.
+	ListExpired(ctx context.Context, cutoff time.Time, limit int) ([]*PendingDrop, error)
+
+	// ListQuarantined returns every row for a target still in the quarantined
+	// state. The reaper reads it while connected to that target so it can adopt
+	// tables present in the quarantine schema with no matching row.
+	ListQuarantined(ctx context.Context, target, environment string) ([]*PendingDrop, error)
+
+	// SetState drives rows to a terminal state after a sweep has established
+	// what happened to each quarantined table.
+	SetState(ctx context.Context, ids []int64, state PendingDropState) error
+
+	// Prune deletes terminal rows whose state last changed at or before cutoff,
+	// so the ledger does not grow without bound once its rows are useful for
+	// neither discovery nor re-run proof. Returns the number of rows deleted.
+	Prune(ctx context.Context, cutoff time.Time, limit int) (int64, error)
+}