fix(list): use row-constructor cursor + restore per-phase publish timings + enable pg_stat_statements (#1215)

## Summary

Three changes from the 2026-04-27 incident:

1. **Cursor SQL fix** — `addCursorCondition` rewritten to use a
row-constructor comparison `(server_name, version) > ($1, $2)` instead
of the OR-decomposed form. Postgres can index-seek on the row
constructor; it can't on the OR form, which scanned from the start of
the index and filtered rows before the cursor. Cost grew linearly with
cursor depth.
2. **Per-phase publish timings restored** — `createServerInTransaction`
was simplified during #1211 review to log only `validate_ms`. The
incident showed publishes spending 50+ s in `acquire_lock` /
`version_checks` / `db_create` while `validate_ms` reported a few
hundred ms — the diagnostic signal was hidden. Restored timings for
every phase, refactored into a small `runPhase` helper.
3. **`pg_stat_statements` enabled** in the CNPG cluster spec. We had no
aggregate query visibility during the incident — could only EXPLAIN
queries we happened to suspect.

## Cursor fix evidence

Local benchmark, 100K rows, cold cache:

| Form | Rows filtered | Buffer reads | Time |
|------|--------------:|-------------:|-----:|
| OLD | 80,001 | 7,679 | 31.6 ms |
| NEW | 2 | 1 | **0.05 ms** |

Maps to prod's measured 8,911 buffer hits → 760 ms. End-to-end API:
`/v0/servers?limit=100&cursor=…` returns in 4–7 ms regardless of depth.

Risk check: confirmed `(server_name, version)` index exists in that
column order (both `servers_pkey` and `idx_servers_name_version`), and
both columns are `NOT NULL` so row-constructor comparison can't silently
drop rows.

## Tests

- New
`TestPostgreSQL_PerformanceScenarios/compound_cursor_across_versions_of_same_server`
pins multi-version pagination semantics. The existing cursor tests only
exercised the fallback (single-component cursor) and degenerate (one
version per server) cases.
- `make lint` clean, `go test -race ./internal/... ./cmd/...` green.

## Deployment

The cursor + slog changes are zero-downtime. The CNPG spec change
triggers a brief PG restart (single-instance cluster). After the
restart, run once on prod:

```bash
kubectl exec -i registry-pg-1 -c postgres \
  --context gke_mcp-registry-prod_us-central1-b_mcp-registry-prod \
  -- psql -U postgres -c "CREATE EXTENSION IF NOT EXISTS pg_stat_statements"
```

Time the merge for a low-traffic UTC window. v1.7.1's DB
retry-with-backoff covers the brief PG restart.

## Out of scope

`MaxConns` bump, per-IP rate limiting at nginx, response caching, the
pre-existing `superfluous WriteHeader` warnings — separate follow-ups.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: rdimitrov

deploy/pkg/k8s/postgres.go

@@ -59,6 +59,26 @@ func DeployPostgresDatabases(ctx *pulumi.Context, cluster *providers.ProviderInf
 				"storage": map[string]any{
 					"size": "50Gi",
 				},
+				// Enable pg_stat_statements so we can attribute slow time to specific
+				// queries (the 2026-04-27 incident took an EXPLAIN-the-one-query-I-saw
+				// approach because we had no aggregate visibility). CNPG triggers a PG
+				// restart on shared_preload_libraries change — with instances: 1 this
+				// is brief downtime. CREATE EXTENSION still needs to run once as a
+				// superuser on existing clusters; new clusters get it via the
+				// postInitApplicationSQL hook below.
+				"postgresql": map[string]any{
+					"shared_preload_libraries": []any{"pg_stat_statements"},
+					"parameters": map[string]any{
+						"pg_stat_statements.track": "all",
+					},
+				},
+				"bootstrap": map[string]any{
+					"initdb": map[string]any{
+						"postInitApplicationSQL": []any{
+							"CREATE EXTENSION IF NOT EXISTS pg_stat_statements",
+						},
+					},
+				},
 			},
 		},
 	}, pulumi.Provider(cluster.Provider), pulumi.DependsOnInputs(cloudNativePG.Ready), pulumi.RetainOnDelete(true))

internal/database/postgres.go

@@ -127,7 +127,16 @@ func buildFilterConditions(filter *ServerFilter, argIndex int) ([]string, []any,
 	return conditions, args, argIndex
 }
 
-// addCursorCondition adds pagination cursor condition to WHERE clause
+// addCursorCondition adds pagination cursor condition to WHERE clause.
+//
+// The compound cursor uses a row-constructor comparison so PostgreSQL can seek
+// directly into the (server_name, version) B-tree index. The OR-decomposed form
+// `server_name > X OR (server_name = X AND version > Y)` is logically equivalent
+// but PostgreSQL's planner cannot use it for an index seek — it scans the index
+// from the start and filters everything before the cursor, making cost grow
+// linearly with cursor depth (a 20K-row table at a deep cursor took ~760ms in
+// prod). The row-constructor form `(server_name, version) > (X, Y)` is special-
+// cased and stays constant-time regardless of cursor depth.
 func addCursorCondition(cursor string, argIndex int) (string, []any, int) {
 	if cursor == "" {
 		return "", nil, argIndex
@@ -138,9 +147,8 @@ func addCursorCondition(cursor string, argIndex int) (string, []any, int) {
 	if len(parts) == 2 {
 		cursorServerName := parts[0]
 		cursorVersion := parts[1]
-		// Use compound condition: (server_name > cursor_name) OR (server_name = cursor_name AND version > cursor_version)
-		condition := fmt.Sprintf("(server_name > $%d OR (server_name = $%d AND version > $%d))", argIndex, argIndex+1, argIndex+2)
-		return condition, []any{cursorServerName, cursorServerName, cursorVersion}, argIndex + 3
+		condition := fmt.Sprintf("(server_name, version) > ($%d, $%d)", argIndex, argIndex+1)
+		return condition, []any{cursorServerName, cursorVersion}, argIndex + 2
 	}
 
 	// Fallback for malformed cursor - treat as server name only for backwards compatibility

internal/database/postgres_test.go

@@ -1150,6 +1150,104 @@ func TestPostgreSQL_PerformanceScenarios(t *testing.T) {
 		// Should have retrieved all servers including the ones we just created
 		assert.GreaterOrEqual(t, len(allResults), serverCount)
 	})
+
+	t.Run("compound cursor across versions of same server", func(t *testing.T) {
+		// Insert multiple versions of two servers so cursor pagination has to
+		// correctly seek across the (server_name, version) boundary. This is the
+		// case that the row-constructor cursor predicate `(server_name, version) > ($1, $2)`
+		// has to handle — the OR-decomposed form had the same semantics but a
+		// linear-scan plan; this test pins the semantic behaviour so a future
+		// rewrite can't silently break it.
+		serverA := "com.example/cursor-test-a"
+		serverB := "com.example/cursor-test-b"
+		versionsA := []string{"1.0.0", "2.0.0", "3.0.0"}
+		versionsB := []string{"1.0.0", "2.0.0"}
+
+		// First mark the latest version of each as latest=true; older ones false
+		// so the pkey + uniqueness constraints are satisfied.
+		mkServer := func(name, version string, isLatest bool) {
+			_, err := db.CreateServer(ctx, nil, &apiv0.ServerJSON{
+				Name: name, Description: "compound cursor test", Version: version,
+			}, &apiv0.RegistryExtensions{
+				Status:          model.StatusActive,
+				StatusChangedAt: timeNow, PublishedAt: timeNow, UpdatedAt: timeNow,
+				IsLatest: isLatest,
+			})
+			require.NoError(t, err)
+		}
+		for i, v := range versionsA {
+			mkServer(serverA, v, i == len(versionsA)-1)
+		}
+		for i, v := range versionsB {
+			mkServer(serverB, v, i == len(versionsB)-1)
+		}
+
+		// Filter to just the two servers we just created so unrelated rows in the
+		// shared test DB don't bleed in.
+		filterTo := func(rs []*apiv0.ServerResponse) []*apiv0.ServerResponse {
+			out := make([]*apiv0.ServerResponse, 0, len(rs))
+			for _, r := range rs {
+				if r.Server.Name == serverA || r.Server.Name == serverB {
+					out = append(out, r)
+				}
+			}
+			return out
+		}
+
+		// Cursor at (serverA, "1.0.0") must skip 1.0.0 and return 2.0.0, 3.0.0,
+		// then both versions of serverB. Specifically tests the compound predicate:
+		// without it, the OR form would still skip 1.0.0 correctly but the version
+		// boundary of (serverA, "3.0.0") → (serverB, "1.0.0") is what depends on
+		// the second-column comparison.
+		results, _, err := db.ListServers(ctx, nil, nil, serverA+":1.0.0", 100)
+		require.NoError(t, err)
+		got := filterTo(results)
+		require.Len(t, got, 4, "expected 4 rows after cursor at A:1.0.0")
+		assert.Equal(t, serverA, got[0].Server.Name)
+		assert.Equal(t, "2.0.0", got[0].Server.Version)
+		assert.Equal(t, serverA, got[1].Server.Name)
+		assert.Equal(t, "3.0.0", got[1].Server.Version)
+		assert.Equal(t, serverB, got[2].Server.Name)
+		assert.Equal(t, "1.0.0", got[2].Server.Version)
+		assert.Equal(t, serverB, got[3].Server.Name)
+		assert.Equal(t, "2.0.0", got[3].Server.Version)
+
+		// Cursor at the *last* version of serverA must cross the server boundary
+		// and return serverB rows only.
+		results, _, err = db.ListServers(ctx, nil, nil, serverA+":3.0.0", 100)
+		require.NoError(t, err)
+		got = filterTo(results)
+		require.Len(t, got, 2, "expected 2 rows after cursor at A:3.0.0")
+		assert.Equal(t, serverB, got[0].Server.Name)
+		assert.Equal(t, "1.0.0", got[0].Server.Version)
+		assert.Equal(t, serverB, got[1].Server.Name)
+		assert.Equal(t, "2.0.0", got[1].Server.Version)
+
+		// Page-by-page traversal with size=2 must produce the same global ordering
+		// (A 1.0.0, A 2.0.0, A 3.0.0, B 1.0.0, B 2.0.0).
+		var paged []*apiv0.ServerResponse
+		cursor := ""
+		for {
+			rs, next, err := db.ListServers(ctx, nil,
+				&database.ServerFilter{SubstringName: stringPtr("cursor-test-")},
+				cursor, 2)
+			require.NoError(t, err)
+			paged = append(paged, rs...)
+			if next == "" || len(rs) < 2 {
+				break
+			}
+			cursor = next
+		}
+		require.Len(t, paged, 5)
+		want := []struct{ name, version string }{
+			{serverA, "1.0.0"}, {serverA, "2.0.0"}, {serverA, "3.0.0"},
+			{serverB, "1.0.0"}, {serverB, "2.0.0"},
+		}
+		for i, w := range want {
+			assert.Equal(t, w.name, paged[i].Server.Name, "row %d name", i)
+			assert.Equal(t, w.version, paged[i].Server.Version, "row %d version", i)
+		}
+	})
 }
 
 func TestPostgreSQL_NewStatusFields(t *testing.T) {

internal/service/registry_service.go

@@ -18,6 +18,18 @@ import (
 
 const maxServerVersionsPerServer = 10000
 
+// Publish phase names emitted on the structured "publish complete"/"publish failed"
+// log from createServerInTransaction. Constants because version_checks is reported
+// from multiple branches.
+const (
+	phaseValidate           = "validate"
+	phaseAcquireLock        = "acquire_lock"
+	phaseValidateRemoteURLs = "validate_remote_urls"
+	phaseVersionChecks      = "version_checks"
+	phaseUnmarkLatest       = "unmark_latest"
+	phaseDBCreate           = "db_create"
+)
+
 // registryServiceImpl implements the RegistryService interface using our Database
 type registryServiceImpl struct {
 	db  database.Database
@@ -87,60 +99,105 @@ func (s *registryServiceImpl) CreateServer(ctx context.Context, req *apiv0.Serve
 }
 
 // createServerInTransaction contains the actual CreateServer logic within a transaction.
-func (s *registryServiceImpl) createServerInTransaction(ctx context.Context, tx pgx.Tx, req *apiv0.ServerJSON) (*apiv0.ServerResponse, error) {
+//
+// Phases are individually timed and emitted as a single structured log event per call.
+// During the 2026-04-27 incident the validate-only timing (the previous shape) hid
+// pool-exhaustion stalls in acquire_lock / version_checks / db_create — we saw 50s+
+// total publish times even though validate_ms was a few hundred ms. With every phase
+// reported the next slow publish tells us which step to blame.
+func (s *registryServiceImpl) createServerInTransaction(ctx context.Context, tx pgx.Tx, req *apiv0.ServerJSON) (resp *apiv0.ServerResponse, err error) {
+	start := time.Now()
 	serverJSON := *req
+	var (
+		validateMs, lockMs, remotesMs, versionChecksMs, unmarkMs, createMs int64
+		failedPhase                                                        string
+	)
 
-	// Validate the request. ValidatePublishRequest fans out to npm/PyPI/OCI for
-	// registry-ownership checks with 10s per-host timeouts, so it's the most likely
-	// contributor to publish latency. Log validate_ms on both success and failure
-	// so the next /v0/publish latency alert is diagnostic — a slow upstream that
-	// times out into a validation error is exactly the case we need to see.
-	validateStart := time.Now()
-	validateErr := validators.ValidatePublishRequest(ctx, serverJSON, s.cfg)
-	validateMs := time.Since(validateStart).Milliseconds()
-	if validateErr != nil {
-		slog.WarnContext(ctx, "publish validate failed",
+	defer func() {
+		attrs := []any{
 			"server_name", serverJSON.Name,
 			"version", serverJSON.Version,
+			"total_ms", time.Since(start).Milliseconds(),
 			"validate_ms", validateMs,
-			"error", validateErr.Error(),
-		)
-		return nil, validateErr
+			"lock_ms", lockMs,
+			"remotes_ms", remotesMs,
+			"version_checks_ms", versionChecksMs,
+			"unmark_ms", unmarkMs,
+			"create_ms", createMs,
+		}
+		if err != nil {
+			attrs = append(attrs, "failed_phase", failedPhase, "error", err.Error())
+			slog.WarnContext(ctx, "publish failed", attrs...)
+		} else {
+			slog.InfoContext(ctx, "publish complete", attrs...)
+		}
+	}()
+
+	// runPhase times fn into *ms and, on error, stashes the phase name + error
+	// onto the closed-over failedPhase / err. Returns true on success so callers
+	// can `if !runPhase(...) { return nil, err }`.
+	runPhase := func(name string, ms *int64, fn func() error) bool {
+		t := time.Now()
+		e := fn()
+		*ms = time.Since(t).Milliseconds()
+		if e != nil {
+			failedPhase = name
+			err = e
+			return false
+		}
+		return true
 	}
 
-	publishTime := time.Now()
-
-	// Acquire advisory lock to prevent concurrent publishes of the same server
-	if err := s.db.AcquirePublishLock(ctx, tx, serverJSON.Name); err != nil {
+	// Validate the request — registry-ownership checks fan out to npm/PyPI/OCI with
+	// 10s per-host timeouts. Was historically the most likely slow phase; now any
+	// phase can be the slow one when the connection pool is starved.
+	if !runPhase(phaseValidate, &validateMs, func() error {
+		return validators.ValidatePublishRequest(ctx, serverJSON, s.cfg)
+	}) {
 		return nil, err
 	}
 
-	// Check for duplicate remote URLs
-	if err := s.validateNoDuplicateRemoteURLs(ctx, tx, serverJSON); err != nil {
-		return nil, err
-	}
+	publishTime := time.Now()
 
-	// Check we haven't exceeded the maximum versions allowed for a server
-	versionCount, err := s.db.CountServerVersions(ctx, tx, serverJSON.Name)
-	if err != nil && !errors.Is(err, database.ErrNotFound) {
+	// Acquire advisory lock to prevent concurrent publishes of the same server
+	if !runPhase(phaseAcquireLock, &lockMs, func() error {
+		return s.db.AcquirePublishLock(ctx, tx, serverJSON.Name)
+	}) {
 		return nil, err
 	}
-	if versionCount >= maxServerVersionsPerServer {
-		return nil, database.ErrMaxServersReached
-	}
 
-	// Check this isn't a duplicate version
-	versionExists, err := s.db.CheckVersionExists(ctx, tx, serverJSON.Name, serverJSON.Version)
-	if err != nil {
+	// Check for duplicate remote URLs
+	if !runPhase(phaseValidateRemoteURLs, &remotesMs, func() error {
+		return s.validateNoDuplicateRemoteURLs(ctx, tx, serverJSON)
+	}) {
 		return nil, err
 	}
-	if versionExists {
-		return nil, database.ErrInvalidVersion
-	}
 
-	// Get current latest version to determine if new version should be latest
-	currentLatest, err := s.db.GetCurrentLatestVersion(ctx, tx, serverJSON.Name)
-	if err != nil && !errors.Is(err, database.ErrNotFound) {
+	// Version checks: count, exists, current-latest (small DB lookups, but on a
+	// starved pool any of them stalls until a connection is free). Bundled under
+	// one phase since they share a logical step.
+	var currentLatest *apiv0.ServerResponse
+	if !runPhase(phaseVersionChecks, &versionChecksMs, func() error {
+		versionCount, e := s.db.CountServerVersions(ctx, tx, serverJSON.Name)
+		if e != nil && !errors.Is(e, database.ErrNotFound) {
+			return e
+		}
+		if versionCount >= maxServerVersionsPerServer {
+			return database.ErrMaxServersReached
+		}
+		versionExists, e := s.db.CheckVersionExists(ctx, tx, serverJSON.Name, serverJSON.Version)
+		if e != nil {
+			return e
+		}
+		if versionExists {
+			return database.ErrInvalidVersion
+		}
+		currentLatest, e = s.db.GetCurrentLatestVersion(ctx, tx, serverJSON.Name)
+		if e != nil && !errors.Is(e, database.ErrNotFound) {
+			return e
+		}
+		return nil
+	}) {
 		return nil, err
 	}
 
@@ -161,7 +218,9 @@ func (s *registryServiceImpl) createServerInTransaction(ctx context.Context, tx
 
 	// Unmark old latest version if needed
 	if isNewLatest && currentLatest != nil {
-		if err := s.db.UnmarkAsLatest(ctx, tx, serverJSON.Name); err != nil {
+		if !runPhase(phaseUnmarkLatest, &unmarkMs, func() error {
+			return s.db.UnmarkAsLatest(ctx, tx, serverJSON.Name)
+		}) {
 			return nil, err
 		}
 	}
@@ -176,16 +235,13 @@ func (s *registryServiceImpl) createServerInTransaction(ctx context.Context, tx
 	}
 
 	// Insert new server version
-	resp, err := s.db.CreateServer(ctx, tx, &serverJSON, officialMeta)
-	if err != nil {
+	if !runPhase(phaseDBCreate, &createMs, func() error {
+		var e error
+		resp, e = s.db.CreateServer(ctx, tx, &serverJSON, officialMeta)
+		return e
+	}) {
 		return nil, err
 	}
-
-	slog.InfoContext(ctx, "publish complete",
-		"server_name", serverJSON.Name,
-		"version", serverJSON.Version,
-		"validate_ms", validateMs,
-	)
 	return resp, nil
 }