diff --git a/deploy/pkg/k8s/postgres.go b/deploy/pkg/k8s/postgres.go index 57a28aa..7a119fd 100644 --- a/deploy/pkg/k8s/postgres.go +++ b/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)) diff --git a/internal/database/postgres.go b/internal/database/postgres.go index e40359a..ff139e2 100644 --- a/internal/database/postgres.go +++ b/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 diff --git a/internal/database/postgres_test.go b/internal/database/postgres_test.go index d0e4b97..6784ae5 100644 --- a/internal/database/postgres_test.go +++ b/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) { diff --git a/internal/service/registry_service.go b/internal/service/registry_service.go index 4403136..16a9100 100644 --- a/internal/service/registry_service.go +++ b/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 + } + + // 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 } 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 { + if !runPhase(phaseAcquireLock, &lockMs, func() error { + return s.db.AcquirePublishLock(ctx, tx, serverJSON.Name) + }) { return nil, err } // Check for duplicate remote URLs - if err := s.validateNoDuplicateRemoteURLs(ctx, tx, serverJSON); err != nil { + if !runPhase(phaseValidateRemoteURLs, &remotesMs, func() error { + return s.validateNoDuplicateRemoteURLs(ctx, tx, serverJSON) + }) { return nil, err } - // 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) { - 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 { - 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 }