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>
This commit is contained in:
Radoslav Dimitrov
GitHub
parent
762a4849a6
commit
27af672c5e
1 changed files with 232 additions and 50 deletions
|
|
@ -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
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue