diff --git a/internal/api/handlers/v0/publish.go b/internal/api/handlers/v0/publish.go index bc01591cd..de563cab2 100644 --- a/internal/api/handlers/v0/publish.go +++ b/internal/api/handlers/v0/publish.go @@ -60,6 +60,11 @@ func RegisterPublishEndpoint(api huma.API, pathPrefix string, registry service.R return nil, huma.Error422UnprocessableEntity("Failed to publish server, invalid schema: call /validate for details") } + // Tag the API version so the publish log can attribute a slow publish to the + // route it arrived on. /v0 and /v0.1 share one service instance, so nothing + // below this point can tell them apart otherwise. + ctx = service.ContextWithAPIVersion(ctx, pathPrefix) + // Publish the server with extensions publishedServer, err := registry.CreateServer(ctx, &input.Body) if err != nil { diff --git a/internal/api/handlers/v0/publish_api_version_test.go b/internal/api/handlers/v0/publish_api_version_test.go new file mode 100644 index 000000000..1870dc5a3 --- /dev/null +++ b/internal/api/handlers/v0/publish_api_version_test.go @@ -0,0 +1,84 @@ +package v0_test + +import ( + "bytes" + "crypto/ed25519" + "crypto/rand" + "encoding/hex" + "encoding/json" + "log/slog" + "net/http" + "net/http/httptest" + "testing" + + "github.com/danielgtaylor/huma/v2" + "github.com/danielgtaylor/huma/v2/adapters/humago" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + v0 "github.com/modelcontextprotocol/registry/internal/api/handlers/v0" + "github.com/modelcontextprotocol/registry/internal/auth" + "github.com/modelcontextprotocol/registry/internal/config" + "github.com/modelcontextprotocol/registry/internal/database" + "github.com/modelcontextprotocol/registry/internal/service" + apiv0 "github.com/modelcontextprotocol/registry/pkg/api/v0" + "github.com/modelcontextprotocol/registry/pkg/model" +) + +// /v0 and /v0.1 register the same handler against the same service instance, so +// the API version a publish arrived on is only knowable if the handler tags the +// context. This exercises that wiring end to end rather than the service in +// isolation: the publish log has to name the prefix the route was registered under. +func TestPublishLogsAPIVersionForEachRoutePrefix(t *testing.T) { + for _, prefix := range []string{"/v0", "/v0.1"} { + t.Run(prefix, func(t *testing.T) { + var logBuf bytes.Buffer + previous := slog.Default() + slog.SetDefault(slog.New(slog.NewTextHandler(&logBuf, nil))) + t.Cleanup(func() { slog.SetDefault(previous) }) + + seed := make([]byte, ed25519.SeedSize) + _, err := rand.Read(seed) + require.NoError(t, err) + cfg := &config.Config{ + JWTPrivateKey: hex.EncodeToString(seed), + EnableRegistryValidation: false, + } + + mux := http.NewServeMux() + api := humago.New(mux, huma.DefaultConfig("Test API", "1.0.0")) + v0.RegisterPublishEndpoint(api, prefix, service.NewRegistryService(database.NewTestDB(t), cfg), cfg) + + token, err := generateTestJWTToken(cfg, auth.JWTClaims{ + AuthMethod: auth.MethodNone, + Permissions: []auth.Permission{{Action: auth.PermissionActionPublish, ResourcePattern: "*"}}, + }) + require.NoError(t, err) + + body, err := json.Marshal(apiv0.ServerJSON{ + Schema: model.CurrentSchemaURL, + Name: "com.example/api-version-attribution", + Description: "server used to assert the API version reaches the publish log", + Version: "1.0.0", + Packages: []model.Package{{ + RegistryType: model.RegistryTypeNPM, + Identifier: "example-package", + Version: "1.0.0", + Transport: model.Transport{Type: model.TransportTypeStdio}, + }}, + }) + require.NoError(t, err) + + req := httptest.NewRequest(http.MethodPost, prefix+"/publish", bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Authorization", "Bearer "+token) + + rr := httptest.NewRecorder() + mux.ServeHTTP(rr, req) + require.Equal(t, http.StatusOK, rr.Code, "publish should succeed: %s", rr.Body.String()) + + assert.Contains(t, logBuf.String(), "api_version="+prefix, + "the publish log must attribute the request to the route prefix it arrived on") + }) + } +} diff --git a/internal/service/registry_service.go b/internal/service/registry_service.go index 16a910002..2c9bf074a 100644 --- a/internal/service/registry_service.go +++ b/internal/service/registry_service.go @@ -19,17 +19,108 @@ 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. +// log. validate and tx_begin happen in CreateServer, before or while starting the +// transaction; the rest inside createServerInTransaction. Constants because +// version_checks is reported from multiple branches. const ( phaseValidate = "validate" + phaseTxBegin = "tx_begin" phaseAcquireLock = "acquire_lock" phaseValidateRemoteURLs = "validate_remote_urls" phaseVersionChecks = "version_checks" phaseUnmarkLatest = "unmark_latest" phaseDBCreate = "db_create" + phaseCommit = "commit" ) +// publishTimings accumulates the per-phase durations reported on the single +// structured "publish complete"/"publish failed" event emitted per publish. +// +// CreateServer records the phases that run before the transaction body starts +// (validate, tx_begin); createServerInTransaction records the phases inside it. +// Splitting it that way is deliberate: the previous shape started its clock after +// pool.Begin had already returned, so a publish that spent 40s getting a +// connection logged total_ms=200 and looked healthy. That is why the HTTP +// histogram and these logs disagreed — over 7 days the histogram saw 17 publishes +// above 10s while no logged publish exceeded 6.5s. +// +// tx_begin_ms is that missing time, but note what it covers: pgxpool's Begin both +// acquires a pooled connection and issues BEGIN over the wire, so this figure is +// acquisition *plus* one database round-trip. BEGIN is normally sub-millisecond, +// so a large value still points at acquisition — but it is not proof on its own, +// and a slow database or network inflates it too. Isolating the two would mean +// timing pool.Acquire inside the database layer. +// +// total_ms covers the whole CreateServer call, so any time not accounted for by +// the named phases is transaction commit plus overhead. +type publishTimings struct { + validateMs int64 + txBeginMs int64 + lockMs int64 + remotesMs int64 + versionChecksMs int64 + unmarkMs int64 + createMs int64 + failedPhase string +} + +// run times fn into *ms. On failure it records which phase failed and returns +// the error so the caller can abort. +func (t *publishTimings) run(name string, ms *int64, fn func() error) error { + started := time.Now() + err := fn() + *ms = time.Since(started).Milliseconds() + if err != nil { + t.failedPhase = name + } + return err +} + +// log emits the one structured event for this publish. +func (t *publishTimings) log(ctx context.Context, serverJSON apiv0.ServerJSON, totalMs int64, err error) { + attrs := []any{ + "server_name", serverJSON.Name, + "version", serverJSON.Version, + "api_version", apiVersionFromContext(ctx), + "total_ms", totalMs, + "validate_ms", t.validateMs, + "tx_begin_ms", t.txBeginMs, + "lock_ms", t.lockMs, + "remotes_ms", t.remotesMs, + "version_checks_ms", t.versionChecksMs, + "unmark_ms", t.unmarkMs, + "create_ms", t.createMs, + } + if err != nil { + attrs = append(attrs, "failed_phase", t.failedPhase, "error", err.Error()) + slog.WarnContext(ctx, "publish failed", attrs...) + } else { + slog.InfoContext(ctx, "publish complete", attrs...) + } +} + +// apiVersionKey carries the API version prefix a request arrived on. Both /v0 and +// /v0.1 are registered against the same service instance, so nothing below the +// handler can otherwise tell them apart — and the publish log's "version" field +// is the server's own semver, not the API version. +type apiVersionKey struct{} + +// ContextWithAPIVersion tags ctx with the API version prefix the request arrived +// on, for attribution in the publish log. +func ContextWithAPIVersion(ctx context.Context, apiVersion string) context.Context { + return context.WithValue(ctx, apiVersionKey{}, apiVersion) +} + +// apiVersionFromContext returns the tagged API version. Callers that do not tag +// the context — the importer, tests — report "unknown" rather than an empty +// value that reads as missing data. +func apiVersionFromContext(ctx context.Context) string { + if v, ok := ctx.Value(apiVersionKey{}).(string); ok && v != "" { + return v + } + return "unknown" +} + // registryServiceImpl implements the RegistryService interface using our Database type registryServiceImpl struct { db database.Database @@ -91,85 +182,81 @@ func (s *registryServiceImpl) GetAllVersionsByServerName(ctx context.Context, se } // CreateServer creates a new server version -func (s *registryServiceImpl) CreateServer(ctx context.Context, req *apiv0.ServerJSON) (*apiv0.ServerResponse, error) { - // Wrap the entire operation in a transaction - return database.InTransactionT(ctx, s.db, func(ctx context.Context, tx pgx.Tx) (*apiv0.ServerResponse, error) { - return s.createServerInTransaction(ctx, tx, req) - }) -} - -// createServerInTransaction contains the actual CreateServer logic within a transaction. -// -// 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) { +func (s *registryServiceImpl) CreateServer(ctx context.Context, req *apiv0.ServerJSON) (resp *apiv0.ServerResponse, err error) { start := time.Now() serverJSON := *req - var ( - validateMs, lockMs, remotesMs, versionChecksMs, unmarkMs, createMs int64 - failedPhase string - ) + timings := &publishTimings{} defer func() { - attrs := []any{ - "server_name", serverJSON.Name, - "version", serverJSON.Version, - "total_ms", time.Since(start).Milliseconds(), - "validate_ms", validateMs, - "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...) - } + timings.log(ctx, serverJSON, time.Since(start).Milliseconds(), err) }() - // 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 { + // Registry-ownership validation fans out to npm/PyPI/NuGet/Cargo/OCI/MCPB with + // a 10s timeout per host. It runs before the transaction opens so those + // round-trips never hold a pgxpool connection. Inside the transaction, a burst + // of publishes each waiting on a third-party registry would occupy connections + // while doing no database work, starving the pool and stalling unrelated + // requests in pool.Begin. Nothing here reads our own database, so hoisting it + // introduces no race; the per-server advisory lock was already taken after + // validation, and still is. + if err = timings.run(phaseValidate, &timings.validateMs, func() error { return validators.ValidatePublishRequest(ctx, serverJSON, s.cfg) - }) { + }); err != nil { return nil, err } + // Everything from here needs the database. InTransaction calls pool.Begin + // before invoking this callback, so the gap between these two timestamps is + // how long it took to get a transaction started — connection acquisition plus + // the BEGIN round-trip. See the note on publishTimings. + beforeBegin := time.Now() + gotConnection := false + resp, err = database.InTransactionT(ctx, s.db, func(ctx context.Context, tx pgx.Tx) (*apiv0.ServerResponse, error) { + gotConnection = true + timings.txBeginMs = time.Since(beforeBegin).Milliseconds() + return s.createServerInTransaction(ctx, tx, req, timings) + }) + switch { + case !gotConnection: + // Begin never handed us a transaction, so the callback above did not run and + // no phase inside it was reached. This is the shape the worst starvation + // takes — Begin blocking until the context deadline — so record the whole + // elapsed time and name the phase, rather than reporting zero against an + // empty phase for the exact case this instrumentation targets. + timings.txBeginMs = time.Since(beforeBegin).Milliseconds() + timings.failedPhase = phaseTxBegin + case err != nil && timings.failedPhase == "": + // Every timed phase succeeded but InTransaction still failed, which leaves + // only the commit: it runs after the callback returns and is not one of the + // phases. Without this the log would report a failure with no phase at all. + timings.failedPhase = phaseCommit + } + return resp, err +} + +// createServerInTransaction contains the actual CreateServer logic that needs the +// database. Registry-ownership validation deliberately happens in CreateServer, +// before the transaction opens — see the comment there. +// +// Each phase is timed into the shared publishTimings so CreateServer can emit one +// structured log event per publish. During the 2026-04-27 incident a validate-only +// timing hid pool-exhaustion stalls in acquire_lock / version_checks / db_create; +// with every phase reported the next slow publish says which step to blame. +func (s *registryServiceImpl) createServerInTransaction(ctx context.Context, tx pgx.Tx, req *apiv0.ServerJSON, t *publishTimings) (resp *apiv0.ServerResponse, err error) { + serverJSON := *req publishTime := time.Now() // Acquire advisory lock to prevent concurrent publishes of the same server - if !runPhase(phaseAcquireLock, &lockMs, func() error { + if err = t.run(phaseAcquireLock, &t.lockMs, func() error { return s.db.AcquirePublishLock(ctx, tx, serverJSON.Name) - }) { + }); err != nil { return nil, err } // Check for duplicate remote URLs - if !runPhase(phaseValidateRemoteURLs, &remotesMs, func() error { + if err = t.run(phaseValidateRemoteURLs, &t.remotesMs, func() error { return s.validateNoDuplicateRemoteURLs(ctx, tx, serverJSON) - }) { + }); err != nil { return nil, err } @@ -177,7 +264,7 @@ func (s *registryServiceImpl) createServerInTransaction(ctx context.Context, tx // 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 { + if err = t.run(phaseVersionChecks, &t.versionChecksMs, func() error { versionCount, e := s.db.CountServerVersions(ctx, tx, serverJSON.Name) if e != nil && !errors.Is(e, database.ErrNotFound) { return e @@ -197,7 +284,7 @@ func (s *registryServiceImpl) createServerInTransaction(ctx context.Context, tx return e } return nil - }) { + }); err != nil { return nil, err } @@ -218,9 +305,9 @@ func (s *registryServiceImpl) createServerInTransaction(ctx context.Context, tx // Unmark old latest version if needed if isNewLatest && currentLatest != nil { - if !runPhase(phaseUnmarkLatest, &unmarkMs, func() error { + if err = t.run(phaseUnmarkLatest, &t.unmarkMs, func() error { return s.db.UnmarkAsLatest(ctx, tx, serverJSON.Name) - }) { + }); err != nil { return nil, err } } @@ -235,11 +322,11 @@ func (s *registryServiceImpl) createServerInTransaction(ctx context.Context, tx } // Insert new server version - if !runPhase(phaseDBCreate, &createMs, func() error { + if err = t.run(phaseDBCreate, &t.createMs, func() error { var e error resp, e = s.db.CreateServer(ctx, tx, &serverJSON, officialMeta) return e - }) { + }); err != nil { return nil, err } return resp, nil @@ -345,7 +432,15 @@ func (s *registryServiceImpl) updateServerInTransaction(ctx context.Context, tx beingDeleted := statusChange != nil && statusChange.NewStatus == model.StatusDeleted skipRegistryValidation := currentlyDeleted || beingDeleted - // Validate the request, potentially skipping registry validation for deleted servers + // Unlike the publish path, this registry validation still runs inside the + // transaction, so its external HTTP calls hold a pool connection. It cannot + // simply be hoisted: skipRegistryValidation is derived from the + // GetServerByNameAndVersion read above, so moving validation out needs a + // second read before the transaction and a decision about the race in whether + // to skip. Left as-is because the edit path carries a small fraction of + // publish traffic. Note it emits no phase log of its own, so there is no + // tx_begin_ms to watch here — use mcp_registry_http_request_duration for the + // edit routes if this needs revisiting. if err := validators.ValidateUpdateRequest(ctx, *req, s.cfg, skipRegistryValidation); err != nil { return nil, err } diff --git a/internal/service/registry_service_publish_transaction_test.go b/internal/service/registry_service_publish_transaction_test.go new file mode 100644 index 000000000..cc5f8d7b2 --- /dev/null +++ b/internal/service/registry_service_publish_transaction_test.go @@ -0,0 +1,366 @@ +//nolint:testpackage +package service + +import ( + "bytes" + "context" + "errors" + "log/slog" + "regexp" + "strconv" + "strings" + "testing" + "time" + + "github.com/jackc/pgx/v5" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/modelcontextprotocol/registry/internal/config" + "github.com/modelcontextprotocol/registry/internal/database" + apiv0 "github.com/modelcontextprotocol/registry/pkg/api/v0" + "github.com/modelcontextprotocol/registry/pkg/model" +) + +// errUnexpectedCall is returned by every countingDB method the test under test +// is not expected to reach. Returning an error rather than panicking keeps a +// surprise call visible as a test failure instead of a crash. +var errUnexpectedCall = errors.New("unexpected database call") + +// countingDB is a database.Database that records how many transactions were +// opened. It exists to assert where work happens relative to the transaction +// boundary, which a real database cannot observe. +type countingDB struct { + transactions int +} + +func (d *countingDB) InTransaction(ctx context.Context, fn func(ctx context.Context, tx pgx.Tx) error) error { + d.transactions++ + // A nil pgx.Tx is safe here: the code paths under test must fail before + // touching it. Any use would panic and surface as a test failure. + return fn(ctx, nil) +} + +func (d *countingDB) CreateServer(context.Context, pgx.Tx, *apiv0.ServerJSON, *apiv0.RegistryExtensions) (*apiv0.ServerResponse, error) { + return nil, errUnexpectedCall +} + +func (d *countingDB) UpdateServer(context.Context, pgx.Tx, string, string, *apiv0.ServerJSON) (*apiv0.ServerResponse, error) { + return nil, errUnexpectedCall +} + +func (d *countingDB) SetServerStatus(context.Context, pgx.Tx, string, string, model.Status, *string) (*apiv0.ServerResponse, error) { + return nil, errUnexpectedCall +} + +func (d *countingDB) SetAllVersionsStatus(context.Context, pgx.Tx, string, model.Status, *string) ([]*apiv0.ServerResponse, error) { + return nil, errUnexpectedCall +} + +func (d *countingDB) ListServers(context.Context, pgx.Tx, *database.ServerFilter, string, int) ([]*apiv0.ServerResponse, string, error) { + return nil, "", errUnexpectedCall +} + +func (d *countingDB) GetServerByName(context.Context, pgx.Tx, string, bool) (*apiv0.ServerResponse, error) { + return nil, errUnexpectedCall +} + +func (d *countingDB) GetServerByNameAndVersion(context.Context, pgx.Tx, string, string, bool) (*apiv0.ServerResponse, error) { + return nil, errUnexpectedCall +} + +func (d *countingDB) GetAllVersionsByServerName(context.Context, pgx.Tx, string, bool) ([]*apiv0.ServerResponse, error) { + return nil, errUnexpectedCall +} + +func (d *countingDB) GetCurrentLatestVersion(context.Context, pgx.Tx, string) (*apiv0.ServerResponse, error) { + return nil, errUnexpectedCall +} + +func (d *countingDB) CountServerVersions(context.Context, pgx.Tx, string) (int, error) { + return 0, errUnexpectedCall +} + +func (d *countingDB) CheckVersionExists(context.Context, pgx.Tx, string, string) (bool, error) { + return false, errUnexpectedCall +} + +func (d *countingDB) UnmarkAsLatest(context.Context, pgx.Tx, string) error { return errUnexpectedCall } + +func (d *countingDB) SetLatestVersion(context.Context, pgx.Tx, string, string) error { + return errUnexpectedCall +} + +func (d *countingDB) AcquirePublishLock(context.Context, pgx.Tx, string) error { + return errUnexpectedCall +} + +func (d *countingDB) Close() error { return nil } + +// serverWithUnvalidatablePackage returns a request whose registry-ownership +// check fails without any network access: ValidatePackage rejects an unknown +// registry type outright. +func serverWithUnvalidatablePackage() *apiv0.ServerJSON { + return &apiv0.ServerJSON{ + Name: "io.github.example/publish-transaction-test", + Description: "server used to assert transaction boundaries", + Version: "1.0.0", + Packages: []model.Package{ + { + RegistryType: "not-a-real-registry", + Identifier: "example", + Version: "1.0.0", + }, + }, + } +} + +// capturePublishLogs redirects the default slog logger for the duration of the +// test and returns a function yielding everything written to it. The publish +// phase timings are only observable through that log line. +func capturePublishLogs(t *testing.T) func() string { + t.Helper() + var buf bytes.Buffer + previous := slog.Default() + slog.SetDefault(slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug}))) + t.Cleanup(func() { slog.SetDefault(previous) }) + return buf.String +} + +// Registry-ownership validation reaches out to npm, PyPI, NuGet, Cargo, OCI and +// MCPB over the network. Doing that inside the transaction pins a pgxpool +// connection for the duration of those round-trips, so under a publish burst +// connections are held by requests that are only waiting on third parties and +// everything behind them blocks in pool.Begin. Validating before the +// transaction opens is what keeps a slow upstream from consuming a connection. +func TestCreateServerValidatesBeforeOpeningTransaction(t *testing.T) { + db := &countingDB{} + svc := NewRegistryService(db, &config.Config{EnableRegistryValidation: true}) + + _, err := svc.CreateServer(context.Background(), serverWithUnvalidatablePackage()) + + require.Error(t, err, "an unknown registry type must fail registry validation") + assert.Equal(t, 0, db.transactions, + "registry validation failed, so no transaction should have been opened: validation must run before the transaction to avoid holding a pool connection during external HTTP calls") +} + +// slowBeginDB simulates a starved connection pool: pool.Begin blocks before the +// transaction body runs. Everything after that succeeds, so a publish against it +// spends nearly all its time waiting for a connection. +type slowBeginDB struct { + countingDB + beginDelay time.Duration +} + +func (d *slowBeginDB) InTransaction(ctx context.Context, fn func(ctx context.Context, tx pgx.Tx) error) error { + time.Sleep(d.beginDelay) + d.transactions++ + return fn(ctx, nil) +} + +func (d *slowBeginDB) AcquirePublishLock(context.Context, pgx.Tx, string) error { return nil } + +func (d *slowBeginDB) CountServerVersions(context.Context, pgx.Tx, string) (int, error) { + return 0, nil +} + +func (d *slowBeginDB) CheckVersionExists(context.Context, pgx.Tx, string, string) (bool, error) { + return false, nil +} + +func (d *slowBeginDB) GetCurrentLatestVersion(context.Context, pgx.Tx, string) (*apiv0.ServerResponse, error) { + return nil, database.ErrNotFound +} + +func (d *slowBeginDB) CreateServer(_ context.Context, _ pgx.Tx, serverJSON *apiv0.ServerJSON, meta *apiv0.RegistryExtensions) (*apiv0.ServerResponse, error) { + return &apiv0.ServerResponse{Server: *serverJSON, Meta: apiv0.ResponseMeta{Official: meta}}, nil +} + +// Time spent waiting for a pool connection was invisible: InTransaction calls +// pool.Begin before invoking the transaction body, and the old timing started +// inside that body. A publish stalled 40s on a starved pool therefore logged +// total_ms=200 and looked healthy, which is why the HTTP metric and these logs +// disagreed. tx_begin_ms has to account for it. +func TestCreateServerReportsTransactionBeginLatency(t *testing.T) { + logs := capturePublishLogs(t) + const beginDelay = 150 * time.Millisecond + db := &slowBeginDB{beginDelay: beginDelay} + svc := NewRegistryService(db, &config.Config{EnableRegistryValidation: false}) + + _, err := svc.CreateServer(context.Background(), &apiv0.ServerJSON{ + Name: "io.github.example/pool-wait-test", + Description: "server used to assert the pool wait is measured", + Version: "1.0.0", + }) + require.NoError(t, err) + + txBeginMs, ok := phaseValueFromLog(t, logs(), "tx_begin_ms") + require.True(t, ok, "the publish log must report tx_begin_ms") + assert.GreaterOrEqual(t, txBeginMs, beginDelay.Milliseconds(), + "tx_begin_ms must cover the time pool.Begin blocked, otherwise a starved pool is invisible in the logs") +} + +// failingBeginDB never hands out a connection: InTransaction blocks and then +// fails without invoking the transaction body, the way pool.Begin behaves when a +// starved pool hits the context deadline. +type failingBeginDB struct { + countingDB + beginDelay time.Duration + beginErr error +} + +func (d *failingBeginDB) InTransaction(context.Context, func(context.Context, pgx.Tx) error) error { + time.Sleep(d.beginDelay) + return d.beginErr +} + +// The worst pool starvation ends with Begin failing on the context deadline, and +// then the transaction body never runs. Recording the wait only from inside that +// body would report tx_begin_ms=0 with no failing phase for precisely the case +// this instrumentation exists to catch. +func TestCreateServerReportsBeginLatencyWhenBeginNeverSucceeds(t *testing.T) { + logs := capturePublishLogs(t) + const beginDelay = 150 * time.Millisecond + db := &failingBeginDB{beginDelay: beginDelay, beginErr: context.DeadlineExceeded} + svc := NewRegistryService(db, &config.Config{EnableRegistryValidation: false}) + + _, err := svc.CreateServer(context.Background(), &apiv0.ServerJSON{ + Name: "io.github.example/begin-failure", + Description: "server used to assert a failed Begin is still attributed", + Version: "1.0.0", + }) + require.Error(t, err) + + out := logs() + txBeginMs, ok := phaseValueFromLog(t, out, "tx_begin_ms") + require.True(t, ok, "the publish log must report tx_begin_ms") + assert.GreaterOrEqual(t, txBeginMs, beginDelay.Milliseconds(), + "the wait before Begin failed must still be reported, otherwise the worst starvation logs as zero") + assert.Contains(t, out, "failed_phase="+phaseTxBegin, + "a publish that never got a connection must be attributed to acquiring one, not to an empty phase") +} + +// The log event moved out of the transaction body, so the deferred logger has to +// observe errors raised inside it too — otherwise failures after the transaction +// opens would stop being reported. Guards that, rather than driving new code. +func TestCreateServerLogsFailureRaisedInsideTransaction(t *testing.T) { + logs := capturePublishLogs(t) + db := &countingDB{} // AcquirePublishLock returns errUnexpectedCall + svc := NewRegistryService(db, &config.Config{EnableRegistryValidation: false}) + + _, err := svc.CreateServer(context.Background(), &apiv0.ServerJSON{ + Name: "io.github.example/in-transaction-failure", + Description: "server used to assert in-transaction failures are logged", + Version: "1.0.0", + }) + + require.Error(t, err) + require.Equal(t, 1, db.transactions, "the failure must occur after the transaction opened") + out := logs() + assert.Contains(t, out, "publish failed") + assert.Contains(t, out, "failed_phase="+phaseAcquireLock, + "a phase failing inside the transaction must still be named in the log event") +} + +// phaseValueFromLog pulls a single integer phase field out of the captured +// "key=value" slog text output. +func phaseValueFromLog(t *testing.T, out, field string) (int64, bool) { + t.Helper() + m := regexp.MustCompile(field + `=(\d+)`).FindStringSubmatch(out) + if m == nil { + return 0, false + } + v, err := strconv.ParseInt(m[1], 10, 64) + require.NoError(t, err) + return v, true +} + +// Moving validation out of the transaction must not cost us the diagnostic log +// line: a validation failure still has to report one "publish failed" event +// naming the phase that failed. +func TestCreateServerLogsValidationFailureWithPhase(t *testing.T) { + logs := capturePublishLogs(t) + svc := NewRegistryService(&countingDB{}, &config.Config{EnableRegistryValidation: true}) + + _, err := svc.CreateServer(context.Background(), serverWithUnvalidatablePackage()) + require.Error(t, err) + + out := logs() + assert.Contains(t, out, "publish failed", "a failed publish must emit its structured log event") + assert.Contains(t, out, "failed_phase=validate", "the event must name validate as the failing phase") + assert.Equal(t, 1, strings.Count(out, "publish failed"), "exactly one event per publish") +} + +// Both /v0 and /v0.1 route to the same service methods, so without an explicit +// tag the publish log cannot say which API version a slow publish arrived on. +// The log's existing "version" field is the server's semver, not the API version. +func TestCreateServerLogsAPIVersionFromContext(t *testing.T) { + logs := capturePublishLogs(t) + svc := NewRegistryService(&slowBeginDB{}, &config.Config{EnableRegistryValidation: false}) + + _, err := svc.CreateServer(ContextWithAPIVersion(context.Background(), "/v0.1"), &apiv0.ServerJSON{ + Name: "io.github.example/api-version-test", + Description: "server used to assert the API version is attributed", + Version: "1.0.0", + }) + require.NoError(t, err) + + assert.Contains(t, logs(), "api_version=/v0.1", + "the publish log must report which API version the request arrived on") +} + +// Callers that do not tag the context (importer, tests, future callers) must not +// produce a blank field that reads as missing data. +func TestCreateServerLogsUnknownAPIVersionWhenUntagged(t *testing.T) { + logs := capturePublishLogs(t) + svc := NewRegistryService(&slowBeginDB{}, &config.Config{EnableRegistryValidation: false}) + + _, err := svc.CreateServer(context.Background(), &apiv0.ServerJSON{ + Name: "io.github.example/api-version-absent", + Description: "server used to assert the untagged default", + Version: "1.0.0", + }) + require.NoError(t, err) + + assert.Contains(t, logs(), "api_version=unknown", + "an untagged context should report unknown rather than an empty value") +} + +// commitFailureDB runs the transaction body successfully and then fails the +// commit, the way PostgreSQL.InTransaction behaves when tx.Commit errors after +// the callback returned nil. +type commitFailureDB struct { + slowBeginDB + commitErr error +} + +func (d *commitFailureDB) InTransaction(ctx context.Context, fn func(ctx context.Context, tx pgx.Tx) error) error { + d.transactions++ + if err := fn(ctx, nil); err != nil { + return err + } + return d.commitErr +} + +// Every timed phase can succeed and the publish still fail, because the commit +// happens after the transaction body returns and is not one of the phases. That +// left the outer logger reporting a failure with no phase attached — the same +// unattributed-failure gap as a failed Begin, at the other end of the transaction. +func TestCreateServerAttributesACommitFailure(t *testing.T) { + logs := capturePublishLogs(t) + db := &commitFailureDB{commitErr: errors.New("failed to commit transaction: connection reset by peer")} + svc := NewRegistryService(db, &config.Config{EnableRegistryValidation: false}) + + _, err := svc.CreateServer(context.Background(), &apiv0.ServerJSON{ + Name: "io.github.example/commit-failure", + Description: "server used to assert a commit failure is attributed", + Version: "1.0.0", + }) + require.Error(t, err) + + out := logs() + assert.Contains(t, out, "failed_phase="+phaseCommit, + "a publish that failed only at commit must name the commit phase") + assert.NotContains(t, out, `failed_phase=""`, + "the log must never report a failure with no phase attached") +}