perf: cut span index size and let retention keep up on large installs

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

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

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

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
 	}

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
 	}

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 {

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

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
 		}

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