diff --git a/internal/api/investigation.go b/internal/api/investigation.go
index cd23d237..dcde2ac7 100644
--- a/internal/api/investigation.go
+++ b/internal/api/investigation.go
@@ -2,6 +2,7 @@ package api
 
 import (
 	"context"
+	"fmt"
 	"net/http"
 	"time"
 
@@ -51,16 +52,37 @@ type environmentQuerier interface {
 	Query(context.Context, string, ...any) (pgx.Rows, error)
 }
 
+// environmentsQuery walks each table's (project_id, environment) partial index
+// one distinct value at a time per project, so a refresh costs one index
+// descent per environment instead of reading every indexed row. Projects come
+// from the projects table so the walk never depends on skip scan, and unknown
+// ids simply match nothing. The environment predicates repeat the partial index
+// predicate so the planner can use it.
+var environmentsQuery = `WITH RECURSIVE scope AS MATERIALIZED (
+ SELECT id FROM projects WHERE cardinality($1::uuid[]) = 0 OR id = ANY($1::uuid[])
+)` + environmentWalk("events") + environmentWalk("transactions") + environmentWalk("logs") + `
+SELECT environment FROM events_environments WHERE environment IS NOT NULL
+UNION SELECT environment FROM transactions_environments WHERE environment IS NOT NULL
+UNION SELECT environment FROM logs_environments WHERE environment IS NOT NULL
+ORDER BY environment`
+
+// environmentWalk yields every environment of table per scoped project, ending
+// each project's walk with a NULL row once no larger value remains.
+func environmentWalk(table string) string {
+	next := `(SELECT environment FROM ` + table + ` WHERE project_id = %s AND environment IS NOT NULL AND environment <> ''%s ORDER BY environment LIMIT 1)`
+	return `, ` + table + `_environments(project_id, environment) AS (
+ SELECT s.id, ` + fmt.Sprintf(next, "s.id", "") + ` FROM scope s
+ UNION ALL SELECT w.project_id, ` + fmt.Sprintf(next, "w.project_id", " AND environment > w.environment") + `
+ FROM ` + table + `_environments w WHERE w.environment IS NOT NULL
+)`
+}
+
 func serveEnvironments(w http.ResponseWriter, r *http.Request, db environmentQuerier) {
 	ids := bearerProjectIDs(r, r.URL.Query()["project_id"])
 	if ids == nil {
 		ids = []string{}
 	}
-	rows, err := db.Query(r.Context(), `SELECT DISTINCT environment FROM (
- SELECT environment FROM events WHERE (cardinality($1::uuid[]) = 0 OR project_id = ANY($1::uuid[]))
- UNION SELECT environment FROM transactions WHERE (cardinality($1::uuid[]) = 0 OR project_id = ANY($1::uuid[]))
- UNION SELECT environment FROM logs WHERE (cardinality($1::uuid[]) = 0 OR project_id = ANY($1::uuid[]))
- ) environments WHERE environment IS NOT NULL AND environment <> '' ORDER BY environment`, ids)
+	rows, err := db.Query(r.Context(), environmentsQuery, ids)
 	if err != nil {
 		http.Error(w, "Could not load environments", 500)
 		return
diff --git a/internal/api/investigation_test.go b/internal/api/investigation_test.go
index dfd41efb..34b0e759 100644
--- a/internal/api/investigation_test.go
+++ b/internal/api/investigation_test.go
@@ -1,11 +1,16 @@
 package api_test
 
 import (
+	"context"
+	"encoding/json"
 	"net/http"
 	"net/http/httptest"
 	"testing"
 
+	"github.com/jackc/pgx/v5"
 	"github.com/stretchr/testify/require"
+
+	"github.com/blendbyte/tindra/internal/storage"
 )
 
 func TestInvestigationValidatesBounds(t *testing.T) {
@@ -43,3 +48,98 @@ func TestNinetyDayInvestigationEndpoints(t *testing.T) {
 		})
 	}
 }
+
+func TestEnvironmentsAcrossTablesAndProjects(t *testing.T) {
+	ctx := context.Background()
+	var ids []string
+	for _, slug := range []string{"env-walk-a", "env-walk-b", "env-walk-c"} {
+		p, err := storage.CreateProject(ctx, testPool, slug, slug)
+		require.NoError(t, err)
+		t.Cleanup(func() { testPool.Exec(ctx, "DELETE FROM projects WHERE id=$1", p.ID) })
+		ids = append(ids, p.ID)
+	}
+	a, b, c := ids[0], ids[1], ids[2]
+	event := func(project string, env any) {
+		t.Helper()
+		_, err := testPool.Exec(ctx, `INSERT INTO events(project_id,timestamp,payload) VALUES ($1,NOW(),jsonb_build_object('environment',$2::text))`, project, env)
+		require.NoError(t, err)
+	}
+	transaction := func(project string, env any) {
+		t.Helper()
+		_, err := testPool.Exec(ctx, `INSERT INTO transactions(project_id,transaction,duration_ms,start_timestamp,timestamp,environment) VALUES ($1,'/env',1,NOW(),NOW(),$2)`, project, env)
+		require.NoError(t, err)
+	}
+	logLine := func(project string, env any) {
+		t.Helper()
+		_, err := testPool.Exec(ctx, `INSERT INTO logs(project_id,timestamp,body,environment) VALUES ($1,NOW(),'env',$2)`, project, env)
+		require.NoError(t, err)
+	}
+	for range 3 {
+		event(a, "envwalk-prod")
+		transaction(a, "envwalk-prod")
+		logLine(a, "envwalk-prod")
+	}
+	event(a, "envwalk-staging")
+	event(a, nil)
+	event(a, "")
+	transaction(a, nil)
+	transaction(a, "")
+	logLine(a, nil)
+	logLine(a, "")
+	transaction(b, "envwalk-preview")
+	transaction(b, "envwalk-prod")
+	logLine(b, "envwalk-b-logs")
+	event(b, "envwalk-Upper")
+	event(b, "envwalk-älpha")
+	logLine(c, nil)
+	logLine(c, "")
+
+	// The previous full-index DISTINCT is the reference for every scope.
+	reference := func(scope []string) []string {
+		t.Helper()
+		rows, err := testPool.Query(ctx, `SELECT DISTINCT environment FROM (
+ SELECT environment FROM events WHERE (cardinality($1::uuid[]) = 0 OR project_id = ANY($1::uuid[]))
+ UNION SELECT environment FROM transactions WHERE (cardinality($1::uuid[]) = 0 OR project_id = ANY($1::uuid[]))
+ UNION SELECT environment FROM logs WHERE (cardinality($1::uuid[]) = 0 OR project_id = ANY($1::uuid[]))
+ ) environments WHERE environment IS NOT NULL AND environment <> '' ORDER BY environment`, scope)
+		require.NoError(t, err)
+		values, err := pgx.CollectRows(rows, pgx.RowTo[string])
+		require.NoError(t, err)
+		return values
+	}
+	fetch := func(query, token string) []string {
+		t.Helper()
+		req := httptest.NewRequest(http.MethodGet, "/api/environments"+query, nil)
+		if token == "" {
+			req.AddCookie(authCookie())
+		} else {
+			req.Header.Set("Authorization", "Bearer "+token)
+		}
+		rec := httptest.NewRecorder()
+		globalHandler().ServeHTTP(rec, req)
+		require.Equal(t, http.StatusOK, rec.Code, rec.Body.String())
+		var values []string
+		require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &values))
+		require.NotNil(t, values)
+		return values
+	}
+
+	got := fetch("?project_id="+a, "")
+	require.Equal(t, []string{"envwalk-prod", "envwalk-staging"}, got)
+	require.Equal(t, reference([]string{a}), got)
+	got = fetch("?project_id="+a+"&project_id="+b, "")
+	// Order follows the database collation, as it always has.
+	require.ElementsMatch(t, []string{"envwalk-Upper", "envwalk-b-logs", "envwalk-preview", "envwalk-prod", "envwalk-staging", "envwalk-älpha"}, got)
+	require.Equal(t, reference([]string{a, b}), got)
+	require.Equal(t, []string{}, fetch("?project_id="+c, ""))
+	require.Equal(t, []string{}, fetch("?project_id=00000000-0000-0000-0000-000000000001", ""))
+	all := fetch("", "")
+	require.Equal(t, reference([]string{}), all)
+	require.Subset(t, all, []string{"envwalk-Upper", "envwalk-b-logs", "envwalk-preview", "envwalk-prod", "envwalk-staging", "envwalk-älpha"})
+	require.NotContains(t, all, "")
+
+	// A project token only ever sees its own project, whatever is requested.
+	token := bearerToken(t, b)
+	require.Equal(t, reference([]string{b}), fetch("?project_id="+a, token))
+	require.Equal(t, reference([]string{b}), fetch("", token))
+}
diff --git a/internal/api/transaction_counts_test.go b/internal/api/transaction_counts_test.go
index 0de78b7d..ed73b590 100644
--- a/internal/api/transaction_counts_test.go
+++ b/internal/api/transaction_counts_test.go
@@ -17,7 +17,7 @@ func TestTransactionCountsAuthAndScope(t *testing.T) {
 	p, err := storage.CreateProject(ctx, testPool, "api-counts-scope", "Counts scope")
 	require.NoError(t, err)
 	t.Cleanup(func() { testPool.Exec(ctx, "DELETE FROM projects WHERE id=$1", p.ID) })
-	_, err = testPool.Exec(ctx, `INSERT INTO transactions(project_id,transaction,duration_ms,start_timestamp,timestamp) VALUES ($1,'/counts',123,NOW() - INTERVAL '1 minute',NOW() - INTERVAL '1 minute')`, p.ID)
+	_, err = testPool.Exec(ctx, `INSERT INTO transactions(project_id,transaction,duration_ms,start_timestamp,timestamp,received_at) VALUES ($1,'/counts',123,NOW() - INTERVAL '1 minute',NOW() - INTERVAL '1 minute',NOW() - INTERVAL '1 minute')`, p.ID)
 	require.NoError(t, err)
 	token := bearerToken(t, p.ID)
 	for _, auth := range []bool{false, true} {
diff --git a/internal/ingest/buffer.go b/internal/ingest/buffer.go
index 6f8cfdcb..aee0ac3f 100644
--- a/internal/ingest/buffer.go
+++ b/internal/ingest/buffer.go
@@ -62,9 +62,9 @@ func writeBatch(ctx context.Context, pool batchSender, batch []BufferedEvent) er
 	b := &pgx.Batch{}
 	rows := make([][]any, 0, len(batch))
 	for _, e := range batch {
-		rows = append(rows, []any{e.ProjectID, e.EventID, e.Timestamp, e.Payload, nilStr(e.TraceID), nilStr(e.SpanID)})
+		rows = append(rows, []any{e.ProjectID, e.EventID, e.Timestamp, e.Payload, nilStr(e.TraceID), nilStr(e.SpanID), newRowID()})
 	}
-	queueInserts(b, "INSERT INTO events (project_id, event_id, timestamp, payload, trace_id, span_id) VALUES ", " ON CONFLICT (project_id, event_id) WHERE event_id IS NOT NULL DO NOTHING", rows, 0)
+	queueInserts(b, "INSERT INTO events (project_id, event_id, timestamp, payload, trace_id, span_id, id) VALUES ", " ON CONFLICT (project_id, event_id) WHERE event_id IS NOT NULL DO NOTHING", rows, 0)
 	if err := execBatch(ctx, pool, b); err != nil {
 		return err
 	}
diff --git a/internal/ingest/logbuffer.go b/internal/ingest/logbuffer.go
index c8f620cb..4b98269d 100644
--- a/internal/ingest/logbuffer.go
+++ b/internal/ingest/logbuffer.go
@@ -54,9 +54,9 @@ func writeLogBatch(ctx context.Context, pool batchSender, batch []BufferedLog) e
 	b := &pgx.Batch{}
 	rows := make([][]any, 0, len(batch))
 	for _, l := range batch {
-		rows = append(rows, []any{l.ProjectID, l.Timestamp, l.Level, l.Body, nilStr(l.TraceID), nilStr(l.SpanID), nilStr(l.Environment), nilStr(l.Release), nilJSONDefault(l.Attributes)})
+		rows = append(rows, []any{l.ProjectID, l.Timestamp, l.Level, l.Body, nilStr(l.TraceID), nilStr(l.SpanID), nilStr(l.Environment), nilStr(l.Release), nilJSONDefault(l.Attributes), newRowID()})
 	}
-	queueInserts(b, "INSERT INTO logs (project_id, timestamp, level, body, trace_id, span_id, environment, release, attributes) VALUES ", "", rows, 0)
+	queueInserts(b, "INSERT INTO logs (project_id, timestamp, level, body, trace_id, span_id, environment, release, attributes, id) VALUES ", "", rows, 0)
 	if err := execBatch(ctx, pool, b); err != nil {
 		return err
 	}
diff --git a/internal/ingest/multiinsert.go b/internal/ingest/multiinsert.go
index 8f49078b..3eca962d 100644
--- a/internal/ingest/multiinsert.go
+++ b/internal/ingest/multiinsert.go
@@ -5,9 +5,19 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/google/uuid"
 	"github.com/jackc/pgx/v5"
 )
 
+// newRowID returns a time-ordered (v7) primary key. The column defaults are
+// random v4 UUIDs, which scatter inserts over the whole primary-key index; on
+// a large spans table that index no longer fits in memory and every insert
+// reads a page from disk. v7 keys land at the right edge, which stays cached.
+// Generated here rather than with uuidv7() so Postgres before 18 still works.
+func newRowID() string {
+	return uuid.Must(uuid.NewV7()).String()
+}
+
 // queueInserts groups rows into bounded statements so usage triggers aggregate
 // once per flush chunk. All chunks retain SendBatch's transaction semantics.
 func queueInserts(batch *pgx.Batch, prefix, suffix string, rows [][]any, projectColumn int) int {
diff --git a/internal/ingest/row_ids_test.go b/internal/ingest/row_ids_test.go
new file mode 100644
index 00000000..a9ea3286
--- /dev/null
+++ b/internal/ingest/row_ids_test.go
@@ -0,0 +1,75 @@
+package ingest_test
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+
+	"github.com/blendbyte/tindra/internal/ingest"
+)
+
+// Events, logs, transactions and spans get time-ordered (v7) primary keys
+// from ingest, so inserts append to their primary-key indexes instead of
+// landing on random pages.
+func TestIngestWritesTimeOrderedRowIDs(t *testing.T) {
+	ctx := context.Background()
+	now := time.Now()
+	eventID := "row-id-v7-event"
+
+	events := ingest.NewBuffer(10)
+	events.Push(ingest.BufferedEvent{ProjectID: testProject.ID, EventID: &eventID, Timestamp: now, Payload: json.RawMessage(`{"level":"error"}`)})
+	logs := ingest.NewLogBuffer(10)
+	logs.Push(ingest.BufferedLog{ProjectID: testProject.ID, Timestamp: now, Level: "info", Body: "row-id-v7-log"})
+	txs := ingest.NewTransactionBuffer(10)
+	txs.Push(ingest.BufferedTransaction{
+		ProjectID: testProject.ID, TraceID: "row-id-v7-trace", SpanID: "row-id-root", Transaction: "/row-ids",
+		Op: "http.server", Status: "ok", DurationMs: 5, StartTimestamp: now, Timestamp: now.Add(5 * time.Millisecond),
+		Spans: []ingest.BufferedSpan{
+			{SpanID: "row-id-a", ParentSpanID: "row-id-root", Op: "db.query", StartTimestamp: now, Timestamp: now, DurationMs: 1, Status: "ok"},
+			{SpanID: "row-id-b", ParentSpanID: "row-id-root", Op: "view.render", StartTimestamp: now, Timestamp: now, DurationMs: 1, Status: "ok"},
+		},
+	})
+
+	// A cancelled context drains each buffer once and returns.
+	runCtx, cancel := context.WithCancel(ctx)
+	cancel()
+	events.Run(runCtx, testPool)
+	logs.Run(runCtx, testPool)
+	txs.Run(runCtx, testPool)
+
+	queries := map[string]string{
+		"events":       `SELECT id::text FROM events WHERE project_id = $1 AND event_id = 'row-id-v7-event'`,
+		"logs":         `SELECT id::text FROM logs WHERE project_id = $1 AND body = 'row-id-v7-log'`,
+		"transactions": `SELECT id::text FROM transactions WHERE project_id = $1 AND trace_id = 'row-id-v7-trace'`,
+		"spans":        `SELECT id::text FROM spans WHERE project_id = $1 AND span_id IN ('row-id-a', 'row-id-b')`,
+	}
+	want := map[string]int{"events": 1, "logs": 1, "transactions": 1, "spans": 2}
+	for table, q := range queries {
+		rows, err := testPool.Query(ctx, q, testProject.ID)
+		if err != nil {
+			t.Fatalf("%s: %v", table, err)
+		}
+		n := 0
+		for rows.Next() {
+			var id string
+			if err := rows.Scan(&id); err != nil {
+				t.Fatalf("%s: %v", table, err)
+			}
+			parsed, err := uuid.Parse(id)
+			if err != nil {
+				t.Fatalf("%s: %v", table, err)
+			}
+			if parsed.Version() != 7 {
+				t.Errorf("%s id %s is version %d, want 7", table, id, parsed.Version())
+			}
+			n++
+		}
+		rows.Close()
+		if n != want[table] {
+			t.Errorf("%s: got %d rows, want %d", table, n, want[table])
+		}
+	}
+}
diff --git a/internal/ingest/txbuffer.go b/internal/ingest/txbuffer.go
index 5fd38220..354c049d 100644
--- a/internal/ingest/txbuffer.go
+++ b/internal/ingest/txbuffer.go
@@ -7,7 +7,6 @@ import (
 	"slices"
 	"time"
 
-	"github.com/google/uuid"
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgxpool"
 )
@@ -103,9 +102,8 @@ func writeTxBatch(ctx context.Context, pool batchSender, batch []BufferedTransac
 	txIDs := make([]string, len(batch))
 	rows := make([][]any, 0, len(batch))
 	for i, tx := range batch {
-		// Time-ordered (v7) IDs keep primary-key inserts at the right edge of
-		// the index instead of scattering them like random v4 IDs.
-		txIDs[i] = uuid.Must(uuid.NewV7()).String()
+		// Explicit, time-ordered IDs (see newRowID).
+		txIDs[i] = newRowID()
 		rows = append(rows, []any{txIDs[i], tx.ProjectID, nilStr(tx.TraceID), nilStr(tx.SpanID), tx.Transaction,
 			tx.Op, tx.Status, tx.DurationMs, tx.StartTimestamp, tx.Timestamp,
 			nilStr(tx.Environment), nilStr(tx.Release), nilStr(tx.Platform), nilJSON(tx.Measurements),
@@ -149,7 +147,7 @@ func writeTxBatch(ctx context.Context, pool batchSender, batch []BufferedTransac
 			spanRows = append(spanRows, []any{
 				txIDs[i], sp.SpanID, nilStr(sp.ParentSpanID), sp.Op, nilStr(sp.Description),
 				sp.StartTimestamp, sp.Timestamp, sp.DurationMs, sp.Status, nilJSON(sp.Data),
-				tx.ProjectID, nilStr(tx.Environment), nilStr(tx.Release),
+				tx.ProjectID, nilStr(tx.Environment), nilStr(tx.Release), newRowID(),
 			})
 		}
 	}
@@ -158,7 +156,7 @@ func writeTxBatch(ctx context.Context, pool batchSender, batch []BufferedTransac
 		queueInserts(spanBatch, `INSERT INTO spans
 			(transaction_id, span_id, parent_span_id, op, description,
 			 start_timestamp, timestamp, duration_ms, status, data,
-			 project_id, environment, release) VALUES `, "", spanRows, 10)
+			 project_id, environment, release, id) VALUES `, "", spanRows, 10)
 		if err := execBatch(ctx, pool, spanBatch); err != nil {
 			return nil, err
 		}
diff --git a/internal/retention/age_performance_test.go b/internal/retention/age_performance_test.go
index 1f7fc74c..cb401e76 100644
--- a/internal/retention/age_performance_test.go
+++ b/internal/retention/age_performance_test.go
@@ -58,7 +58,7 @@ func TestAgePerformance(t *testing.T) {
 							total += d
 							longest = max(longest, d)
 						}
-						require.Equal(t, []int64{5000, 5000, 0}, trace.batches)
+						require.Equal(t, wantBatches(10000, ageBatch(table)), trace.batches)
 						pool.Close()
 					} else {
 						start := time.Now()
diff --git a/internal/retention/age_test.go b/internal/retention/age_test.go
index 4dea1bfb..d46fa483 100644
--- a/internal/retention/age_test.go
+++ b/internal/retention/age_test.go
@@ -3,6 +3,7 @@ package retention_test
 import (
 	"context"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 
@@ -35,7 +36,7 @@ func TestWorkerAgeBatchesReceiptTimeAndCascades(t *testing.T) {
 			seedAgeRows(t, table, 12000, 7)
 			trace := &capTrace{matchSQL: "DELETE FROM " + table}
 			retention.NewWorker(capPool(t, trace), 90).RunOnce(context.Background())
-			require.Equal(t, []int64{5000, 5000, 2000}, trace.batches)
+			require.Equal(t, wantBatches(12000, ageBatch(table)), trace.batches)
 			require.Equal(t, 7, capCount(t, table))
 			var fresh int
 			require.NoError(t, testPool.QueryRow(context.Background(), "SELECT count(*) FROM "+table+" WHERE received_at > NOW()-interval '1 day'").Scan(&fresh))
@@ -63,23 +64,46 @@ func TestWorkerAgeCancellationAndArrivals(t *testing.T) {
 				require.NoError(t, err)
 			}
 			retention.NewWorker(capPool(t, trace), 90).RunOnce(ctx)
-			require.Equal(t, []int64{5000}, trace.batches)
-			require.Equal(t, 7008, capCount(t, table))
+			require.Equal(t, []int64{ageBatch(table)}, trace.batches)
+			require.Equal(t, 12008-int(ageBatch(table)), capCount(t, table))
 			retention.NewWorker(testPool, 90).RunOnce(context.Background())
 			require.Equal(t, 8, capCount(t, table))
 		})
 	}
 }
-func TestWorkerAgePassBudget(t *testing.T) {
+
+// ageBatch is the largest batch the worker uses for table. Transactions
+// cascade to their spans, so they go in smaller batches.
+func ageBatch(table string) int64 {
+	if table == "transactions" {
+		return retention.TxPurgeBatch
+	}
+	return retention.AgePurgeBatch
+}
+
+// A pass used to stop at 200k rows per table. With time to spare it now
+// drains the whole backlog in one pass and reports nothing left behind.
+func TestWorkerAgePassContinuesPastOldCap(t *testing.T) {
 	seedAgeRows(t, "logs", 205000, 7)
 	trace := &capTrace{matchSQL: "DELETE FROM logs"}
-	worker := retention.NewWorker(capPool(t, trace), 90)
+	clock := newStepClock()
+	worker := retention.NewWorker(capPool(t, trace), 90).WithClock(clock.Now, time.Minute)
+	require.False(t, worker.RunOnce(context.Background()))
+	require.Equal(t, wantBatches(205000, retention.AgePurgeBatch), trace.batches)
+	require.Equal(t, 7, capCount(t, "logs"))
+}
+
+// When the table budget runs out mid-backlog the pass stops between batches,
+// reports the backlog, and the next pass picks up where it left off.
+func TestWorkerAgeBudgetStopsBetweenBatchesAndResumes(t *testing.T) {
+	seedAgeRows(t, "logs", 12000, 7)
+	clock := newStepClock()
+	trace := &capTrace{matchSQL: "DELETE FROM logs", afterBatch: func() { clock.Advance(600 * time.Millisecond) }}
+	worker := retention.NewWorker(capPool(t, trace), 90).WithClock(clock.Now, time.Second)
 	require.True(t, worker.RunOnce(context.Background()))
-	require.Len(t, trace.batches, 40)
-	for _, n := range trace.batches {
-		require.EqualValues(t, 5000, n)
-	}
-	require.Equal(t, 5007, capCount(t, "logs"))
+	require.Equal(t, []int64{5000, 5000}, trace.batches)
+	require.Equal(t, 2007, capCount(t, "logs"))
 	require.False(t, worker.RunOnce(context.Background()))
+	require.Equal(t, []int64{5000, 5000, 2000}, trace.batches)
 	require.Equal(t, 7, capCount(t, "logs"))
 }
diff --git a/internal/retention/budget_test.go b/internal/retention/budget_test.go
new file mode 100644
index 00000000..74a6d66d
--- /dev/null
+++ b/internal/retention/budget_test.go
@@ -0,0 +1,77 @@
+package retention_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/blendbyte/tindra/internal/retention"
+)
+
+// stepClock only moves when a test advances it, so table budgets run out
+// exactly where the test says.
+type stepClock struct{ t time.Time }
+
+func newStepClock() *stepClock               { return &stepClock{t: time.Unix(0, 0)} }
+func (c *stepClock) Now() time.Time          { return c.t }
+func (c *stepClock) Advance(d time.Duration) { c.t = c.t.Add(d) }
+
+// wantBatches lists the batch sizes a drain of total rows reports: full
+// batches, then the short batch that tells it the table caught up.
+func wantBatches(total, size int64) []int64 {
+	var out []int64
+	for ; total >= size; total -= size {
+		out = append(out, size)
+	}
+	return append(out, total)
+}
+
+// Events go first in a pass. Their first batch alone overruns the budget, yet
+// every later table still gets its own full budget in the same pass.
+func TestWorkerSlowTableDoesNotStarveOthers(t *testing.T) {
+	truncate(t)
+	ctx := context.Background()
+	issue := seedRetainedIssue(t, "open", 12000, 7)
+	seedAgeRows(t, "logs", 12000, 7)
+	seedUptimeBacklog(t, 12000, 7)
+	clock := newStepClock()
+	trace := &capTrace{matchSQL: "DELETE FROM events", afterBatch: func() { clock.Advance(2 * time.Minute) }}
+	worker := retention.NewWorker(capPool(t, trace), 90).WithClock(clock.Now, time.Minute)
+	require.True(t, worker.RunOnce(ctx), "events are still behind")
+	require.Equal(t, []int64{5000}, trace.batches)
+	require.Equal(t, 7007, capCount(t, "events"))
+	require.Equal(t, 7007, issueEventCount(t, issue))
+	require.Equal(t, 7, capCount(t, "logs"))
+	require.Equal(t, 7, capCount(t, "uptime_checks"))
+}
+
+// A slow statement shrinks the next batch in proportion to the overrun, and
+// fast ones grow it back toward the table's maximum.
+func TestWorkerBatchSizeAdaptsToStatementDuration(t *testing.T) {
+	seedAgeRows(t, "logs", 12000, 7)
+	clock := newStepClock()
+	trace := &capTrace{matchSQL: "DELETE FROM logs"}
+	trace.afterBatch = func() {
+		if len(trace.batches) == 1 {
+			clock.Advance(4 * time.Second)
+		}
+	}
+	require.False(t, retention.NewWorker(capPool(t, trace), 90).WithClock(clock.Now, time.Hour).RunOnce(context.Background()))
+	require.Equal(t, []int64{5000, 1250, 2500, 3250}, trace.batches)
+	require.Equal(t, 7, capCount(t, "logs"))
+}
+
+// The adapted size carries over to the next pass, so a table with heavy
+// cascades does not open every pass with a slow statement.
+func TestWorkerBatchSizeCarriesAcrossPasses(t *testing.T) {
+	seedAgeRows(t, "logs", 12000, 7)
+	clock := newStepClock()
+	trace := &capTrace{matchSQL: "DELETE FROM logs"}
+	trace.afterBatch = func() { clock.Advance(2 * time.Second) }
+	worker := retention.NewWorker(capPool(t, trace), 90).WithClock(clock.Now, time.Second)
+	require.True(t, worker.RunOnce(context.Background()))
+	require.True(t, worker.RunOnce(context.Background()))
+	require.Equal(t, []int64{5000, 2500}, trace.batches)
+}
diff --git a/internal/retention/cron_batch_test.go b/internal/retention/cron_batch_test.go
index 98421bf6..dd010dc2 100644
--- a/internal/retention/cron_batch_test.go
+++ b/internal/retention/cron_batch_test.go
@@ -76,19 +76,17 @@ func TestCronRetentionCancellationResumes(t *testing.T) {
 }
 
 func TestCronRetentionBudgetAndDisabled(t *testing.T) {
-	seedCronBacklog(t, 205000)
+	seedCronBacklog(t, 12000)
 	ctx := t.Context()
 	// Other policies can enable the worker without enabling cron age cleanup.
 	require.False(t, retention.NewWorker(testPool, 0).WithRowLimits(1000000, 1000000).RunOnce(ctx))
-	require.Equal(t, 205003, capCount(t, "cron_checkins"))
-	trace := &capTrace{matchSQL: "DELETE FROM cron_checkins"}
-	w := retention.NewWorker(capPool(t, trace), 90)
-	require.True(t, w.RunOnce(ctx), "a capped pass must request an earlier follow-up")
-	require.Len(t, trace.batches, 40)
-	for _, n := range trace.batches {
-		require.EqualValues(t, 5000, n)
-	}
-	require.Equal(t, 5003, capCount(t, "cron_checkins"))
+	require.Equal(t, 12003, capCount(t, "cron_checkins"))
+	clock := newStepClock()
+	trace := &capTrace{matchSQL: "DELETE FROM cron_checkins", afterBatch: func() { clock.Advance(600 * time.Millisecond) }}
+	w := retention.NewWorker(capPool(t, trace), 90).WithClock(clock.Now, time.Second)
+	require.True(t, w.RunOnce(ctx), "a pass out of budget must request an earlier follow-up")
+	require.Equal(t, []int64{5000, 5000}, trace.batches)
+	require.Equal(t, 2003, capCount(t, "cron_checkins"))
 	require.False(t, w.RunOnce(ctx))
 	require.Equal(t, 3, capCount(t, "cron_checkins"))
 }
diff --git a/internal/retention/event_batch_test.go b/internal/retention/event_batch_test.go
index 6c7213e6..0688bb14 100644
--- a/internal/retention/event_batch_test.go
+++ b/internal/retention/event_batch_test.go
@@ -38,7 +38,8 @@ func TestWorkerEventBatchesReconcileAndPreserveShells(t *testing.T) {
 	require.NoError(t, err)
 	_, err = testPool.Exec(ctx, `INSERT INTO event_tags(event_id,issue_id,project_id,key,value) SELECT id,issue_id,project_id,'test','value' FROM events`)
 	require.NoError(t, err)
-	// Retention must still repair drift for affected issues, not merely subtract.
+	// Retention subtracts what it deleted instead of recounting the issue, so a
+	// drifted count stays drifted by the same amount.
 	_, err = testPool.Exec(ctx, "UPDATE issues SET event_count=999999 WHERE id=$1", open)
 	require.NoError(t, err)
 	trace := &capTrace{matchSQL: "DELETE FROM events"}
@@ -49,7 +50,7 @@ func TestWorkerEventBatchesReconcileAndPreserveShells(t *testing.T) {
 	require.Equal(t, 2, capCount(t, "issues"))
 	var count int
 	require.NoError(t, testPool.QueryRow(ctx, "SELECT event_count FROM issues WHERE id=$1", open).Scan(&count))
-	require.Equal(t, 7, count)
+	require.Equal(t, 999999-12000, count)
 	require.NoError(t, testPool.QueryRow(ctx, "SELECT event_count FROM issues WHERE id=$1", regressed).Scan(&count))
 	require.Zero(t, count)
 }
@@ -103,7 +104,7 @@ func (b *beforeIssueLock) TraceQueryStart(ctx context.Context, _ *pgx.Conn, d pg
 	return ctx
 }
 func (*beforeIssueLock) TraceQueryEnd(context.Context, *pgx.Conn, pgx.TraceQueryEndData) {}
-func TestWorkerEventRecountIncludesConcurrentGrouping(t *testing.T) {
+func TestWorkerEventDecrementIncludesConcurrentGrouping(t *testing.T) {
 	truncate(t)
 	ctx := context.Background()
 	issue := seedRetainedIssue(t, "open", 1, 0)
@@ -134,18 +135,116 @@ func TestWorkerEventRecountIncludesConcurrentGrouping(t *testing.T) {
 	require.Equal(t, 1, count)
 	require.Equal(t, 1, capCount(t, "events"))
 }
-func TestWorkerEventPassBudget(t *testing.T) {
+func TestWorkerEventBudgetStopsBetweenBatchesAndResumes(t *testing.T) {
 	truncate(t)
 	ctx := context.Background()
-	id := seedRetainedIssue(t, "open", 200001, 0)
-	retention.NewWorker(testPool, 90).RunOnce(ctx)
+	id := seedRetainedIssue(t, "open", 12000, 1)
+	clock := newStepClock()
+	trace := &capTrace{matchSQL: "DELETE FROM events", afterBatch: func() { clock.Advance(600 * time.Millisecond) }}
+	worker := retention.NewWorker(capPool(t, trace), 90).WithClock(clock.Now, time.Second)
+	require.True(t, worker.RunOnce(ctx))
+	require.Equal(t, []int64{5000, 5000}, trace.batches)
+	require.Equal(t, 2001, capCount(t, "events"))
+	require.Equal(t, 2001, issueEventCount(t, id))
+	require.False(t, worker.RunOnce(ctx))
 	require.Equal(t, 1, capCount(t, "events"))
-	var count int
-	require.NoError(t, testPool.QueryRow(ctx, "SELECT event_count FROM issues WHERE id=$1", id).Scan(&count))
-	require.Equal(t, 1, count)
-	retention.NewWorker(testPool, 90).RunOnce(ctx)
+	require.Equal(t, 1, issueEventCount(t, id))
+}
+
+func issueEventCount(t *testing.T, id string) int {
+	t.Helper()
+	var n int
+	require.NoError(t, testPool.QueryRow(context.Background(), "SELECT event_count FROM issues WHERE id=$1", id).Scan(&n))
+	return n
+}
+
+func issueExists(t *testing.T, id string) bool {
+	t.Helper()
+	var ok bool
+	require.NoError(t, testPool.QueryRow(context.Background(), "SELECT EXISTS (SELECT 1 FROM issues WHERE id=$1)", id).Scan(&ok))
+	return ok
+}
+
+// Two issues of 6000 expired events each span three batches: the first issue
+// fills batch one and starts batch two, which the second issue shares and
+// finishes in batch three. Each batch subtracts only its own rows.
+func TestWorkerEventDecrementAcrossBatches(t *testing.T) {
+	truncate(t)
+	ctx := context.Background()
+	a := seedRetainedIssue(t, "open", 6000, 3)
+	b := seedRetainedIssue(t, "regressed", 6000, 4)
+	var counts [][2]int
+	trace := &capTrace{matchSQL: "DELETE FROM events"}
+	pool := capPool(t, trace)
+	worker := retention.NewWorker(pool, 90)
+	// Snapshot both counts after each committed batch. The hook runs while the
+	// batch transaction is still open, so record the previous batch's result.
+	trace.afterBatch = func() {
+		counts = append(counts, [2]int{issueEventCount(t, a), issueEventCount(t, b)})
+	}
+	require.False(t, worker.RunOnce(ctx))
+	require.Equal(t, []int64{5000, 5000, 2000}, trace.batches)
+	require.Equal(t, [][2]int{{6003, 6004}, {1003, 6004}, {3, 2004}}, counts)
+	require.Equal(t, 3, issueEventCount(t, a))
+	require.Equal(t, 4, issueEventCount(t, b))
+	require.Equal(t, 7, capCount(t, "events"))
+}
+
+// A resolved or ignored issue goes once a later batch removes its last event,
+// not before. Open and regressed issues keep their shell at zero.
+func TestWorkerEventDeletesIssueWhenLastEventPurged(t *testing.T) {
+	truncate(t)
+	ctx := context.Background()
+	ids := map[string]string{}
+	for _, status := range []string{"resolved", "ignored", "open", "regressed"} {
+		ids[status] = seedRetainedIssue(t, status, 3000, 0)
+	}
+	var afterFirst map[string]bool
+	trace := &capTrace{matchSQL: "DELETE FROM events"}
+	trace.afterBatch = func() {
+		if len(trace.batches) == 2 {
+			// The first batch has committed: it emptied resolved and touched
+			// ignored, which still has events.
+			afterFirst = map[string]bool{}
+			for status, id := range ids {
+				afterFirst[status] = issueExists(t, id)
+			}
+		}
+	}
+	retention.NewWorker(capPool(t, trace), 90).RunOnce(ctx)
+	require.Equal(t, []int64{5000, 5000, 2000}, trace.batches)
+	require.Equal(t, map[string]bool{"resolved": false, "ignored": true, "open": true, "regressed": true}, afterFirst)
+	require.False(t, issueExists(t, ids["resolved"]))
+	require.False(t, issueExists(t, ids["ignored"]))
+	for _, status := range []string{"open", "regressed"} {
+		require.True(t, issueExists(t, ids[status]), status)
+		require.Zero(t, issueEventCount(t, ids[status]), status)
+	}
 	require.Zero(t, capCount(t, "events"))
 }
+
+// Counts drift, so emptiness comes from the events themselves. An issue whose
+// stored count overstates its rows is still emptied, and one whose count
+// understates them is clamped at zero but kept while events remain.
+func TestWorkerEventEmptinessIgnoresDriftedCounts(t *testing.T) {
+	truncate(t)
+	ctx := context.Background()
+	overstated := seedRetainedIssue(t, "resolved", 2, 0)
+	overstatedOpen := seedRetainedIssue(t, "open", 2, 0)
+	understated := seedRetainedIssue(t, "ignored", 3, 2)
+	_, err := testPool.Exec(ctx, "UPDATE issues SET event_count=50 WHERE id = ANY($1::uuid[])", []string{overstated, overstatedOpen})
+	require.NoError(t, err)
+	_, err = testPool.Exec(ctx, "UPDATE issues SET event_count=1 WHERE id=$1", understated)
+	require.NoError(t, err)
+	retention.NewWorker(testPool, 90).RunOnce(ctx)
+	require.False(t, issueExists(t, overstated))
+	require.True(t, issueExists(t, overstatedOpen))
+	require.Zero(t, issueEventCount(t, overstatedOpen))
+	require.True(t, issueExists(t, understated))
+	require.Zero(t, issueEventCount(t, understated))
+	require.Equal(t, 2, capCount(t, "events"))
+}
+
 func TestMembershipMovesWaitForRetention(t *testing.T) {
 	truncate(t)
 	ctx := context.Background()
diff --git a/internal/retention/export_test.go b/internal/retention/export_test.go
new file mode 100644
index 00000000..accfa134
--- /dev/null
+++ b/internal/retention/export_test.go
@@ -0,0 +1,18 @@
+package retention
+
+import "time"
+
+// WithClock replaces the time source behind table budgets and statement
+// durations, and the per-table budget itself, so tests can run a table out of
+// budget without waiting.
+func (w *Worker) WithClock(now func() time.Time, budget time.Duration) *Worker {
+	w.now = now
+	w.tableBudget = budget
+	return w
+}
+
+// Batch limits, exported for assertions on batch sizes.
+const (
+	AgePurgeBatch = agePurgeBatch
+	TxPurgeBatch  = txPurgeBatch
+)
diff --git a/internal/retention/failures_internal_test.go b/internal/retention/failures_internal_test.go
index a569f77a..62fb8fff 100644
--- a/internal/retention/failures_internal_test.go
+++ b/internal/retention/failures_internal_test.go
@@ -70,21 +70,29 @@ func TestRetentionUnavailableDatabaseStopsWithoutReportingDeletes(t *testing.T)
 	require.NoError(t, err)
 	pool.Close()
 	w := NewWorker(pool, 90).WithRowLimits(1, 1).WithProfileLimits(1, 1)
-	require.Zero(t, w.purgeProfiles(ctx, time.Now()))
+	// A failed statement reports neither deletions nor a backlog, so the loop
+	// waits the full hour rather than hammering an unavailable database.
+	none := func(n int64, behind bool) {
+		t.Helper()
+		require.Zero(t, n)
+		require.False(t, behind)
+	}
+	none(w.purgeProfiles(ctx, time.Now()))
 	require.Zero(t, w.purgeProfilesStorageCap(ctx))
-	e, i := w.purgeEvents(ctx, time.Now())
+	e, i, behind := w.purgeEvents(ctx, time.Now())
 	require.Zero(t, e)
 	require.Zero(t, i)
-	require.Zero(t, w.purgeTransactions(ctx, time.Now()))
-	require.Zero(t, w.purgeCronCheckins(ctx, time.Now()))
+	require.False(t, behind)
+	none(w.purgeTransactions(ctx, time.Now()))
+	none(w.purgeCronCheckins(ctx, time.Now()))
 	require.Zero(t, w.purgeAlertFirings(ctx))
-	require.Zero(t, w.purgeLogsRowCap(ctx))
-	w.purgeExpiredAuthTokens(ctx)
+	none(w.purgeLogsRowCap(ctx))
+	require.False(t, w.purge(ctx))
 	cancelled, cancel := context.WithCancel(ctx)
 	cancel()
 	require.Zero(t, w.purgeAlertFirings(cancelled))
-	require.Zero(t, w.purgeLogsRowCap(cancelled))
-	require.Zero(t, w.purgeCronCheckins(cancelled, time.Now()))
+	none(w.purgeLogsRowCap(cancelled))
+	none(w.purgeCronCheckins(cancelled, time.Now()))
 }
 
 func TestTokenRetentionDeleteFailurePreservesTokenForRetry(t *testing.T) {
diff --git a/internal/retention/profile_age_budget_test.go b/internal/retention/profile_age_budget_test.go
index d3c578c6..5f24c9ac 100644
--- a/internal/retention/profile_age_budget_test.go
+++ b/internal/retention/profile_age_budget_test.go
@@ -3,6 +3,7 @@ package retention_test
 import (
 	"context"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 
@@ -22,22 +23,20 @@ func seedProfileAgeBacklog(t *testing.T, expired, fresh int) {
 	require.NoError(t, err)
 }
 func TestWorkerProfileAgePassBudget(t *testing.T) {
-	seedProfileAgeBacklog(t, 200001, 7)
+	seedProfileAgeBacklog(t, 10001, 7)
 	ctx := context.Background()
-	trace := &capTrace{matchSQL: "DELETE FROM profile_chunks"}
-	worker := retention.NewWorker(capPool(t, trace), 0).WithProfileLimits(7, 0)
-	worker.RunOnce(ctx)
-	require.Len(t, trace.batches, 40)
-	for _, n := range trace.batches {
-		require.EqualValues(t, 5000, n)
-	}
+	clock := newStepClock()
+	trace := &capTrace{matchSQL: "DELETE FROM profile_chunks", afterBatch: func() { clock.Advance(600 * time.Millisecond) }}
+	worker := retention.NewWorker(capPool(t, trace), 0).WithProfileLimits(7, 0).WithClock(clock.Now, time.Second)
+	require.True(t, worker.RunOnce(ctx))
+	require.Equal(t, []int64{5000, 5000}, trace.batches)
 	require.Equal(t, 8, countProfiles(t))
 	var fresh int
 	require.NoError(t, testPool.QueryRow(ctx, "SELECT count(*) FROM profile_chunks WHERE received_at > NOW()-interval '1 day'").Scan(&fresh))
 	require.Equal(t, 7, fresh)
-	worker.RunOnce(ctx)
+	require.False(t, worker.RunOnce(ctx))
 	require.Equal(t, 7, countProfiles(t))
-	require.Equal(t, int64(1), trace.batches[40])
+	require.Equal(t, []int64{5000, 5000, 1}, trace.batches)
 }
 func TestWorkerProfileAgeCancellationResumes(t *testing.T) {
 	seedProfileAgeBacklog(t, 12000, 7)
diff --git a/internal/retention/row_cap_test.go b/internal/retention/row_cap_test.go
index 3326f242..e9f95112 100644
--- a/internal/retention/row_cap_test.go
+++ b/internal/retention/row_cap_test.go
@@ -102,7 +102,7 @@ func TestWorkerRowCapBatchesAndTies(t *testing.T) {
 			}
 			trace := &capTrace{}
 			runCap(ctx, capPool(t, trace), table, 7)
-			require.Equal(t, []int64{5000, 5000, 2000}, trace.batches)
+			require.Equal(t, wantBatches(12000, ageBatch(table)), trace.batches)
 			var ids []string
 			rows, err := testPool.Query(ctx, "SELECT id::text FROM "+table+" ORDER BY id")
 			require.NoError(t, err)
@@ -134,24 +134,22 @@ func TestWorkerRowCapCancellationKeepsCommittedBatch(t *testing.T) {
 			defer cancel()
 			trace := &capTrace{cancel: cancel}
 			runCap(ctx, capPool(t, trace), table, 7)
-			require.Equal(t, []int64{5000}, trace.batches)
-			require.Equal(t, 7007, capCount(t, table))
+			require.Equal(t, []int64{ageBatch(table)}, trace.batches)
+			require.Equal(t, 12007-int(ageBatch(table)), capCount(t, table))
 			runCap(context.Background(), testPool, table, 7)
 			require.Equal(t, 7, capCount(t, table))
 		})
 	}
 }
 func TestWorkerRowCapPassBudget(t *testing.T) {
-	seedCapRows(t, "logs", 205007)
-	trace := &capTrace{}
-	p := capPool(t, trace)
-	runCap(context.Background(), p, "logs", 7)
-	require.Len(t, trace.batches, 40)
-	for _, n := range trace.batches {
-		require.EqualValues(t, 5000, n)
-	}
-	require.Equal(t, 5007, capCount(t, "logs"))
-	runCap(context.Background(), p, "logs", 7)
+	seedCapRows(t, "logs", 12007)
+	clock := newStepClock()
+	trace := &capTrace{afterBatch: func() { clock.Advance(600 * time.Millisecond) }}
+	w := retention.NewWorker(capPool(t, trace), 0).WithRowLimits(7, 0).WithClock(clock.Now, time.Second)
+	require.True(t, w.RunOnce(context.Background()))
+	require.Equal(t, []int64{5000, 5000}, trace.batches)
+	require.Equal(t, 2007, capCount(t, "logs"))
+	require.False(t, w.RunOnce(context.Background()))
 	require.Equal(t, 7, capCount(t, "logs"))
 }
 
@@ -171,7 +169,7 @@ func TestWorkerRowCapPreservesNewerArrivalsBetweenBatches(t *testing.T) {
 				require.NoError(t, err)
 			}
 			runCap(ctx, capPool(t, trace), table, 7)
-			require.Equal(t, []int64{5000, 5000, 2000}, trace.batches)
+			require.Equal(t, wantBatches(12000, ageBatch(table)), trace.batches)
 			require.Equal(t, 27, capCount(t, table))
 			column := "timestamp"
 			if table == "transactions" {
diff --git a/internal/retention/scheduling_internal_test.go b/internal/retention/scheduling_internal_test.go
index cee26542..d4aff7ff 100644
--- a/internal/retention/scheduling_internal_test.go
+++ b/internal/retention/scheduling_internal_test.go
@@ -8,7 +8,7 @@ import (
 	"time"
 )
 
-func TestRetentionRetriesExhaustedBudgetThenReturnsToHourly(t *testing.T) {
+func TestRetentionRetriesBacklogSoonThenReturnsToHourly(t *testing.T) {
 	synctest.Test(t, func(t *testing.T) {
 		ctx, cancel := context.WithCancel(context.Background())
 		defer cancel()
@@ -18,17 +18,24 @@ func TestRetentionRetriesExhaustedBudgetThenReturnsToHourly(t *testing.T) {
 		if calls.Load() != 1 {
 			t.Fatalf("startup calls=%d", calls.Load())
 		}
-		time.Sleep(5 * time.Minute)
+		// A pass that left a table behind is followed half a minute later,
+		// not after the five minutes that used to cap throughput.
+		time.Sleep(29 * time.Second)
+		synctest.Wait()
+		if calls.Load() != 1 {
+			t.Fatalf("backlog retry ran early: %d", calls.Load())
+		}
+		time.Sleep(time.Second)
 		synctest.Wait()
 		if calls.Load() != 2 {
 			t.Fatalf("backlog retry calls=%d", calls.Load())
 		}
-		time.Sleep(59 * time.Minute)
+		time.Sleep(time.Hour - time.Second)
 		synctest.Wait()
 		if calls.Load() != 2 {
 			t.Fatalf("quiet cycle ran early: %d", calls.Load())
 		}
-		time.Sleep(time.Minute)
+		time.Sleep(time.Second)
 		synctest.Wait()
 		if calls.Load() != 3 {
 			t.Fatalf("hourly calls=%d", calls.Load())
@@ -52,12 +59,12 @@ func TestRetentionWaitStartsAfterPassFinishes(t *testing.T) {
 			return true
 		})
 		synctest.Wait()
-		time.Sleep(14 * time.Minute)
+		time.Sleep(10*time.Minute + backlogDelay - time.Second)
 		synctest.Wait()
 		if calls.Load() != 1 {
 			t.Fatalf("passes overlapped: %d", calls.Load())
 		}
-		time.Sleep(time.Minute)
+		time.Sleep(time.Second)
 		synctest.Wait()
 		if calls.Load() != 2 {
 			t.Fatalf("backlog did not resume: %d", calls.Load())
diff --git a/internal/retention/uptime_batch_test.go b/internal/retention/uptime_batch_test.go
index fe792ff2..dabdda8a 100644
--- a/internal/retention/uptime_batch_test.go
+++ b/internal/retention/uptime_batch_test.go
@@ -3,6 +3,7 @@ package retention_test
 import (
 	"context"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 
@@ -62,21 +63,19 @@ func TestWorkerUptimeCancellationResumes(t *testing.T) {
 	require.Equal(t, 7, capCount(t, "uptime_checks"))
 }
 func TestWorkerUptimeBudgetAndDisabled(t *testing.T) {
-	seedUptimeBacklog(t, 205000, 7)
+	seedUptimeBacklog(t, 12000, 7)
 	ctx := context.Background()
 	retention.NewWorker(testPool, 0).RunOnce(ctx)
-	require.Equal(t, 205007, capCount(t, "uptime_checks"))
-	trace := &capTrace{matchSQL: "DELETE FROM uptime_checks"}
-	worker := retention.NewWorker(capPool(t, trace), 90)
-	worker.RunOnce(ctx)
-	require.Len(t, trace.batches, 40)
-	for _, n := range trace.batches {
-		require.EqualValues(t, 5000, n)
-	}
-	require.Equal(t, 5007, capCount(t, "uptime_checks"))
+	require.Equal(t, 12007, capCount(t, "uptime_checks"))
+	clock := newStepClock()
+	trace := &capTrace{matchSQL: "DELETE FROM uptime_checks", afterBatch: func() { clock.Advance(600 * time.Millisecond) }}
+	worker := retention.NewWorker(capPool(t, trace), 90).WithClock(clock.Now, time.Second)
+	require.True(t, worker.RunOnce(ctx))
+	require.Equal(t, []int64{5000, 5000}, trace.batches)
+	require.Equal(t, 2007, capCount(t, "uptime_checks"))
 	var first string
 	require.NoError(t, testPool.QueryRow(ctx, "SELECT id::text FROM uptime_checks ORDER BY checked_at,id LIMIT 1").Scan(&first))
-	require.Equal(t, "00000000-0000-0000-0000-000000030d41", first) // 200001
-	worker.RunOnce(ctx)
+	require.Equal(t, "00000000-0000-0000-0000-000000002711", first) // 10001
+	require.False(t, worker.RunOnce(ctx))
 	require.Equal(t, 7, capCount(t, "uptime_checks"))
 }
diff --git a/internal/retention/worker.go b/internal/retention/worker.go
index 385ae3bf..367b9f91 100644
--- a/internal/retention/worker.go
+++ b/internal/retention/worker.go
@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
+	"slices"
 	"time"
 
 	"github.com/jackc/pgx/v5"
@@ -23,27 +24,70 @@ type Worker struct {
 
 	profileRetentionDays  int
 	profileStorageLimitMB int
+
+	// now measures table budgets and statement durations. Cutoffs still use
+	// wall-clock time. Tests swap it for a clock they advance by hand.
+	now         func() time.Time
+	tableBudget time.Duration
+	// batchSizes remembers each table's adapted batch size across passes, so a
+	// table with heavy cascades does not start every pass with a slow statement.
+	// Passes never overlap, so the map needs no lock.
+	batchSizes map[string]int64
 }
 
-// profilePurgeBatch caps how many profile rows one DELETE removes. Profiles
-// are the only TOASTed payload in the schema, so purging a day of them in a
-// single statement means a long lock and a large WAL burst. Looping in batches
-// keeps both bounded.
-const profilePurgeBatch = 5000
+// tableBudget bounds how long one pass spends on a single table. The budget is
+// per table, so a large backlog in one table cannot starve the others: every
+// table gets its turn in every pass. One minute of 5000-row batches removes
+// millions of rows, against the fixed 200k a pass used to allow, which capped
+// age retention at about 2.4M rows an hour per table and let a busy install
+// grow forever.
+const tableBudget = time.Minute
+
+// idleDelay is the wait between passes when every table is caught up.
+// backlogDelay is the wait when a table ran out of budget with rows still to
+// delete. It is short so a backlog keeps draining at most of the budget's
+// rate, yet still leaves the database a quiet gap between passes.
+const (
+	idleDelay    = time.Hour
+	backlogDelay = 30 * time.Second
+)
 
-// rowCapBatch limits parent rows per transaction; cascaded children may exceed it.
-const rowCapBatch = 5000
+// batchTarget is the statement duration batch sizes adapt toward. A batch
+// that takes longer is scaled down in proportion, one that finishes in under
+// half of it doubles, up to the table's maximum. Statements, and the locks and
+// WAL bursts that come with them, stay around a second however many child rows
+// a parent row cascades to on this install.
+const batchTarget = time.Second
+
+// minBatch keeps a pathologically slow table making progress.
+const minBatch = 100
+
+// agePurgeBatch is the largest batch for tables whose rows delete on their
+// own (logs, uptime checks, cron check-ins) or with a few children (events
+// and their tags). It also caps profile deletes: profiles are the only
+// TOASTed payload in the schema, so a single statement for a day of them means
+// a long lock and a large WAL burst.
+const agePurgeBatch = 5000
 
-// maxRowCapDeletesPerPass lets a large backlog yield to the next scheduled cycle.
-const maxRowCapDeletesPerPass = 200_000
+// txPurgeBatch is the largest batch for transactions. Each one cascades to its
+// spans and performance events, about 40 spans per transaction on real
+// installs, so 1000 transactions already delete some 40k rows in a statement.
+const txPurgeBatch = 1000
 
-// maxStorageCapIDsPerPass bounds how many ids one storage-cap pass holds in
-// memory. Dropping PROFILE_STORAGE_LIMIT_MB on a large instance would otherwise
-// load every over-budget id into the retention goroutine at once.
-const maxStorageCapIDsPerPass = 200_000
+// maxIDsPerPass bounds how many ids a select-then-delete pass holds in memory
+// (alert firings, the profile storage cap, expired auth tokens). Dropping
+// PROFILE_STORAGE_LIMIT_MB on a large instance would otherwise load every
+// over-budget id into the retention goroutine at once.
+const maxIDsPerPass = 200_000
 
 func NewWorker(pool *pgxpool.Pool, retentionDays int) *Worker {
-	return &Worker{pool: pool, retentionDays: retentionDays}
+	return &Worker{
+		pool:          pool,
+		retentionDays: retentionDays,
+		now:           time.Now,
+		tableBudget:   tableBudget,
+		batchSizes:    make(map[string]int64),
+	}
 }
 
 // WithRowLimits sets instance-wide row caps for logs and transactions.
@@ -73,17 +117,18 @@ func (w *Worker) enabled() bool {
 		w.profileRetentionDays > 0 || w.profileStorageLimitMB > 0
 }
 
-// RunOnce runs a single purge cycle and reports whether a telemetry deletion
-// budget was exhausted. Useful for testing and explicit maintenance passes.
+// RunOnce runs a single purge cycle and reports whether a table ran out of its
+// time budget with rows still to delete. Useful for testing and explicit
+// maintenance passes.
 func (w *Worker) RunOnce(ctx context.Context) bool {
 	return w.purge(ctx)
 }
 
 // Run starts the retention loop. Call in a dedicated goroutine.
-// Runs immediately, then hourly when caught up or after five minutes when a
-// telemetry deletion budget was exhausted. Passes never overlap. It runs even
-// without retention limits, because housekeeping (expired sessions and tokens,
-// the per-rule alert firing cap) is needed regardless.
+// Runs immediately, then hourly when caught up or after backlogDelay when a
+// table is still behind. Passes never overlap. It runs even without retention
+// limits, because housekeeping (expired sessions and tokens, the per-rule
+// alert firing cap) is needed regardless.
 func (w *Worker) Run(ctx context.Context) {
 	if !w.enabled() {
 		slog.Info("retention: no retention limits configured; only housekeeping runs")
@@ -93,9 +138,9 @@ func (w *Worker) Run(ctx context.Context) {
 
 func runPurgeLoop(ctx context.Context, purge func(context.Context) bool) {
 	for ctx.Err() == nil {
-		delay := time.Hour
+		delay := idleDelay
 		if purge(ctx) {
-			delay = 5 * time.Minute
+			delay = backlogDelay
 		}
 		timer := time.NewTimer(delay)
 		select {
@@ -115,7 +160,12 @@ func (w *Worker) purge(ctx context.Context) bool {
 		logsCapDeleted, txCapDeleted        int64
 		profilesDeleted, profilesCapDeleted int64
 		cronCheckinsDeleted                 int64
+		behind                              bool
 	)
+	track := func(n int64, tableBehind bool) int64 {
+		behind = behind || tableBehind
+		return n
+	}
 
 	// Housekeeping does not depend on any retention setting.
 	firingsDeleted = w.purgeAlertFirings(ctx)
@@ -126,26 +176,26 @@ func (w *Worker) purge(ctx context.Context) bool {
 		cutoff := time.Now().AddDate(0, 0, -w.retentionDays)
 		slog.Info("retention: purging", "cutoff", cutoff.Format(time.DateOnly))
 
-		eventsDeleted, issuesDeleted = w.purgeEvents(ctx, cutoff)
-		txDeleted = w.purgeTransactions(ctx, cutoff)
-		logsDeleted = w.purgeLogs(ctx, cutoff)
-		uptimeChecksDeleted = w.purgeUptimeChecks(ctx, cutoff)
-		cronCheckinsDeleted = w.purgeCronCheckins(ctx, cutoff)
+		var eventsBehind bool
+		eventsDeleted, issuesDeleted, eventsBehind = w.purgeEvents(ctx, cutoff)
+		behind = behind || eventsBehind
+		txDeleted = track(w.purgeTransactions(ctx, cutoff))
+		logsDeleted = track(w.purgeLogs(ctx, cutoff))
+		uptimeChecksDeleted = track(w.purgeUptimeChecks(ctx, cutoff))
+		cronCheckinsDeleted = track(w.purgeCronCheckins(ctx, cutoff))
 	}
-	logsCapDeleted = w.purgeLogsRowCap(ctx)
-	txCapDeleted = w.purgeTransactionsRowCap(ctx)
+	logsCapDeleted = track(w.purgeLogsRowCap(ctx))
+	txCapDeleted = track(w.purgeTransactionsRowCap(ctx))
 
 	if w.profileRetentionDays > 0 {
 		profileCutoff := time.Now().AddDate(0, 0, -w.profileRetentionDays)
-		profilesDeleted = w.purgeProfiles(ctx, profileCutoff)
+		profilesDeleted = track(w.purgeProfiles(ctx, profileCutoff))
 	}
 	profilesCapDeleted = w.purgeProfilesStorageCap(ctx)
 
-	budgetReached := eventsDeleted >= maxAgeDeletesPerPass || txDeleted >= maxAgeDeletesPerPass ||
-		logsDeleted >= maxAgeDeletesPerPass || uptimeChecksDeleted >= maxAgeDeletesPerPass ||
-		profilesDeleted >= maxAgeDeletesPerPass || firingsDeleted >= maxRowCapDeletesPerPass ||
-		logsCapDeleted >= maxRowCapDeletesPerPass || txCapDeleted >= maxRowCapDeletesPerPass ||
-		profilesCapDeleted >= maxStorageCapIDsPerPass || cronCheckinsDeleted >= maxAgeDeletesPerPass
+	// The id-list purges hold at most maxIDsPerPass ids; reaching it means more
+	// are probably waiting.
+	behind = behind || firingsDeleted >= maxIDsPerPass || profilesCapDeleted >= maxIDsPerPass
 	slog.Info("retention: done",
 		"events", eventsDeleted,
 		"issues_removed", issuesDeleted,
@@ -158,36 +208,82 @@ func (w *Worker) purge(ctx context.Context) bool {
 		"tx_row_cap", txCapDeleted,
 		"profiles", profilesDeleted,
 		"profiles_storage_cap", profilesCapDeleted,
-		"budget_reached", budgetReached,
+		"behind", behind,
 	)
-	return budgetReached
+	return behind
 }
 
-// purgeProfiles deletes profiles past the profile-specific retention window,
-// with bounded batches and a per-pass row budget. A larger backlog continues
-// on the next scheduled pass. Neither limit bounds payload bytes or elapsed time.
-func (w *Worker) purgeProfiles(ctx context.Context, cutoff time.Time) int64 {
-	var total int64
-	for total < maxAgeDeletesPerPass && ctx.Err() == nil {
-		batch := min(int64(profilePurgeBatch), int64(maxAgeDeletesPerPass)-total)
-		tag, err := w.pool.Exec(ctx, `
-			DELETE FROM profile_chunks
-			WHERE id IN (
-				SELECT id FROM profile_chunks
-				WHERE received_at < $1
-				ORDER BY received_at ASC
-				LIMIT $2
-			)`, cutoff, batch)
+// drain deletes one table's rows in batches until a batch comes back short,
+// meaning the table has caught up, the table's time budget is spent, or ctx is
+// cancelled. del removes at most batch rows in one bounded statement or
+// transaction and reports how many it removed. The first batch always runs, so
+// every table makes progress in every pass. behind reports that the budget ran
+// out after a full batch, so rows are probably still waiting.
+func (w *Worker) drain(ctx context.Context, table string, maxBatch int64, del func(batch int64) (int64, error)) (total int64, behind bool, err error) {
+	start := w.now()
+	for ctx.Err() == nil {
+		if w.now().Sub(start) >= w.tableBudget {
+			return total, true, nil
+		}
+		batch := w.batchSize(table, maxBatch)
+		began := w.now()
+		n, err := del(batch)
 		if err != nil {
-			slog.Error("retention: delete profiles", "err", err)
-			return total
+			return total, false, err
 		}
-		total += tag.RowsAffected()
-		if tag.RowsAffected() < batch {
-			return total
+		total += n
+		w.adapt(table, maxBatch, batch, w.now().Sub(began))
+		if n < batch {
+			return total, false, nil
 		}
 	}
-	return total
+	return total, false, nil
+}
+
+// drainStatement drains a table with a single DELETE per batch. The batch size
+// is passed as the query's last argument.
+func (w *Worker) drainStatement(ctx context.Context, table string, maxBatch int64, msg, query string, args ...any) (int64, bool) {
+	total, behind, err := w.drain(ctx, table, maxBatch, func(batch int64) (int64, error) {
+		tag, err := w.pool.Exec(ctx, query, append(args[:len(args):len(args)], batch)...)
+		return tag.RowsAffected(), err
+	})
+	if err != nil {
+		slog.Error(msg, "table", table, "err", err)
+	}
+	return total, behind
+}
+
+func (w *Worker) batchSize(table string, maxBatch int64) int64 {
+	if n, ok := w.batchSizes[table]; ok && n <= maxBatch {
+		return n
+	}
+	return maxBatch
+}
+
+// adapt sizes the table's next batch from how long this one took, steering
+// statement duration toward batchTarget.
+func (w *Worker) adapt(table string, maxBatch, batch int64, took time.Duration) {
+	switch {
+	case took > batchTarget:
+		batch = max(minBatch, int64(float64(batch)*float64(batchTarget)/float64(took)))
+	case took < batchTarget/2:
+		batch = min(maxBatch, batch*2)
+	}
+	w.batchSizes[table] = batch
+}
+
+// purgeProfiles deletes profiles past the profile-specific retention window,
+// oldest first, within the table's time budget. A larger backlog continues on
+// the next pass.
+func (w *Worker) purgeProfiles(ctx context.Context, cutoff time.Time) (int64, bool) {
+	return w.drainStatement(ctx, "profile_chunks", agePurgeBatch, "retention: delete profiles", `
+		DELETE FROM profile_chunks
+		WHERE id IN (
+			SELECT id FROM profile_chunks
+			WHERE received_at < $1
+			ORDER BY received_at ASC
+			LIMIT $2
+		)`, cutoff)
 }
 
 // purgeProfilesStorageCap enforces the instance-wide profile storage budget,
@@ -221,7 +317,7 @@ func (w *Worker) purgeProfilesStorageCap(ctx context.Context) int64 {
 		) ranked
 		WHERE running > $1
 		ORDER BY running DESC
-		LIMIT $2`, limitBytes, maxStorageCapIDsPerPass)
+		LIMIT $2`, limitBytes, maxIDsPerPass)
 	if err != nil {
 		slog.Error("retention: rank profiles for storage cap", "err", err)
 		return 0
@@ -243,7 +339,7 @@ func (w *Worker) purgeProfilesStorageCap(ctx context.Context) int64 {
 
 	var total int64
 	for len(doomed) > 0 {
-		n := min(profilePurgeBatch, len(doomed))
+		n := min(agePurgeBatch, len(doomed))
 		tag, err := w.pool.Exec(ctx,
 			`DELETE FROM profile_chunks WHERE id = ANY($1::uuid[])`, doomed[:n])
 		if err != nil {
@@ -256,26 +352,31 @@ func (w *Worker) purgeProfilesStorageCap(ctx context.Context) int64 {
 	return total
 }
 
-func (w *Worker) purgeEvents(ctx context.Context, cutoff time.Time) (eventsDeleted, issuesDeleted int64) {
-	for eventsDeleted < maxAgeDeletesPerPass && ctx.Err() == nil {
-		batch := min(int64(agePurgeBatch), int64(maxAgeDeletesPerPass)-eventsDeleted)
+func (w *Worker) purgeEvents(ctx context.Context, cutoff time.Time) (eventsDeleted, issuesDeleted int64, behind bool) {
+	var err error
+	eventsDeleted, behind, err = w.drain(ctx, "events", agePurgeBatch, func(batch int64) (int64, error) {
 		events, issues, err := w.purgeEventBatch(ctx, cutoff, batch)
-		if err != nil {
-			slog.Error("retention: purge event batch", "err", err)
-			return
-		}
-		eventsDeleted += events
 		issuesDeleted += issues
-		if events < batch {
-			return
-		}
+		return events, err
+	})
+	if err != nil {
+		slog.Error("retention: purge event batch", "err", err)
 	}
-	return
+	return eventsDeleted, issuesDeleted, behind
 }
 
-// Keep event deletion and issue reconciliation atomic. Recount in a separate
-// statement after acquiring issue locks, so its snapshot includes grouping
-// transactions that committed while we were waiting for those locks.
+// Keep event deletion and issue reconciliation atomic. Each affected issue's
+// event_count drops by the rows this batch removed from it, which costs the
+// same however large the issue is. A full recount took seconds per batch on an
+// issue with millions of events, all while holding the row lock the grouper's
+// issue update waits on. The decrement runs in a separate statement after the
+// issue locks are taken, so it starts from the count left by grouping
+// transactions that committed while we waited, and the existence probes see
+// their events.
+//
+// Counts can drift, so whether an issue is empty is decided by probing
+// events_issue_timestamp rather than by trusting event_count. An empty issue's
+// count is reset to 0, and a resolved or ignored one is deleted.
 func (w *Worker) purgeEventBatch(ctx context.Context, cutoff time.Time, batch int64) (int64, int64, error) {
 	tx, err := w.pool.Begin(ctx)
 	if err != nil {
@@ -294,7 +395,7 @@ func (w *Worker) purgeEventBatch(ctx context.Context, cutoff time.Time, batch in
 		return 0, 0, err
 	}
 	var deleted int64
-	affected := make(map[string]struct{})
+	perIssue := make(map[string]int32)
 	for rows.Next() {
 		var id *string
 		if err := rows.Scan(&id); err != nil {
@@ -303,7 +404,7 @@ func (w *Worker) purgeEventBatch(ctx context.Context, cutoff time.Time, batch in
 		}
 		deleted++
 		if id != nil {
-			affected[*id] = struct{}{}
+			perIssue[*id]++
 		}
 	}
 	rows.Close()
@@ -311,22 +412,32 @@ func (w *Worker) purgeEventBatch(ctx context.Context, cutoff time.Time, batch in
 		return 0, 0, err
 	}
 	var issuesDeleted int64
-	if len(affected) > 0 {
-		ids := make([]string, 0, len(affected))
-		for id := range affected {
+	if len(perIssue) > 0 {
+		ids := make([]string, 0, len(perIssue))
+		for id := range perIssue {
 			ids = append(ids, id)
 		}
+		slices.Sort(ids)
+		counts := make([]int32, len(ids))
+		for i, id := range ids {
+			counts[i] = perIssue[id]
+		}
+		// Lock in id order so concurrent lockers cannot deadlock with us.
 		if _, err := tx.Exec(ctx, `SELECT id FROM issues WHERE id = ANY($1::uuid[]) ORDER BY id FOR UPDATE`, ids); err != nil {
 			return 0, 0, err
 		}
 		if _, err := tx.Exec(ctx, `UPDATE issues
-   SET event_count = (SELECT COUNT(*) FROM events WHERE events.issue_id = issues.id)
-   WHERE id = ANY($1::uuid[])`, ids); err != nil {
+   SET event_count = CASE
+       WHEN EXISTS (SELECT 1 FROM events WHERE events.issue_id = issues.id)
+       THEN GREATEST(issues.event_count - d.deleted, 0)
+       ELSE 0 END
+   FROM unnest($1::uuid[], $2::int[]) AS d(id, deleted)
+   WHERE issues.id = d.id`, ids, counts); err != nil {
 			return 0, 0, err
 		}
 		tag, err := tx.Exec(ctx, `DELETE FROM issues
-   WHERE event_count = 0 AND status IN ('resolved', 'ignored')
-   AND id = ANY($1::uuid[])`, ids)
+   WHERE id = ANY($1::uuid[]) AND status IN ('resolved', 'ignored')
+   AND NOT EXISTS (SELECT 1 FROM events WHERE events.issue_id = issues.id)`, ids)
 		if err != nil {
 			return 0, 0, err
 		}
@@ -338,69 +449,39 @@ func (w *Worker) purgeEventBatch(ctx context.Context, cutoff time.Time, batch in
 	return deleted, issuesDeleted, nil
 }
 
-// agePurgeBatch bounds parent rows, not cascaded children or elapsed time.
-const agePurgeBatch = 5000
-const maxAgeDeletesPerPass = 200_000
-
-func (w *Worker) purgeTransactions(ctx context.Context, cutoff time.Time) int64 {
-	return w.purgeByAge(ctx, "transactions", "received_at", cutoff)
+func (w *Worker) purgeTransactions(ctx context.Context, cutoff time.Time) (int64, bool) {
+	return w.purgeByAge(ctx, "transactions", "received_at", txPurgeBatch, cutoff)
 }
 
-func (w *Worker) purgeLogs(ctx context.Context, cutoff time.Time) int64 {
-	return w.purgeByAge(ctx, "logs", "received_at", cutoff)
+func (w *Worker) purgeLogs(ctx context.Context, cutoff time.Time) (int64, bool) {
+	return w.purgeByAge(ctx, "logs", "received_at", agePurgeBatch, cutoff)
 }
 
 // purgeByAge uses a cutoff fixed for the whole pass. Table and time-column
 // names come only from fixed wrappers. The corresponding time/ID index
 // supports ordered candidate selection without scanning retained rows.
-func (w *Worker) purgeByAge(ctx context.Context, table, timeColumn string, cutoff time.Time) int64 {
+func (w *Worker) purgeByAge(ctx context.Context, table, timeColumn string, maxBatch int64, cutoff time.Time) (int64, bool) {
 	query := fmt.Sprintf(`
 		DELETE FROM %s WHERE id = ANY(ARRAY(
 			SELECT id FROM %s WHERE %s < $1
 			ORDER BY %s ASC, id ASC LIMIT $2
 		))`, table, table, timeColumn, timeColumn)
-	var total int64
-	for total < maxAgeDeletesPerPass && ctx.Err() == nil {
-		batch := min(int64(agePurgeBatch), int64(maxAgeDeletesPerPass)-total)
-		tag, err := w.pool.Exec(ctx, query, cutoff, batch)
-		if err != nil {
-			slog.Error("retention: delete expired rows", "table", table, "err", err)
-			return total
-		}
-		total += tag.RowsAffected()
-		if tag.RowsAffected() < batch {
-			break
-		}
-	}
-	return total
+	return w.drainStatement(ctx, table, maxBatch, "retention: delete expired rows", query, cutoff)
 }
 
-func (w *Worker) purgeUptimeChecks(ctx context.Context, cutoff time.Time) int64 {
-	return w.purgeByAge(ctx, "uptime_checks", "checked_at", cutoff)
+func (w *Worker) purgeUptimeChecks(ctx context.Context, cutoff time.Time) (int64, bool) {
+	return w.purgeByAge(ctx, "uptime_checks", "checked_at", agePurgeBatch, cutoff)
 }
 
 // Retain running jobs regardless of age. Completion time keeps recently finished
 // long-running jobs available to alerts; legacy terminal pings use receipt time.
-func (w *Worker) purgeCronCheckins(ctx context.Context, cutoff time.Time) int64 {
-	var total int64
-	for total < maxAgeDeletesPerPass && ctx.Err() == nil {
-		batch := min(int64(agePurgeBatch), int64(maxAgeDeletesPerPass)-total)
-		tag, err := w.pool.Exec(ctx, `
-			DELETE FROM cron_checkins WHERE id = ANY(ARRAY(
-				SELECT id FROM cron_checkins
-				WHERE status IN ('ok', 'error') AND COALESCE(finished_at, received_at) < $1
-				ORDER BY COALESCE(finished_at, received_at), id LIMIT $2
-			))`, cutoff, batch)
-		if err != nil {
-			slog.Error("retention: delete completed cron check-ins", "err", err)
-			return total
-		}
-		total += tag.RowsAffected()
-		if tag.RowsAffected() < batch {
-			break
-		}
-	}
-	return total
+func (w *Worker) purgeCronCheckins(ctx context.Context, cutoff time.Time) (int64, bool) {
+	return w.drainStatement(ctx, "cron_checkins", agePurgeBatch, "retention: delete completed cron check-ins", `
+		DELETE FROM cron_checkins WHERE id = ANY(ARRAY(
+			SELECT id FROM cron_checkins
+			WHERE status IN ('ok', 'error') AND COALESCE(finished_at, received_at) < $1
+			ORDER BY COALESCE(finished_at, received_at), id LIMIT $2
+		))`, cutoff)
 }
 
 // purgeAlertFirings retains the newest 1000 firings per rule. Rank once per
@@ -415,7 +496,7 @@ func (w *Worker) purgeAlertFirings(ctx context.Context) int64 {
 				PARTITION BY rule_id ORDER BY fired_at DESC, id DESC
 			) AS rn FROM alert_firings
 		) ranked
-		WHERE rn > 1000 ORDER BY fired_at ASC, id ASC LIMIT $1`, maxRowCapDeletesPerPass)
+		WHERE rn > 1000 ORDER BY fired_at ASC, id ASC LIMIT $1`, maxIDsPerPass)
 	if err != nil {
 		slog.Error("retention: rank alert firings", "err", err)
 		return 0
@@ -427,7 +508,7 @@ func (w *Worker) purgeAlertFirings(ctx context.Context) int64 {
 	}
 	var total int64
 	for len(ids) > 0 && ctx.Err() == nil {
-		n := min(rowCapBatch, len(ids))
+		n := min(agePurgeBatch, len(ids))
 		tag, err := w.pool.Exec(ctx, `DELETE FROM alert_firings WHERE id = ANY($1::uuid[])`, ids[:n])
 		if err != nil {
 			slog.Error("retention: purge alert firings", "err", err)
@@ -440,24 +521,24 @@ func (w *Worker) purgeAlertFirings(ctx context.Context) int64 {
 }
 
 // purgeLogsRowCap enforces an instance-wide row cap on the logs table.
-// The globally oldest rows are deleted first, within the per-pass work budget.
-func (w *Worker) purgeLogsRowCap(ctx context.Context) int64 {
-	return w.purgeRowCap(ctx, "logs", "timestamp", w.logRowLimit)
+// The globally oldest rows are deleted first, within the table's time budget.
+func (w *Worker) purgeLogsRowCap(ctx context.Context) (int64, bool) {
+	return w.purgeRowCap(ctx, "logs", "timestamp", agePurgeBatch, w.logRowLimit)
 }
 
 // purgeTransactionsRowCap enforces an instance-wide row cap on transactions.
 // Deletion cascades to child spans and performance events via foreign keys.
-func (w *Worker) purgeTransactionsRowCap(ctx context.Context) int64 {
-	return w.purgeRowCap(ctx, "transactions", "start_timestamp", w.txRowLimit)
+func (w *Worker) purgeTransactionsRowCap(ctx context.Context) (int64, bool) {
+	return w.purgeRowCap(ctx, "transactions", "start_timestamp", txPurgeBatch, w.txRowLimit)
 }
 
 // purgeRowCap finds the newest excess row once, using the global time/ID index.
 // Keeping this boundary fixed avoids repeated counts and prevents concurrent
 // cleanup workers from independently subtracting the same excess row count.
 // Table and column names come only from the fixed wrappers above.
-func (w *Worker) purgeRowCap(ctx context.Context, table, timeColumn string, limit int) int64 {
+func (w *Worker) purgeRowCap(ctx context.Context, table, timeColumn string, maxBatch int64, limit int) (int64, bool) {
 	if limit <= 0 || ctx.Err() != nil {
-		return 0
+		return 0, false
 	}
 	var cutoff time.Time
 	var cutoffID string
@@ -465,31 +546,18 @@ func (w *Worker) purgeRowCap(ctx context.Context, table, timeColumn string, limi
 		SELECT %s, id FROM %s ORDER BY %s DESC, id DESC OFFSET $1 LIMIT 1`,
 		timeColumn, table, timeColumn), limit).Scan(&cutoff, &cutoffID)
 	if errors.Is(err, pgx.ErrNoRows) {
-		return 0
+		return 0, false
 	}
 	if err != nil {
 		slog.Error("retention: find row cap boundary", "table", table, "err", err)
-		return 0
+		return 0, false
 	}
 	query := fmt.Sprintf(`
 		DELETE FROM %s WHERE id IN (
 			SELECT id FROM %s WHERE (%s, id) <= ($1, $2::uuid)
 			ORDER BY %s ASC, id ASC LIMIT $3
 		)`, table, table, timeColumn, timeColumn)
-	var total int64
-	for total < maxRowCapDeletesPerPass && ctx.Err() == nil {
-		batch := min(int64(rowCapBatch), int64(maxRowCapDeletesPerPass)-total)
-		tag, err := w.pool.Exec(ctx, query, cutoff, cutoffID, batch)
-		if err != nil {
-			slog.Error("retention: purge row cap", "table", table, "err", err)
-			return total
-		}
-		total += tag.RowsAffected()
-		if tag.RowsAffected() < batch {
-			break
-		}
-	}
-	return total
+	return w.drainStatement(ctx, table, maxBatch, "retention: purge row cap", query, cutoff, cutoffID)
 }
 
 // expiredInviteGrace keeps expired invites listed for a while, so admins can
@@ -521,7 +589,7 @@ func (w *Worker) purgeExpiredAuthTokens(ctx context.Context) {
 // Table names come only from purgeExpiredAuthTokens. These small auth tables
 // need no additional index just to bound deletion work.
 func (w *Worker) purgeExpiredTokenTable(ctx context.Context, table string, cutoff time.Time) {
-	rows, err := w.pool.Query(ctx, "SELECT token_hash FROM "+table+" WHERE expires_at < $1 LIMIT $2", cutoff, maxAgeDeletesPerPass)
+	rows, err := w.pool.Query(ctx, "SELECT token_hash FROM "+table+" WHERE expires_at < $1 LIMIT $2", cutoff, maxIDsPerPass)
 	if err != nil {
 		slog.Error("retention: select expired tokens", "table", table, "err", err)
 		return
diff --git a/internal/storage/spans.go b/internal/storage/spans.go
index a2614526..450a0933 100644
--- a/internal/storage/spans.go
+++ b/internal/storage/spans.go
@@ -3,6 +3,7 @@ package storage
 import (
 	"context"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/jackc/pgx/v5/pgxpool"
@@ -31,6 +32,12 @@ type SpanTimeseries struct {
 	BucketSize string       `json:"bucket_size"`
 }
 
+// Each span page (database, cache, jobs) aggregates one category of ops, and
+// each category has its own partial index (spans_db_start, spans_cache_start,
+// spans_job_start; migration 0037) holding only those spans. A query can use
+// one only if it implies the index predicate, so these filters must match the
+// migration text arm for arm; a different spelling (ILIKE, a bind parameter)
+// cannot be proven and falls back to reading every span.
 func spanOpFilter(category string) string {
 	switch category {
 	case "db":
@@ -40,7 +47,36 @@ func spanOpFilter(category string) string {
 	case "job":
 		return "(s.op LIKE 'task%' OR s.op LIKE 'queue%' OR s.op LIKE 'celery%' OR s.op LIKE 'job%')"
 	}
-	return "TRUE"
+	// Every caller passes one of the categories above. Anything else matches
+	// nothing rather than scanning all spans without an index.
+	return "FALSE"
+}
+
+// spanCategory returns the span page category of op, or "" when no page
+// shows it. Prefixes match the LIKE filters above (both case sensitive).
+func spanCategory(op string) string {
+	switch {
+	case strings.HasPrefix(op, "db"):
+		return "db"
+	case strings.HasPrefix(op, "cache"):
+		return "cache"
+	case strings.HasPrefix(op, "task"), strings.HasPrefix(op, "queue"),
+		strings.HasPrefix(op, "celery"), strings.HasPrefix(op, "job"):
+		return "job"
+	}
+	return ""
+}
+
+// spanProjectScope adds the project condition. Without a project filter it
+// lists every project, so the leading project_id column of the category
+// indexes is one index range per project on any Postgres version; before 18
+// there is no skip scan, and the whole index would be read.
+func spanProjectScope(args []any, projectIDs []string) ([]any, string) {
+	if len(projectIDs) > 0 {
+		args = append(args, projectIDs)
+		return args, fmt.Sprintf(" AND s.project_id = ANY($%d::uuid[])", len(args))
+	}
+	return args, " AND s.project_id = ANY(ARRAY(SELECT id FROM projects))"
 }
 
 func GetSpanSummaries(ctx context.Context, pool *pgxpool.Pool, category string, projectIDs []string, hours int, env, release string, userIdentity ...string) ([]*SpanSummary, error) {
@@ -51,10 +87,8 @@ func GetSpanSummaries(ctx context.Context, pool *pgxpool.Pool, category string,
 	window := ResolveTimeRange(ctx, hours, 0)
 	args := []any{window.From, window.To}
 	where := fmt.Sprintf(`WHERE %s AND s.start_timestamp >= $1 AND s.start_timestamp < $2`, spanOpFilter(category))
-	if len(projectIDs) > 0 {
-		args = append(args, projectIDs)
-		where += " AND s.project_id = ANY($3::uuid[])"
-	}
+	args, scope := spanProjectScope(args, projectIDs)
+	where += scope
 
 	if env != "" {
 		args = append(args, env)
@@ -152,10 +186,8 @@ func GetSpanTimeseries(ctx context.Context, pool *pgxpool.Pool, category string,
 	window := ResolveTimeRange(ctx, hours, 0)
 	args := []any{window.From, window.To}
 	where := fmt.Sprintf(`WHERE %s AND s.start_timestamp >= $1 AND s.start_timestamp < $2`, spanOpFilter(category))
-	if len(projectIDs) > 0 {
-		args = append(args, projectIDs)
-		where += " AND s.project_id = ANY($3::uuid[])"
-	}
+	args, scope := spanProjectScope(args, projectIDs)
+	where += scope
 
 	if env != "" {
 		args = append(args, env)
@@ -213,17 +245,20 @@ func GetSpanSamples(ctx context.Context, pool *pgxpool.Pool, op, description str
 	if hours <= 0 || hours > 2160 {
 		hours = 24
 	}
-	if projectIDs == nil {
-		projectIDs = []string{}
-	}
-
 	window := ResolveTimeRange(ctx, hours, 0)
-	args := []any{window.From, projectIDs, op, description, window.To}
+	args := []any{window.From, window.To, op, description}
 	where := `
 		WHERE s.op = $3
 		  AND COALESCE(s.description, '') = $4
-		  AND s.start_timestamp >= $1 AND s.start_timestamp < $5
-		  AND (CARDINALITY($2::uuid[]) = 0 OR s.project_id = ANY($2::uuid[]))`
+		  AND s.start_timestamp >= $1 AND s.start_timestamp < $2`
+	// Samples open from a span page row. Adding that page's filter lets the
+	// planner use its category index; s.op = $3 alone can't prove it. Other
+	// ops still work, without an index.
+	if category := spanCategory(op); category != "" {
+		where += " AND " + spanOpFilter(category)
+	}
+	args, scope := spanProjectScope(args, projectIDs)
+	where += scope
 
 	if env != "" {
 		args = append(args, env)
diff --git a/internal/storage/spans_index_test.go b/internal/storage/spans_index_test.go
new file mode 100644
index 00000000..0379a1bb
--- /dev/null
+++ b/internal/storage/spans_index_test.go
@@ -0,0 +1,167 @@
+package storage_test
+
+import (
+	"context"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/stretchr/testify/require"
+
+	"github.com/blendbyte/tindra/internal/storage"
+)
+
+// spanQueryRecorder keeps the last statement that read from spans, so a test
+// can EXPLAIN exactly what a storage function sent.
+type spanQueryRecorder struct {
+	mu   sync.Mutex
+	sql  string
+	args []any
+}
+
+func (r *spanQueryRecorder) TraceQueryStart(ctx context.Context, _ *pgx.Conn, data pgx.TraceQueryStartData) context.Context {
+	if strings.Contains(data.SQL, "FROM spans") {
+		r.mu.Lock()
+		r.sql, r.args = data.SQL, data.Args
+		r.mu.Unlock()
+	}
+	return ctx
+}
+
+func (r *spanQueryRecorder) TraceQueryEnd(context.Context, *pgx.Conn, pgx.TraceQueryEndData) {}
+
+func (r *spanQueryRecorder) last() (string, []any) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return r.sql, r.args
+}
+
+// explainWithoutSeqScan plans the statement with sequential scans disabled. If
+// the planner can prove the partial index predicate it picks the index;
+// otherwise the plan names some other access path.
+func explainWithoutSeqScan(t *testing.T, sql string, args []any) string {
+	t.Helper()
+	ctx := context.Background()
+	conn, err := testPool.Acquire(ctx)
+	require.NoError(t, err)
+	defer conn.Release()
+	_, err = conn.Exec(ctx, "SET enable_seqscan = off")
+	require.NoError(t, err)
+	defer conn.Exec(ctx, "RESET enable_seqscan") //nolint:errcheck
+	rows, err := conn.Query(ctx, "EXPLAIN "+sql, args...)
+	require.NoError(t, err)
+	defer rows.Close()
+	var plan strings.Builder
+	for rows.Next() {
+		var line string
+		require.NoError(t, rows.Scan(&line))
+		plan.WriteString(line + "\n")
+	}
+	require.NoError(t, rows.Err())
+	return plan.String()
+}
+
+// Every span page query (summaries, chart, samples), with and without a
+// project filter, must be able to use its category's partial index. The index
+// predicate has to be implied by the query text, so a reworded filter would
+// silently fall back to reading every span.
+func TestSpanPageQueriesUseCategoryIndexes(t *testing.T) {
+	p := setupProjectForSpans(t)
+	ctx := context.Background()
+	now := time.Now().UTC()
+	tx := seedTransaction(t, p.ID, "/plan", 100, now)
+	seedSpan(t, tx.ID, "plan-db", "db.query", "SELECT 1", 5, now)
+
+	rec := &spanQueryRecorder{}
+	cfg := testPool.Config()
+	cfg.ConnConfig.Tracer = rec
+	traced, err := pgxpool.NewWithConfig(ctx, cfg)
+	require.NoError(t, err)
+	defer traced.Close()
+
+	summaries := func(category string) func([]string) error {
+		return func(ps []string) error {
+			_, err := storage.GetSpanSummaries(ctx, traced, category, ps, 24, "production", "1.0")
+			return err
+		}
+	}
+	samples := func(op string) func([]string) error {
+		return func(ps []string) error {
+			_, err := storage.GetSpanSamples(ctx, traced, op, "SELECT 1", ps, 24, "", "")
+			return err
+		}
+	}
+	cases := []struct {
+		name, index string
+		run         func([]string) error
+	}{
+		{"summaries db", "spans_db_start", summaries("db")},
+		{"summaries cache", "spans_cache_start", summaries("cache")},
+		{"summaries job", "spans_job_start", summaries("job")},
+		{"timeseries cache", "spans_cache_start", func(ps []string) error {
+			_, err := storage.GetSpanTimeseries(ctx, traced, "cache", ps, 24, "", "")
+			return err
+		}},
+		{"samples db op", "spans_db_start", samples("db.query")},
+		{"samples cache op", "spans_cache_start", samples("cache.get")},
+		{"samples queue op", "spans_job_start", samples("queue.process")},
+		{"samples celery op", "spans_job_start", samples("celery.task")},
+	}
+	for _, c := range cases {
+		for _, projects := range [][]string{{p.ID}, nil} {
+			require.NoError(t, c.run(projects), c.name)
+			sql, args := rec.last()
+			require.NotEmpty(t, sql, c.name)
+			plan := explainWithoutSeqScan(t, sql, args)
+			require.Contains(t, plan, c.index, "%s (projects %v) plan:\n%s", c.name, projects, plan)
+		}
+	}
+}
+
+// Ops outside the span pages are not in the partial index, but samples for
+// them still come back, and so do samples across all projects.
+func TestGetSpanSamplesOutsideCategoryIndex(t *testing.T) {
+	p := setupProjectForSpans(t)
+	ctx := context.Background()
+	now := time.Now().UTC()
+	tx := seedTransaction(t, p.ID, "/views", 100, now)
+	seedSpan(t, tx.ID, "view-1", "view.render", "home.blade.php", 7, now)
+	seedSpan(t, tx.ID, "db-1", "db.query", "SELECT 1", 3, now)
+
+	views, err := storage.GetSpanSamples(ctx, testPool, "view.render", "home.blade.php", nil, 24, "", "")
+	require.NoError(t, err)
+	require.Len(t, views, 1)
+	require.Equal(t, "view-1", views[0].SpanID)
+
+	dbAll, err := storage.GetSpanSamples(ctx, testPool, "db.query", "SELECT 1", nil, 24, "", "")
+	require.NoError(t, err)
+	require.Len(t, dbAll, 1)
+	dbOne, err := storage.GetSpanSamples(ctx, testPool, "db.query", "SELECT 1", []string{p.ID}, 24, "", "")
+	require.NoError(t, err)
+	require.Len(t, dbOne, 1)
+	none, err := storage.GetSpanSamples(ctx, testPool, "db.query", "SELECT 1", []string{"00000000-0000-0000-0000-000000000000"}, 24, "", "")
+	require.NoError(t, err)
+	require.Empty(t, none)
+}
+
+// Without a project filter the summaries cover every project.
+func TestGetSpanSummariesAcrossAllProjects(t *testing.T) {
+	p := setupProjectForSpans(t)
+	ctx := context.Background()
+	other, err := storage.CreateProject(ctx, testPool, "spans-other", "Spans Other")
+	require.NoError(t, err)
+	now := time.Now().UTC()
+	seedSpan(t, seedTransaction(t, p.ID, "/a", 100, now).ID, "a-db", "db.query", "SELECT a", 5, now)
+	seedSpan(t, seedTransaction(t, other.ID, "/b", 100, now).ID, "b-db", "db.query", "SELECT b", 5, now)
+
+	all, err := storage.GetSpanSummaries(ctx, testPool, "db", nil, 24, "", "")
+	require.NoError(t, err)
+	descriptions := []string{}
+	for _, s := range all {
+		descriptions = append(descriptions, s.Description)
+	}
+	require.ElementsMatch(t, []string{"SELECT a", "SELECT b"}, descriptions)
+}
diff --git a/internal/storage/spans_test.go b/internal/storage/spans_test.go
index 5b331760..13be8089 100644
--- a/internal/storage/spans_test.go
+++ b/internal/storage/spans_test.go
@@ -294,7 +294,7 @@ func TestGetSpanSummaries_jobCategory(t *testing.T) {
 	}
 }
 
-// TestGetSpanSummaries_unknownCategory verifies that an unknown category uses the default filter (TRUE).
+// TestGetSpanSummaries_unknownCategory verifies that an unknown category falls back to the span page ops.
 func TestGetSpanSummaries_unknownCategory(t *testing.T) {
 	p := setupProjectForSpans(t)
 	now := time.Now().UTC()
@@ -313,8 +313,8 @@ func TestGetSpanSummaries_unknownCategory(t *testing.T) {
 			break
 		}
 	}
-	if !found {
-		t.Error("expected custom.thing span visible with unknown category (default filter = TRUE)")
+	if found {
+		t.Error("an unknown category must stay within the span page ops, not show custom.thing")
 	}
 }
 
diff --git a/internal/storage/transaction_counts_performance_test.go b/internal/storage/transaction_counts_performance_test.go
index ddd01ab6..a5d743c8 100644
--- a/internal/storage/transaction_counts_performance_test.go
+++ b/internal/storage/transaction_counts_performance_test.go
@@ -26,8 +26,8 @@ func TestTransactionCountsPerformance(t *testing.T) {
 	ctx := context.Background()
 	p, err := storage.CreateProject(ctx, testPool, "count-perf", "Count performance")
 	require.NoError(t, err)
-	_, err = testPool.Exec(ctx, `INSERT INTO transactions(project_id,transaction,duration_ms,start_timestamp,timestamp)
- SELECT $1,'/heatmap',(i%10000)::int,NOW()-(i%10000)*interval '1 minute',NOW() FROM generate_series(1,$2::int) i`, p.ID, n)
+	_, err = testPool.Exec(ctx, `INSERT INTO transactions(project_id,transaction,duration_ms,start_timestamp,timestamp,received_at)
+ SELECT $1,'/heatmap',(i%10000)::int,NOW()-(i%10000)*interval '1 minute',NOW(),NOW()-(i%10000)*interval '1 minute' FROM generate_series(1,$2::int) i`, p.ID, n)
 	require.NoError(t, err)
 	_, err = testPool.Exec(ctx, "VACUUM ANALYZE transactions")
 	require.NoError(t, err)
diff --git a/internal/storage/transaction_counts_test.go b/internal/storage/transaction_counts_test.go
index ff7055ab..6be327a3 100644
--- a/internal/storage/transaction_counts_test.go
+++ b/internal/storage/transaction_counts_test.go
@@ -21,7 +21,7 @@ func TestTransactionCountsMatchTimeseries(t *testing.T) {
 		for _, env := range []string{"prod", "dev"} {
 			for _, name := range []string{"/one", "/two"} {
 				tx := seedTransaction(t, project, name, 123, time.Now().Add(-10*time.Minute))
-				_, err := testPool.Exec(ctx, "UPDATE transactions SET environment=$2,op='http.server',user_identity='alice' WHERE id=$1", tx.ID, env)
+				_, err := testPool.Exec(ctx, "UPDATE transactions SET environment=$2,op='http.server',user_identity='alice',received_at=start_timestamp WHERE id=$1", tx.ID, env)
 				require.NoError(t, err)
 			}
 		}
@@ -56,3 +56,111 @@ func TestTransactionCountsMatchTimeseries(t *testing.T) {
 	require.NotNil(t, empty.Buckets)
 	require.Empty(t, empty.Buckets)
 }
+
+func seedCountedTransaction(t *testing.T, projectID, env string, start, received time.Time) {
+	t.Helper()
+	_, err := testPool.Exec(context.Background(), `INSERT INTO transactions(project_id,transaction,duration_ms,start_timestamp,timestamp,received_at,environment)
+ VALUES ($1,'/counted',5,$2,$2,$3,NULLIF($4,''))`, projectID, start, received, env)
+	require.NoError(t, err)
+}
+
+func requireSameCounts(t *testing.T, want *storage.TxTimeseries, got *storage.TxCountTimeseries) {
+	t.Helper()
+	require.Equal(t, want.BucketSize, got.BucketSize)
+	require.Len(t, got.Buckets, len(want.Buckets))
+	for i, b := range got.Buckets {
+		require.True(t, want.Buckets[i].Time.Equal(b.Time), "bucket %d: %s != %s", i, want.Buckets[i].Time, b.Time)
+		require.Equal(t, want.Buckets[i].Count, b.Count)
+	}
+}
+
+func TestTransactionCountsUsageMatchesRawBuckets(t *testing.T) {
+	truncateProjects(t)
+	ctx := context.Background()
+	p, err := storage.CreateProject(ctx, testPool, "usage-counts", "Usage counts")
+	require.NoError(t, err)
+	other, err := storage.CreateProject(ctx, testPool, "usage-counts-other", "Other")
+	require.NoError(t, err)
+	now := time.Now().UTC()
+	// Received in the same hour each started, so both bucketings agree. Ages
+	// cover the partial first hour, complete hours and the partial current hour.
+	for i, age := range []time.Duration{167*time.Hour + 50*time.Minute, 167*time.Hour + 50*time.Minute, 100 * time.Hour, 30 * time.Hour, 3 * time.Hour, 3 * time.Hour, time.Minute} {
+		start := now.Add(-age)
+		seedCountedTransaction(t, p.ID, "prod", start, start)
+		if i%2 == 0 {
+			seedCountedTransaction(t, other.ID, "", start, start)
+		}
+	}
+	// Older than the window: neither path may count it.
+	seedCountedTransaction(t, p.ID, "prod", now.Add(-170*time.Hour), now.Add(-170*time.Hour))
+	for _, ids := range [][]string{nil, {p.ID}, {other.ID}, {p.ID, other.ID}} {
+		raw, err := storage.GetTransactionTimeseries(ctx, testPool, ids, 168, "", "", "", "")
+		require.NoError(t, err)
+		counts, err := storage.GetTransactionCounts(ctx, testPool, ids, 168, "", "", "", "")
+		require.NoError(t, err)
+		requireSameCounts(t, raw, counts)
+		require.NotEmpty(t, counts.Buckets)
+	}
+	unknown, err := storage.GetTransactionCounts(ctx, testPool, []string{"00000000-0000-0000-0000-000000000001"}, 168, "", "", "", "")
+	require.NoError(t, err)
+	require.NotNil(t, unknown.Buckets)
+	require.Empty(t, unknown.Buckets)
+}
+
+func TestTransactionCountsUsageExactWindowEdges(t *testing.T) {
+	truncateProjects(t)
+	ctx := context.Background()
+	p, err := storage.CreateProject(ctx, testPool, "usage-edges", "Usage edges")
+	require.NoError(t, err)
+	hour := time.Now().UTC().Truncate(time.Hour).Add(-48 * time.Hour)
+	for _, offset := range []time.Duration{10 * time.Minute, 40 * time.Minute, 2*time.Hour + 5*time.Minute, 3 * time.Hour, 5*time.Hour + 10*time.Minute, 5*time.Hour + 20*time.Minute} {
+		seedCountedTransaction(t, p.ID, "prod", hour.Add(offset), hour.Add(offset))
+	}
+	for _, window := range []storage.TimeRange{
+		{From: hour.Add(30 * time.Minute), To: hour.Add(5*time.Hour + 15*time.Minute)},
+		{From: hour, To: hour.Add(5 * time.Hour)},
+		{From: hour.Add(5 * time.Minute), To: hour.Add(50 * time.Minute)},
+		{From: hour.Add(3 * time.Hour), To: hour.Add(3*time.Hour + time.Second)},
+	} {
+		scoped := storage.WithTimeRange(ctx, window)
+		raw, err := storage.GetTransactionTimeseries(scoped, testPool, nil, 168, "", "", "", "")
+		require.NoError(t, err)
+		counts, err := storage.GetTransactionCounts(scoped, testPool, nil, 168, "", "", "", "")
+		require.NoError(t, err)
+		requireSameCounts(t, raw, counts)
+		require.NotEmpty(t, counts.Buckets)
+	}
+}
+
+func TestTransactionCountsBucketByReceivedUnlessFiltered(t *testing.T) {
+	truncateProjects(t)
+	ctx := context.Background()
+	p, err := storage.CreateProject(ctx, testPool, "usage-received", "Usage received")
+	require.NoError(t, err)
+	started := time.Now().UTC().Truncate(time.Hour).Add(-5*time.Hour + 50*time.Minute)
+	received := started.Add(20 * time.Minute)
+	seedCountedTransaction(t, p.ID, "prod", started, received)
+	_, err = testPool.Exec(ctx, "UPDATE transactions SET op='http.server',user_identity='alice',release='1.0' WHERE project_id=$1", p.ID)
+	require.NoError(t, err)
+
+	counts, err := storage.GetTransactionCounts(ctx, testPool, []string{p.ID}, 168, "", "", "", "")
+	require.NoError(t, err)
+	require.Len(t, counts.Buckets, 1)
+	require.True(t, received.Truncate(time.Hour).Equal(counts.Buckets[0].Time))
+
+	type filter struct{ env, name, op, user, release string }
+	for _, f := range []filter{{env: "prod"}, {name: "/counted"}, {op: "http.server"}, {user: "alice"}, {release: "1.0"}} {
+		counts, err := storage.GetTransactionCounts(ctx, testPool, []string{p.ID}, 168, f.env, f.name, f.op, f.user, f.release)
+		require.NoError(t, err)
+		require.Len(t, counts.Buckets, 1, "%+v", f)
+		require.True(t, started.Truncate(time.Hour).Equal(counts.Buckets[0].Time), "%+v", f)
+	}
+	// Minute and day buckets cannot come from hourly counters.
+	for _, hours := range []int{1, 720} {
+		raw, err := storage.GetTransactionTimeseries(ctx, testPool, []string{p.ID}, hours, "", "", "", "")
+		require.NoError(t, err)
+		counts, err := storage.GetTransactionCounts(ctx, testPool, []string{p.ID}, hours, "", "", "", "")
+		require.NoError(t, err)
+		requireSameCounts(t, raw, counts)
+	}
+}
diff --git a/internal/storage/transactions.go b/internal/storage/transactions.go
index 759d0dcb..9a1325d9 100644
--- a/internal/storage/transactions.go
+++ b/internal/storage/transactions.go
@@ -329,7 +329,12 @@ type TxCountTimeseries struct {
 }
 
 func GetTransactionCounts(ctx context.Context, pool *pgxpool.Pool, projectIDs []string, hours int, env, name, op, userIdentity string, release ...string) (*TxCountTimeseries, error) {
-	q, args, bucketSize := transactionTimeseriesQuery(projectIDs, hours, env, name, op, userIdentity, true, ResolveTimeRange(ctx, hours, 0), release)
+	window := ResolveTimeRange(ctx, hours, 0)
+	q, args, bucketSize := transactionTimeseriesQuery(projectIDs, hours, env, name, op, userIdentity, true, window, release)
+	filtered := env != "" || name != "" || op != "" || userIdentity != "" || (len(release) > 0 && release[0] != "")
+	if bucketSize == "hour" && !filtered {
+		q, args = transactionUsageCountsQuery(projectIDs, window)
+	}
 	rows, err := pool.Query(ctx, q, args...)
 	if err != nil {
 		return nil, fmt.Errorf("query: %w", err)
@@ -346,6 +351,48 @@ func GetTransactionCounts(ctx context.Context, pool *pgxpool.Pool, projectIDs []
 	return result, rows.Err()
 }
 
+// transactionUsageCountsQuery reads unfiltered hourly counts from the
+// telemetry_usage counters instead of aggregating raw transactions. The
+// counters bucket by received_at, not start_timestamp, so a transaction that
+// started in one hour and arrived in the next lands in the later hour. Complete
+// UTC hours come from the counters; the partial hours at either edge of the
+// window are counted from raw rows so the bounds stay exact.
+func transactionUsageCountsQuery(projectIDs []string, window TimeRange) (string, []any) {
+	fullFrom := window.From.Truncate(time.Hour)
+	if fullFrom.Before(window.From) {
+		fullFrom = fullFrom.Add(time.Hour)
+	}
+	fullTo := window.To.Truncate(time.Hour)
+	if fullTo.Before(fullFrom) {
+		fullTo = fullFrom
+	}
+	headTo := fullFrom
+	if window.To.Before(headTo) {
+		headTo = window.To
+	}
+	args := []any{window.From, headTo, fullFrom, fullTo, window.To}
+	project := ""
+	if len(projectIDs) > 0 {
+		args = append(args, projectIDs)
+		project = " AND project_id = ANY($6::uuid[])"
+	}
+	q := `
+		SELECT bucket, SUM(n)::bigint AS count FROM (
+			SELECT bucket, n FROM telemetry_usage
+			WHERE kind = 'transactions' AND bucket >= $3 AND bucket < $4` + project + `
+			UNION ALL
+			SELECT date_trunc('hour', received_at, 'UTC'), COUNT(*) FROM transactions
+			WHERE received_at >= $1 AND received_at < $2` + project + ` GROUP BY 1
+			UNION ALL
+			SELECT date_trunc('hour', received_at, 'UTC'), COUNT(*) FROM transactions
+			WHERE received_at >= $4 AND received_at < $5` + project + ` GROUP BY 1
+		) usage(bucket, n)
+		GROUP BY bucket
+		ORDER BY bucket ASC
+	`
+	return q, args
+}
+
 func transactionTimeseriesQuery(projectIDs []string, hours int, env, name, op, userIdentity string, countsOnly bool, window TimeRange, releases []string) (string, []any, string) {
 
 	if hours <= 0 || hours > 2160 {
diff --git a/internal/storage/uptime.go b/internal/storage/uptime.go
index 01d3e3a9..2f1d57bf 100644
--- a/internal/storage/uptime.go
+++ b/internal/storage/uptime.go
@@ -367,7 +367,8 @@ func ListUptimeChecks(ctx context.Context, pool *pgxpool.Pool, monitorID string,
 }
 
 // GetUptimeStats returns aggregate uptime percentages and average response
-// time over 24-hour, 7-day, and 30-day windows.
+// time over 24-hour, 7-day, and 30-day windows. Every figure lies within the
+// 30-day window, so older checks kept for retention are never read.
 func GetUptimeStats(ctx context.Context, pool *pgxpool.Pool, monitorID string) (*UptimeStats, error) {
 	var s UptimeStats
 	var avgMs *float64
@@ -386,7 +387,7 @@ func GetUptimeStats(ctx context.Context, pool *pgxpool.Pool, monitorID string) (
                 / NULLIF(SUM(CASE WHEN checked_at > NOW() - INTERVAL '30 days' THEN 1 ELSE 0 END)::float, 0),
             0),
             AVG(CASE WHEN checked_at > NOW() - INTERVAL '24 hours' AND status='up' THEN response_ms ELSE NULL END)
-        FROM uptime_checks WHERE monitor_id = $1`, monitorID,
+        FROM uptime_checks WHERE monitor_id = $1 AND checked_at > NOW() - INTERVAL '30 days'`, monitorID,
 	).Scan(&s.UptimePct24h, &s.UptimePct7d, &s.UptimePct30d, &avgMs)
 	if err != nil {
 		return nil, fmt.Errorf("query: %w", err)
diff --git a/internal/storage/uptime_test.go b/internal/storage/uptime_test.go
index f7c50407..93822684 100644
--- a/internal/storage/uptime_test.go
+++ b/internal/storage/uptime_test.go
@@ -622,6 +622,43 @@ func TestGetUptimeStats_halfUp(t *testing.T) {
 	}
 }
 
+func TestGetUptimeStats_ignoresChecksOlderThanThirtyDays(t *testing.T) {
+	p := setupProjectForUptime(t)
+	m := seedUptimeMonitor(t, p.ID, "stats-windows", "https://example.com")
+	ctx := context.Background()
+	insert := func(status string, responseMs int, age string) {
+		t.Helper()
+		_, err := testPool.Exec(ctx, `INSERT INTO uptime_checks (monitor_id, status, response_ms, checked_at)
+			VALUES ($1, $2, $3, NOW() - $4::interval)`, m.ID, status, responseMs, age)
+		require.NoError(t, err)
+	}
+	// 24h: 1 of 2 up; 7d: 1 of 4 up; 30d: 5 of 8 up.
+	insert("up", 100, "1 hour")
+	insert("down", 900, "2 hours")
+	insert("down", 900, "2 days")
+	insert("down", 900, "6 days")
+	for _, age := range []string{"8 days", "20 days", "29 days", "29 days 23 hours"} {
+		insert("up", 500, age)
+	}
+	want := storage.UptimeStats{UptimePct24h: 50, UptimePct7d: 25, UptimePct30d: 62.5}
+	check := func() {
+		t.Helper()
+		s, err := storage.GetUptimeStats(ctx, testPool, m.ID)
+		require.NoError(t, err)
+		require.Equal(t, want.UptimePct24h, s.UptimePct24h)
+		require.Equal(t, want.UptimePct7d, s.UptimePct7d)
+		require.Equal(t, want.UptimePct30d, s.UptimePct30d)
+		require.NotNil(t, s.AvgResponseMs)
+		require.Equal(t, 100, *s.AvgResponseMs)
+	}
+	check()
+	for _, age := range []string{"30 days 1 hour", "45 days", "89 days"} {
+		insert("down", 5000, age)
+		insert("up", 5000, age)
+	}
+	check()
+}
+
 // ---------- GetDueUptimeMonitors ----------
 
 func TestGetDueUptimeMonitors_neverChecked(t *testing.T) {
diff --git a/migrations/0037_spans_category_index.up.sql b/migrations/0037_spans_category_index.up.sql
new file mode 100644
index 00000000..00e78d87
--- /dev/null
+++ b/migrations/0037_spans_category_index.up.sql
@@ -0,0 +1,56 @@
+-- The span pages (database, cache, jobs) each aggregate one category of ops,
+-- and on real installs those are a small share of all spans (about 20% on a
+-- large Laravel install, where view spans alone are 68%). One small partial
+-- index per category replaces three full-width ones that every span insert and
+-- retention delete had to maintain. Each span lands in at most one of them.
+--
+-- The predicates must match spanOpFilter in internal/storage/spans.go arm for
+-- arm, or the planner cannot use them.
+--
+-- On a large install each build reads the whole spans table twice, and startup
+-- waits until the migration is done. They can be run by hand beforehand (the
+-- statements below, verbatim), without blocking ingest; the migration then
+-- skips them. If a build was interrupted, drop the INVALID index first, e.g.
+-- DROP INDEX CONCURRENTLY spans_db_start;
+CREATE INDEX CONCURRENTLY IF NOT EXISTS spans_db_start
+    ON spans (project_id, start_timestamp DESC, op text_pattern_ops)
+    WHERE op LIKE 'db%';
+
+-- tindra:next-batch
+
+CREATE INDEX CONCURRENTLY IF NOT EXISTS spans_cache_start
+    ON spans (project_id, start_timestamp DESC, op text_pattern_ops)
+    WHERE op LIKE 'cache%';
+
+-- tindra:next-batch
+
+CREATE INDEX CONCURRENTLY IF NOT EXISTS spans_job_start
+    ON spans (project_id, start_timestamp DESC, op text_pattern_ops)
+    WHERE op LIKE 'task%' OR op LIKE 'queue%' OR op LIKE 'celery%' OR op LIKE 'job%';
+
+-- tindra:next-batch
+
+-- Replaced by the category indexes above.
+DROP INDEX CONCURRENTLY IF EXISTS spans_proj_start_op;
+
+-- tindra:next-batch
+
+-- Replaced by the category indexes above.
+DROP INDEX CONCURRENTLY IF EXISTS spans_op_start;
+
+-- tindra:next-batch
+
+-- Served an all-ops span view that no page or API has.
+DROP INDEX CONCURRENTLY IF EXISTS spans_start_op;
+
+-- tindra:next-batch
+
+-- Duplicates transactions_pkey. Its main reader was the foreign key check on
+-- each span insert, which the much smaller primary key serves as well.
+DROP INDEX CONCURRENTLY IF EXISTS transactions_id_covering;
+
+-- tindra:next-batch
+
+-- transactions_project_perf leads with the same columns and serves the same
+-- reads; on a large install this one was never scanned.
+DROP INDEX CONCURRENTLY IF EXISTS transactions_project_start;