feat(server): add retention for idempotency, outbox and session records

Each ten-second queue heartbeat mints a fresh idempotency key and
permanently inserts a row. Published outbox rows and expired/revoked
sessions were never purged either -- the maintenance role performed
lifecycle reconciliation only. 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
service intended to scale horizontally.

Add retention windows chosen to exceed every retry and recovery horizon
that could still consult the row -- deleting an idempotency key early
would turn a client replay into a second real mutation, so this is a
correctness bound, not just a housekeeping one. Dead-lettered outbox
rows are kept longest, being the record of events never delivered.

Deletes run in bounded SKIP LOCKED batches so a purge never blocks live
traffic, never holds a long transaction, and concurrent maintenance
replicas do not contend. Indexes back each predicate so a pass cannot
degrade into a sequential scan of the table it is bounding. The
maintenance role reports rows purged, the backlog past its window
(deletion lag), and any dead-lettered events.

Also make the migration-rollback test derive its step counts instead of
hardcoding them: adding a migration silently shifted the fixed counts so
the failure surfaced as an unrelated "0006 rollback did not drop
matches.allocation_id".

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)

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);

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;

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
+}

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
+}