From b597ec7851bd1708406c15b9030259b211bd66fb Mon Sep 17 00:00:00 2001 From: Armand Parajon Date: Mon, 17 Aug 2026 10:55:24 +0800 Subject: [PATCH] feat(storage): add the pending drops quarantine ledger Records the tables a deployment moved into an engine's pending-drops quarantine, in that deployment's own storage. Discovery of servers holding expired quarantines becomes an index lookup rather than a scan, so cleanup cost scales with drops instead of with the number of registered databases, and a deployment that never quarantines writes no rows and no-ops by construction. Rows also carry the run identifier that performed the move, which is what lets an interrupted DROP phase tell its own completed rename apart from drift or from an earlier apply's quarantined copy. No caller yet: the quarantine path and the reaper are wired in follow-ups. --- pkg/api/handlers_test.go | 1 + pkg/schema/mysql/pending_drops.sql | 20 ++ pkg/schema/postgres/pending_drops.sql | 20 ++ .../internal/sqlstore/pending_drops.go | 212 ++++++++++++ pkg/storage/internal/sqlstore/storage.go | 7 + pkg/storage/storage.go | 62 ++++ pkg/storage/storagetest/pending_drops.go | 320 ++++++++++++++++++ pkg/storage/storagetest/storagetest.go | 1 + pkg/storage/types.go | 66 ++++ 9 files changed, 709 insertions(+) create mode 100644 pkg/schema/mysql/pending_drops.sql create mode 100644 pkg/schema/postgres/pending_drops.sql create mode 100644 pkg/storage/internal/sqlstore/pending_drops.go create mode 100644 pkg/storage/storagetest/pending_drops.go 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 +}