diff --git a/pkg/api/handlers_test.go b/pkg/api/handlers_test.go index fa07546ed..203c34a9b 100644 --- a/pkg/api/handlers_test.go +++ b/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 } diff --git a/pkg/schema/mysql/pending_drops.sql b/pkg/schema/mysql/pending_drops.sql new file mode 100644 index 000000000..735dd3435 --- /dev/null +++ b/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 diff --git a/pkg/schema/postgres/pending_drops.sql b/pkg/schema/postgres/pending_drops.sql new file mode 100644 index 000000000..3d99fe00d --- /dev/null +++ b/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); diff --git a/pkg/storage/internal/sqlstore/pending_drops.go b/pkg/storage/internal/sqlstore/pending_drops.go new file mode 100644 index 000000000..54c73680f --- /dev/null +++ b/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 +} diff --git a/pkg/storage/internal/sqlstore/storage.go b/pkg/storage/internal/sqlstore/storage.go index 0689989ce..efad7bb1f 100644 --- a/pkg/storage/internal/sqlstore/storage.go +++ b/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) diff --git a/pkg/storage/storage.go b/pkg/storage/storage.go index 06864410d..bf50c6d11 100644 --- a/pkg/storage/storage.go +++ b/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) +} diff --git a/pkg/storage/storagetest/pending_drops.go b/pkg/storage/storagetest/pending_drops.go new file mode 100644 index 000000000..7a14d303f --- /dev/null +++ b/pkg/storage/storagetest/pending_drops.go @@ -0,0 +1,320 @@ +package storagetest + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/block/schemabot/pkg/storage" +) + +// pendingDropAt builds a quarantined ledger row for one source table on the +// canonical test target, stamped at the given quarantine time. +func pendingDropAt(originalTable, quarantinedName, runID string, quarantinedAt time.Time) *storage.PendingDrop { + return &storage.PendingDrop{ + Target: "shard-a", + Environment: "staging", + DatabaseName: "orders", + OriginalTable: originalTable, + QuarantinedName: quarantinedName, + QuarantinedAt: quarantinedAt, + RunID: runID, + Engine: "spirit", + State: storage.PendingDropQuarantined, + } +} + +// TestPendingDrops runs the behavioral parity suite for +// storage.PendingDropStore. +func TestPendingDrops(t *testing.T, h Harness) { + // The suite pins wall-clock-independent times so ordering and cutoff + // assertions cannot depend on how long a subtest takes to run. + base := time.Date(2026, 8, 1, 12, 0, 0, 0, time.UTC) + + t.Run("RecordAndListQuarantined", func(t *testing.T) { + ctx := t.Context() + store := h.NewStorage(t) + + // Written out of quarantine order so the ordering assertion cannot pass + // by insertion order alone. + require.NoError(t, store.PendingDrops().Record(ctx, []*storage.PendingDrop{ + pendingDropAt("shipments", "20260801120500000_shipments", "run-1", base.Add(5*time.Minute)), + pendingDropAt("invoices", "20260801120000000_invoices", "run-1", base), + })) + + drops, err := store.PendingDrops().ListQuarantined(ctx, "shard-a", "staging") + require.NoError(t, err) + require.Len(t, drops, 2) + require.Equal(t, "invoices", drops[0].OriginalTable) + require.Equal(t, "shipments", drops[1].OriginalTable) + + oldest := drops[0] + require.Equal(t, "shard-a", oldest.Target) + require.Equal(t, "staging", oldest.Environment) + require.Equal(t, "orders", oldest.DatabaseName) + require.Equal(t, "20260801120000000_invoices", oldest.QuarantinedName) + require.Equal(t, "run-1", oldest.RunID) + require.Equal(t, "spirit", oldest.Engine) + require.Equal(t, storage.PendingDropQuarantined, oldest.State) + require.WithinDuration(t, base, oldest.QuarantinedAt.UTC(), time.Second) + }) + + t.Run("RecordIsIdempotentPerQuarantinedName", func(t *testing.T) { + ctx := t.Context() + store := h.NewStorage(t) + + drop := pendingDropAt("invoices", "20260801120000000_invoices", "run-1", base) + require.NoError(t, store.PendingDrops().Record(ctx, []*storage.PendingDrop{drop})) + + // A retried write of the same quarantined name must neither fail nor + // duplicate the row: the reaper would otherwise try to drop one table + // twice and count the second attempt as an error. + require.NoError(t, store.PendingDrops().Record(ctx, []*storage.PendingDrop{drop})) + + drops, err := store.PendingDrops().ListQuarantined(ctx, "shard-a", "staging") + require.NoError(t, err) + require.Len(t, drops, 1) + }) + + t.Run("RecordEmptyIsNoOp", func(t *testing.T) { + ctx := t.Context() + store := h.NewStorage(t) + + require.NoError(t, store.PendingDrops().Record(ctx, nil)) + + drops, err := store.PendingDrops().ListQuarantined(ctx, "shard-a", "staging") + require.NoError(t, err) + require.Empty(t, drops) + }) + + t.Run("ListQuarantinedIsScopedToTargetAndEnvironment", func(t *testing.T) { + ctx := t.Context() + store := h.NewStorage(t) + + otherEnv := pendingDropAt("invoices", "20260801120000000_invoices", "run-1", base) + otherEnv.Environment = "production" + otherTarget := pendingDropAt("invoices", "20260801120000000_invoices", "run-1", base) + otherTarget.Target = "shard-b" + + require.NoError(t, store.PendingDrops().Record(ctx, []*storage.PendingDrop{ + pendingDropAt("invoices", "20260801120000000_invoices", "run-1", base), + otherEnv, + otherTarget, + })) + + // The same quarantined name on a different target or environment is a + // different server's table, so the unique key admits all three rows and + // a sweep of one target sees only its own. + drops, err := store.PendingDrops().ListQuarantined(ctx, "shard-a", "staging") + require.NoError(t, err) + require.Len(t, drops, 1) + require.Equal(t, "shard-a", drops[0].Target) + require.Equal(t, "staging", drops[0].Environment) + }) + + t.Run("LatestForTableReturnsNewestRecord", func(t *testing.T) { + ctx := t.Context() + store := h.NewStorage(t) + + require.NoError(t, store.PendingDrops().Record(ctx, []*storage.PendingDrop{ + pendingDropAt("invoices", "20260801120000000_invoices", "run-1", base), + pendingDropAt("invoices", "20260801130000000_invoices", "run-2", base.Add(time.Hour)), + })) + + // Two applies dropping the same table leave rows whose quarantined names + // differ only by timestamp. The caller compares RunID to decide whether + // the newest one is its own proof or an earlier apply's data. + latest, err := store.PendingDrops().LatestForTable(ctx, "shard-a", "staging", "orders", "invoices") + require.NoError(t, err) + require.NotNil(t, latest) + require.Equal(t, "run-2", latest.RunID) + require.Equal(t, "20260801130000000_invoices", latest.QuarantinedName) + }) + + t.Run("LatestForTableReturnsNilWhenUnrecorded", func(t *testing.T) { + ctx := t.Context() + store := h.NewStorage(t) + + require.NoError(t, store.PendingDrops().Record(ctx, []*storage.PendingDrop{ + pendingDropAt("invoices", "20260801120000000_invoices", "run-1", base), + })) + + // A nil result is what tells a re-run that the missing source table is + // drift from outside SchemaBot rather than its own completed move, so it + // must not be confused with an error. + latest, err := store.PendingDrops().LatestForTable(ctx, "shard-a", "staging", "orders", "shipments") + require.NoError(t, err) + require.Nil(t, latest) + }) + + t.Run("ListExpiredReturnsOnlyElapsedQuarantines", func(t *testing.T) { + ctx := t.Context() + store := h.NewStorage(t) + + require.NoError(t, store.PendingDrops().Record(ctx, []*storage.PendingDrop{ + pendingDropAt("invoices", "20260801120000000_invoices", "run-1", base), + pendingDropAt("shipments", "20260808120000000_shipments", "run-2", base.Add(7*24*time.Hour)), + })) + + expired, err := store.PendingDrops().ListExpired(ctx, base.Add(24*time.Hour), 10) + require.NoError(t, err) + require.Len(t, expired, 1) + require.Equal(t, "invoices", expired[0].OriginalTable) + }) + + t.Run("ListExpiredIsOldestFirstAndBounded", func(t *testing.T) { + ctx := t.Context() + store := h.NewStorage(t) + + require.NoError(t, store.PendingDrops().Record(ctx, []*storage.PendingDrop{ + pendingDropAt("c", "20260801120200000_c", "run-1", base.Add(2*time.Minute)), + pendingDropAt("a", "20260801120000000_a", "run-1", base), + pendingDropAt("b", "20260801120100000_b", "run-1", base.Add(time.Minute)), + })) + + expired, err := store.PendingDrops().ListExpired(ctx, base.Add(time.Hour), 2) + require.NoError(t, err) + require.Len(t, expired, 2) + require.Equal(t, "a", expired[0].OriginalTable) + require.Equal(t, "b", expired[1].OriginalTable) + }) + + t.Run("SetStateRemovesRowsFromDiscovery", func(t *testing.T) { + ctx := t.Context() + store := h.NewStorage(t) + + require.NoError(t, store.PendingDrops().Record(ctx, []*storage.PendingDrop{ + pendingDropAt("invoices", "20260801120000000_invoices", "run-1", base), + pendingDropAt("shipments", "20260801120100000_shipments", "run-1", base.Add(time.Minute)), + })) + + expired, err := store.PendingDrops().ListExpired(ctx, base.Add(time.Hour), 10) + require.NoError(t, err) + require.Len(t, expired, 2) + + require.NoError(t, store.PendingDrops().SetState(ctx, []int64{expired[0].ID}, storage.PendingDropReaped)) + require.NoError(t, store.PendingDrops().SetState(ctx, []int64{expired[1].ID}, storage.PendingDropVanished)) + + // A terminal row is no longer a sweep candidate, but it is still the + // proof a re-run reads, so it stays queryable by table. + remaining, err := store.PendingDrops().ListExpired(ctx, base.Add(time.Hour), 10) + require.NoError(t, err) + require.Empty(t, remaining) + + latest, err := store.PendingDrops().LatestForTable(ctx, "shard-a", "staging", "orders", "invoices") + require.NoError(t, err) + require.NotNil(t, latest) + require.Equal(t, storage.PendingDropReaped, latest.State) + }) + + t.Run("SetStateEmptyIsNoOp", func(t *testing.T) { + ctx := t.Context() + store := h.NewStorage(t) + + require.NoError(t, store.PendingDrops().SetState(ctx, nil, storage.PendingDropReaped)) + }) + + t.Run("PruneRemovesOnlyTerminalRows", func(t *testing.T) { + ctx := t.Context() + store := h.NewStorage(t) + + require.NoError(t, store.PendingDrops().Record(ctx, []*storage.PendingDrop{ + pendingDropAt("invoices", "20260801120000000_invoices", "run-1", base), + pendingDropAt("shipments", "20260801120100000_shipments", "run-1", base.Add(time.Minute)), + })) + + expired, err := store.PendingDrops().ListExpired(ctx, base.Add(time.Hour), 10) + require.NoError(t, err) + require.Len(t, expired, 2) + require.NoError(t, store.PendingDrops().SetState(ctx, []int64{expired[0].ID}, storage.PendingDropReaped)) + + // SetState stamps updated_at with the server's clock, so the cutoff has + // to be in the future rather than derived from the pinned base time. + pruned, err := store.PendingDrops().Prune(ctx, time.Now().Add(time.Hour), 10) + require.NoError(t, err) + require.Equal(t, int64(1), pruned) + + // Pruning must never reach a still-quarantined row: it describes a table + // that is still sitting on the target, and dropping the row would strand + // it outside discovery. + remaining, err := store.PendingDrops().ListQuarantined(ctx, "shard-a", "staging") + require.NoError(t, err) + require.Len(t, remaining, 1) + require.Equal(t, "shipments", remaining[0].OriginalTable) + }) + + t.Run("PruneIsBounded", func(t *testing.T) { + ctx := t.Context() + store := h.NewStorage(t) + + require.NoError(t, store.PendingDrops().Record(ctx, []*storage.PendingDrop{ + pendingDropAt("a", "20260801120000000_a", "run-1", base), + pendingDropAt("b", "20260801120100000_b", "run-1", base.Add(time.Minute)), + pendingDropAt("c", "20260801120200000_c", "run-1", base.Add(2*time.Minute)), + })) + + expired, err := store.PendingDrops().ListExpired(ctx, base.Add(time.Hour), 10) + require.NoError(t, err) + require.Len(t, expired, 3) + ids := []int64{expired[0].ID, expired[1].ID, expired[2].ID} + require.NoError(t, store.PendingDrops().SetState(ctx, ids, storage.PendingDropReaped)) + + pruned, err := store.PendingDrops().Prune(ctx, time.Now().Add(time.Hour), 2) + require.NoError(t, err) + require.Equal(t, int64(2), pruned) + + pruned, err = store.PendingDrops().Prune(ctx, time.Now().Add(time.Hour), 2) + require.NoError(t, err) + require.Equal(t, int64(1), pruned) + }) + + t.Run("AdoptedRowsCarryArrivalTargetWithoutOrigin", func(t *testing.T) { + ctx := t.Context() + store := h.NewStorage(t) + + // A sweep that finds an unrecorded table in the quarantine schema cannot + // recover its origin from the name, so it records where it was found and + // leaves attribution empty rather than guessing. + require.NoError(t, store.PendingDrops().Record(ctx, []*storage.PendingDrop{{ + Target: "shard-a", + Environment: "staging", + QuarantinedName: "20260724152851280_days", + QuarantinedAt: base, + Engine: "spirit", + State: storage.PendingDropQuarantined, + ArrivalTarget: "shard-a", + }})) + + drops, err := store.PendingDrops().ListQuarantined(ctx, "shard-a", "staging") + require.NoError(t, err) + require.Len(t, drops, 1) + require.Empty(t, drops[0].DatabaseName) + require.Empty(t, drops[0].OriginalTable) + require.Empty(t, drops[0].RunID) + require.Equal(t, "shard-a", drops[0].ArrivalTarget) + }) + + t.Run("SurfacesConnectionFailures", func(t *testing.T) { + ctx := t.Context() + store := h.NewUnreachableStorage(t) + + require.Error(t, store.PendingDrops().Record(ctx, []*storage.PendingDrop{ + pendingDropAt("invoices", "20260801120000000_invoices", "run-1", base), + })) + + _, err := store.PendingDrops().LatestForTable(ctx, "shard-a", "staging", "orders", "invoices") + require.Error(t, err) + + _, err = store.PendingDrops().ListExpired(ctx, base, 10) + require.Error(t, err) + + _, err = store.PendingDrops().ListQuarantined(ctx, "shard-a", "staging") + require.Error(t, err) + + require.Error(t, store.PendingDrops().SetState(ctx, []int64{1}, storage.PendingDropReaped)) + + _, err = store.PendingDrops().Prune(ctx, base, 10) + require.Error(t, err) + }) +} diff --git a/pkg/storage/storagetest/storagetest.go b/pkg/storage/storagetest/storagetest.go index 5f281e782..50df31614 100644 --- a/pkg/storage/storagetest/storagetest.go +++ b/pkg/storage/storagetest/storagetest.go @@ -61,6 +61,7 @@ type Harness interface { func Run(t *testing.T, h Harness) { t.Run("Settings", func(t *testing.T) { TestSettings(t, h) }) t.Run("ApplyLogs", func(t *testing.T) { TestApplyLogs(t, h) }) + t.Run("PendingDrops", func(t *testing.T) { TestPendingDrops(t, h) }) } // Fixture helpers. These build the canonical Lock/Apply rows used by the diff --git a/pkg/storage/types.go b/pkg/storage/types.go index f32317378..478ccdade 100644 --- a/pkg/storage/types.go +++ b/pkg/storage/types.go @@ -1600,3 +1600,69 @@ type WebhookEvent struct { CreatedAt time.Time UpdatedAt time.Time } + +// PendingDropState is the lifecycle state of a pending-drops ledger row. +type PendingDropState string + +const ( + // PendingDropQuarantined means the table is expected to be sitting in the + // engine's quarantine schema, recoverable until retention expires. + PendingDropQuarantined PendingDropState = "quarantined" + // PendingDropReaped means a sweep permanently removed the quarantined table. + PendingDropReaped PendingDropState = "reaped" + // PendingDropVanished means a sweep reached the target and found no such + // table. The usual cause is a crash between recording the row and performing + // the move, so the row describes a table that was never created; an operator + // recovering the table by hand produces the same result. + PendingDropVanished PendingDropState = "vanished" +) + +// PendingDrop is one row of the pending-drops quarantine ledger: a table this +// deployment moved into an engine's quarantine schema, or one it adopted after +// finding it there unrecorded. +// +// Target and Environment together are the sweep key, the pair the deployment +// re-resolves to an endpoint on every cleanup pass. DatabaseName and +// OriginalTable are provenance, and both are recorded because neither survives +// the move: the quarantined name discards the origin schema entirely and +// truncates the source table name to fit the server's identifier limit, so +// origin exists only at quarantine time. +type PendingDrop struct { + ID int64 + Target string + Environment string + + // DatabaseName is the origin database. It is empty on adopted rows, where + // the quarantined name carries no origin and attributing one would be a + // guess. + DatabaseName string + // OriginalTable is the full source table name before truncation. It is empty + // on adopted rows for the same reason as DatabaseName. + OriginalTable string + + // QuarantinedName is the table's name inside the engine's quarantine schema. + QuarantinedName string + QuarantinedAt time.Time + + // RunID is the durable run identifier of the apply that performed the + // quarantine, and is what makes the row usable as proof that an interrupted + // DROP phase already executed. It is empty on adopted rows. + RunID string + + // Engine is the engine that wrote the row. Quarantine is a per-engine + // behavior, so the reaper dispatches its sweep on this value. + Engine string + + State PendingDropState + + // ArrivalTarget records which target a sweep was visiting when it adopted an + // unrecorded table, so unattributed rows still say where they were found. + ArrivalTarget string + + // Metadata holds engine-specific data as JSON, keeping engine semantics out + // of the shared row. + Metadata []byte + + CreatedAt time.Time + UpdatedAt time.Time +}