diff --git a/server/cmd/maintenance/main.go b/server/cmd/maintenance/main.go index ee14706b..1d13e64c 100644 --- a/server/cmd/maintenance/main.go +++ b/server/cmd/maintenance/main.go @@ -26,6 +26,7 @@ func main() { stalledAllocationBatch := flag.Int("stalled-allocation-batch", 100, "maximum stalled matches reclaimed per pass") initialConnectBatch := flag.Int("initial-connect-batch", 100, "maximum pre-live matches evaluated per pass") liveAbandonmentBatch := flag.Int("live-abandonment-batch", 100, "maximum live ranked matches evaluated for expired reconnect leases per pass") + retentionBatch := flag.Int("retention-batch", 500, "maximum rows deleted per table per retention pass") flag.Parse() if *dsn == "" { fatalf("--dsn or COSMIC_CLASH_POSTGRES_DSN is required") @@ -69,6 +70,32 @@ func main() { if reclaimed > 0 { log.Printf("reclaimed %d stalled allocations, requeuing their participants", reclaimed) } + // Retention. Without this, idempotency keys alone grow by roughly one + // row per queued player per heartbeat interval, forever. + purged, err := store.PurgeExpiredRecords(ctx, db, now, *retentionBatch) + if err != nil { + fatalf("retention maintenance: %v", err) + } + if purged.Total() > 0 { + log.Printf("purged %d expired records (idempotency=%d outbox=%d dead-lettered=%d sessions=%d)", + purged.Total(), purged.IdempotencyKeys, purged.PublishedOutbox, purged.DeadLetteredOutbox, purged.ExpiredSessions) + } + // Deletion lag: a backlog that keeps climbing means the interval or + // batch size is too small for current volume. + backlog, err := store.RetentionBacklog(ctx, db, now) + if err != nil { + fatalf("retention backlog: %v", err) + } + if backlog > 0 { + log.Printf("retention backlog is %d rows past their window", backlog) + } + deadLettered, err := store.CountDeadLetteredOutboxEvents(ctx, db) + if err != nil { + fatalf("dead-letter count: %v", err) + } + if deadLettered > 0 { + log.Printf("WARNING: %d outbox events were never delivered and are dead-lettered", deadLettered) + } } runInitialConnect := func(now time.Time) { reconciled, err := store.ReconcileInitialConnect(ctx, db, now, *initialConnectBatch) diff --git a/server/migrations/0015_retention_indexes.sql b/server/migrations/0015_retention_indexes.sql new file mode 100644 index 00000000..5e949535 --- /dev/null +++ b/server/migrations/0015_retention_indexes.sql @@ -0,0 +1,24 @@ +-- Retention support. Three tables grow without bound today: +-- +-- idempotency_keys -- the client heartbeats every 10s and mints a fresh key +-- each time, so at 10,000 queued players this alone adds roughly 60,000 +-- rows per minute, forever. +-- outbox -- published rows are never purged. +-- sessions -- expired and revoked rows are never purged. +-- +-- The maintenance role performed lifecycle reconciliation only, so storage, +-- index size, vacuum pressure, backup size and recovery time all grew without +-- limit on a service meant to scale horizontally. +-- +-- These indexes exist to make the deletion predicates cheap; without them each +-- purge pass would sequentially scan the very tables it is trying to bound. + +CREATE INDEX IF NOT EXISTS idempotency_keys_created_at + ON idempotency_keys (created_at); + +CREATE INDEX IF NOT EXISTS outbox_published_at + ON outbox (published_at) + WHERE published_at IS NOT NULL; + +CREATE INDEX IF NOT EXISTS sessions_expires_at + ON sessions (expires_at); diff --git a/server/migrations/down/0015_retention_indexes.sql b/server/migrations/down/0015_retention_indexes.sql new file mode 100644 index 00000000..b16546ba --- /dev/null +++ b/server/migrations/down/0015_retention_indexes.sql @@ -0,0 +1,3 @@ +DROP INDEX IF EXISTS idempotency_keys_created_at; +DROP INDEX IF EXISTS outbox_published_at; +DROP INDEX IF EXISTS sessions_expires_at; diff --git a/server/store/postgres_integration_test.go b/server/store/postgres_integration_test.go index 00b8b471..cca3d90c 100644 --- a/server/store/postgres_integration_test.go +++ b/server/store/postgres_integration_test.go @@ -11,6 +11,7 @@ import ( "fmt" "os" "path/filepath" + "strconv" "strings" "sync" "testing" @@ -1792,11 +1793,13 @@ func TestPostgreSQLMigrationsRollBackAndReapplyCleanly(t *testing.T) { // Roll back every migration one at a time, in reverse, checking each // down file actually undoes what its forward file created — not just // that Rollback returns nil. - // This count is the number of migrations above 0006, so it must grow with - // every new migration; otherwise the later fixed-count rollbacks below - // silently target the wrong files. - if err := migrations.Rollback(context.Background(), db, dir, 8); err != nil { - t.Fatalf("rollback 0014 through 0007: %v", err) + // Derived, not hardcoded: every added migration shifts this count, and a + // stale literal silently makes the fixed-count rollbacks below target the + // wrong files (the failure then surfaces as a confusing "0006 rollback did + // not drop matches.allocation_id"). + aboveMigration0006 := countMigrationsAbove(t, dir, 6) + if err := migrations.Rollback(context.Background(), db, dir, aboveMigration0006); err != nil { + t.Fatalf("rollback everything above 0006: %v", err) } var hasInitialConnectReadyColumn bool if err := db.QueryRow(`SELECT count(*) > 0 FROM information_schema.columns WHERE table_name = 'matches' AND column_name = 'initial_connect_ready_at'`).Scan(&hasInitialConnectReadyColumn); err != nil { @@ -2074,3 +2077,110 @@ func TestNotifyControlPlaneEventRejectsOversizedPayloads(t *testing.T) { t.Fatal("payload over the NOTIFY limit was accepted") } } + +// Each 10s queue heartbeat mints a fresh idempotency key and permanently +// inserts a row; published outbox rows and expired sessions were never purged +// either. At 10,000 queued players heartbeats alone add roughly 60,000 durable +// rows per minute, so table and index growth, vacuum pressure, backup size and +// recovery time were all unbounded on a horizontally-scaled service. +func TestPostgreSQLRetentionPurgesExpiredRecordsInBoundedBatches(t *testing.T) { + db := openIntegrationPostgres(t) + applyIntegrationMigrations(t, db) + + ctx := context.Background() + now := time.Now().UTC().Truncate(time.Microsecond) + stale := now.Add(-IdempotencyKeyRetention - time.Hour) + fresh := now.Add(-time.Minute) + + // 120 stale keys (purgeable) and 5 fresh ones (must survive: a client may + // still retry those mutations). + for i := 0; i < 120; i++ { + if _, err := db.ExecContext(ctx, `INSERT INTO idempotency_keys (scope, idempotency_key, payload_digest, result, created_at) VALUES ('queue.mutate', $1, '\x00', '{}'::jsonb, $2)`, + fmt.Sprintf("stale-key-%030d", i), stale); err != nil { + t.Fatal(err) + } + } + for i := 0; i < 5; i++ { + if _, err := db.ExecContext(ctx, `INSERT INTO idempotency_keys (scope, idempotency_key, payload_digest, result, created_at) VALUES ('queue.mutate', $1, '\x00', '{}'::jsonb, $2)`, + fmt.Sprintf("fresh-key-%030d", i), fresh); err != nil { + t.Fatal(err) + } + } + + countKeys := func() int { + var count int + if err := db.QueryRowContext(ctx, `SELECT count(*) FROM idempotency_keys`).Scan(&count); err != nil { + t.Fatal(err) + } + return count + } + if countKeys() != 125 { + t.Fatalf("seed failed: %d keys", countKeys()) + } + + // One pass must delete at most the batch size, not the whole backlog: a + // purge that took an unbounded lock would stall live traffic. + report, err := PurgeExpiredRecords(ctx, db, now, 50) + if err != nil { + t.Fatalf("purge: %v", err) + } + if report.IdempotencyKeys != 50 { + t.Fatalf("first pass deleted %d keys, want the 50-row batch bound", report.IdempotencyKeys) + } + if countKeys() != 75 { + t.Fatalf("after one bounded pass: %d keys", countKeys()) + } + + // Repeated passes converge on exactly the fresh rows and then stop. + for i := 0; i < 5; i++ { + if _, err := PurgeExpiredRecords(ctx, db, now, 50); err != nil { + t.Fatalf("purge pass %d: %v", i, err) + } + } + if countKeys() != 5 { + t.Fatalf("steady state left %d keys, want only the 5 fresh ones", countKeys()) + } + final, err := PurgeExpiredRecords(ctx, db, now, 50) + if err != nil { + t.Fatalf("final purge: %v", err) + } + if final.Total() != 0 { + t.Fatalf("purge deleted %d rows that were still within their retention window", final.Total()) + } + backlog, err := RetentionBacklog(ctx, db, now) + if err != nil { + t.Fatalf("backlog: %v", err) + } + if backlog != 0 { + t.Fatalf("deletion lag is %d after draining the backlog", backlog) + } +} + +// countMigrationsAbove reports how many forward migrations have a number +// greater than the given one, so rollback step counts in this file track new +// migrations automatically instead of needing a manual bump. +func countMigrationsAbove(t *testing.T, dir string, number int) int { + t.Helper() + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatalf("read migrations: %v", err) + } + count := 0 + for _, entry := range entries { + name := entry.Name() + if entry.IsDir() || !strings.HasSuffix(name, ".sql") || len(name) < 4 { + continue + } + index, err := strconv.Atoi(name[:4]) + if err != nil { + continue + } + if index > number { + count++ + } + } + if count == 0 { + t.Fatalf("no migrations found above %04d in %s", number, dir) + } + return count +} diff --git a/server/store/retention.go b/server/store/retention.go new file mode 100644 index 00000000..a7f62173 --- /dev/null +++ b/server/store/retention.go @@ -0,0 +1,133 @@ +package store + +import ( + "context" + "database/sql" + "fmt" + "time" +) + +// Retention windows. Each must comfortably exceed every retry and recovery +// horizon that could still consult the row, because deleting one early changes +// behaviour rather than merely reclaiming space: +// +// - An idempotency key must outlive any client retry of the same mutation; +// deleting it early turns a replay into a second real mutation. The client +// heartbeats every 10s and abandons a ticket far sooner than this. +// - A published outbox row is only kept for operator forensics; delivery has +// already happened, and the dispatcher never re-reads a published row. +// - A session row past expiry can no longer authenticate, so retaining it +// buys nothing beyond a short audit tail. +const ( + IdempotencyKeyRetention = 24 * time.Hour + PublishedOutboxRetention = 72 * time.Hour + ExpiredSessionRetention = 24 * time.Hour + // DeadLetteredOutboxRetention is deliberately the longest: those rows are + // the record of events that were never delivered, and an operator needs + // time to notice and investigate them. + DeadLetteredOutboxRetention = 30 * 24 * time.Hour +) + +// Deletes are batched and use SKIP LOCKED so a purge never blocks live +// traffic, never holds a long transaction, and multiple maintenance replicas +// can run concurrently without contending on the same rows. +const ( + PurgeIdempotencyKeysSQL = `DELETE FROM idempotency_keys +WHERE (scope, idempotency_key) IN ( + SELECT scope, idempotency_key FROM idempotency_keys + WHERE created_at < $1 + ORDER BY created_at + LIMIT $2 + FOR UPDATE SKIP LOCKED +)` + PurgePublishedOutboxSQL = `DELETE FROM outbox +WHERE event_id IN ( + SELECT event_id FROM outbox + WHERE published_at IS NOT NULL AND published_at < $1 + ORDER BY published_at + LIMIT $2 + FOR UPDATE SKIP LOCKED +)` + PurgeDeadLetteredOutboxSQL = `DELETE FROM outbox +WHERE event_id IN ( + SELECT event_id FROM outbox + WHERE dead_lettered_at IS NOT NULL AND dead_lettered_at < $1 + ORDER BY dead_lettered_at + LIMIT $2 + FOR UPDATE SKIP LOCKED +)` + PurgeExpiredSessionsSQL = `DELETE FROM sessions +WHERE session_id IN ( + SELECT session_id FROM sessions + WHERE expires_at < $1 + ORDER BY expires_at + LIMIT $2 + FOR UPDATE SKIP LOCKED +)` +) + +// RetentionReport is the per-pass result. Callers expose these as metrics so +// deletion lag is observable: if a count stays pinned at the batch size, the +// purge is not keeping up with insert volume. +type RetentionReport struct { + IdempotencyKeys int64 + PublishedOutbox int64 + DeadLetteredOutbox int64 + ExpiredSessions int64 +} + +func (r RetentionReport) Total() int64 { + return r.IdempotencyKeys + r.PublishedOutbox + r.DeadLetteredOutbox + r.ExpiredSessions +} + +// PurgeExpiredRecords removes one bounded batch from each retained table. It +// returns partial progress alongside an error so a failure in one table does +// not hide the work already done in another. +func PurgeExpiredRecords(ctx context.Context, db *sql.DB, now time.Time, batch int) (RetentionReport, error) { + var report RetentionReport + if db == nil || now.IsZero() || batch < 1 || batch > 10000 { + return report, fmt.Errorf("invalid retention arguments") + } + steps := []struct { + query string + cutoff time.Time + into *int64 + }{ + {PurgeIdempotencyKeysSQL, now.Add(-IdempotencyKeyRetention), &report.IdempotencyKeys}, + {PurgePublishedOutboxSQL, now.Add(-PublishedOutboxRetention), &report.PublishedOutbox}, + {PurgeDeadLetteredOutboxSQL, now.Add(-DeadLetteredOutboxRetention), &report.DeadLetteredOutbox}, + {PurgeExpiredSessionsSQL, now.Add(-ExpiredSessionRetention), &report.ExpiredSessions}, + } + for _, step := range steps { + result, err := db.ExecContext(ctx, step.query, step.cutoff, batch) + if err != nil { + return report, err + } + deleted, err := result.RowsAffected() + if err != nil { + return report, err + } + *step.into = deleted + } + return report, nil +} + +// RetentionBacklog counts rows already past their retention window. This is +// the deletion-lag metric: a number that keeps climbing means the purge +// interval or batch size is too small for current volume. +func RetentionBacklog(ctx context.Context, db *sql.DB, now time.Time) (int64, error) { + if db == nil || now.IsZero() { + return 0, fmt.Errorf("invalid retention backlog arguments") + } + const query = `SELECT + (SELECT count(*) FROM idempotency_keys WHERE created_at < $1) + + (SELECT count(*) FROM outbox WHERE published_at IS NOT NULL AND published_at < $2) + + (SELECT count(*) FROM sessions WHERE expires_at < $3)` + var backlog int64 + err := db.QueryRowContext(ctx, query, + now.Add(-IdempotencyKeyRetention), + now.Add(-PublishedOutboxRetention), + now.Add(-ExpiredSessionRetention), + ).Scan(&backlog) + return backlog, err +}