From 2a3e8d7ad18534cb7b0f74f90ad1b7920ecc7267 Mon Sep 17 00:00:00 2001 From: Radoslav Dimitrov Date: Fri, 21 Aug 2026 15:08:38 +0300 Subject: [PATCH 1/4] fix(publish): validate registries outside the transaction and time the pool wait Two changes to the publish path, both driven by the same measurement. Over 7 days the HTTP histogram recorded 17 publishes above 10s while the per-phase publish log recorded none above 6.5s, with validate_ms peaking at 3.9s. The two layers disagreed because the log could not see the time that mattered: InTransaction calls pool.Begin before invoking the transaction body, and the timing started inside that body. A publish that waited 40s for a free connection logged total_ms=200 and looked healthy. Time the pool wait. The gap between the timestamp taken before InTransactionT and the first line of its callback is exactly how long pool.Begin blocked, reported as pool_wait_ms. Move registry-ownership validation ahead of the transaction. It fans out to npm, PyPI, NuGet, Cargo, OCI and MCPB with a 10s timeout per host, and running it inside the transaction pinned a pgxpool connection for the duration of those round-trips. Under a publish burst - up to 276 in a 4-minute window - connections are then held by requests doing no database work at all, starving a pool that already sits at 150 of 200 connections and stalling unrelated requests in pool.Begin. The check reads none of our own state, so hoisting it introduces no race, and the per-server advisory lock was already acquired after validation. To keep one log event per publish, the timing and logging move up to CreateServer via a shared publishTimings value; createServerInTransaction fills in the phases that still run inside the transaction. Note total_ms now covers the whole CreateServer call rather than just the transaction body, so the residual after the named phases is commit plus overhead. No dashboard or alert rule consumes these log fields; all four Grafana rules are built on mcp_registry_* metrics. The edit path keeps its validation inside the transaction, with a comment explaining why: skipRegistryValidation is derived from a read inside that transaction, so hoisting it needs a second read and a decision about the resulting race. It carries a fraction of publish traffic. Co-Authored-By: Claude Opus 5 (1M context) --- internal/service/registry_service.go | 176 +++++++----- ...gistry_service_publish_transaction_test.go | 252 ++++++++++++++++++ 2 files changed, 361 insertions(+), 67 deletions(-) create mode 100644 internal/service/registry_service_publish_transaction_test.go diff --git a/internal/service/registry_service.go b/internal/service/registry_service.go index 16a910002..3780f0420 100644 --- a/internal/service/registry_service.go +++ b/internal/service/registry_service.go @@ -30,6 +30,65 @@ const ( phaseDBCreate = "db_create" ) +// 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 opens +// (validate) plus the wait for a pool connection; 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 waiting for a free 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. +// pool_wait_ms is the missing time. +// +// 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 + poolWaitMs 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, + "total_ms", totalMs, + "validate_ms", t.validateMs, + "pool_wait_ms", t.poolWaitMs, + "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...) + } +} + // registryServiceImpl implements the RegistryService interface using our Database type registryServiceImpl struct { db database.Database @@ -91,85 +150,62 @@ 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 + // exactly how long the publish waited for a free connection. + beforeBegin := time.Now() + return database.InTransactionT(ctx, s.db, func(ctx context.Context, tx pgx.Tx) (*apiv0.ServerResponse, error) { + timings.poolWaitMs = time.Since(beforeBegin).Milliseconds() + return s.createServerInTransaction(ctx, tx, req, timings) + }) +} + +// 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 +213,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 +233,7 @@ func (s *registryServiceImpl) createServerInTransaction(ctx context.Context, tx return e } return nil - }) { + }); err != nil { return nil, err } @@ -218,9 +254,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 +271,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 +381,13 @@ 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; revisit if it shows up in pool_wait_ms. 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..3327889c4 --- /dev/null +++ b/internal/service/registry_service_publish_transaction_test.go @@ -0,0 +1,252 @@ +//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. pool_wait_ms has to account for it. +func TestCreateServerReportsTimeWaitingForAPoolConnection(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) + + poolWaitMs, ok := phaseValueFromLog(t, logs(), "pool_wait_ms") + require.True(t, ok, "the publish log must report pool_wait_ms") + assert.GreaterOrEqual(t, poolWaitMs, beginDelay.Milliseconds(), + "pool_wait_ms must cover the time pool.Begin blocked, otherwise a starved pool is invisible in the logs") +} + +// 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") +} From a9cbe6f2a6023067859e075b4aa4def916b9ed8f Mon Sep 17 00:00:00 2001 From: Radoslav Dimitrov Date: Fri, 21 Aug 2026 18:00:15 +0300 Subject: [PATCH 2/4] fix(publish): attribute a failed Begin to pool_begin instead of reporting zero MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit pool_wait_ms was only assigned from inside the transaction callback, but InTransaction skips that callback entirely when pool.Begin fails or when the context is already cancelled. A publish that blocked on a starved pool until its deadline therefore logged pool_wait_ms=0 with an empty failed_phase — hiding the exact condition this instrumentation was added to expose: total_ms=151 validate_ms=0 pool_wait_ms=0 ... failed_phase="" \ error="context deadline exceeded" Track whether the callback was entered. When it was not, record the full elapsed wait and attribute it to a new pool_begin phase, so the worst starvation reads as the longest wait rather than the shortest. Co-Authored-By: Claude Opus 5 (1M context) --- internal/service/registry_service.go | 20 ++++++++-- ...gistry_service_publish_transaction_test.go | 40 +++++++++++++++++++ 2 files changed, 57 insertions(+), 3 deletions(-) diff --git a/internal/service/registry_service.go b/internal/service/registry_service.go index 3780f0420..27f108b13 100644 --- a/internal/service/registry_service.go +++ b/internal/service/registry_service.go @@ -19,10 +19,12 @@ 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 pool_begin happen in CreateServer, before or while acquiring the +// transaction; the rest inside createServerInTransaction. Constants because +// version_checks is reported from multiple branches. const ( phaseValidate = "validate" + phasePoolBegin = "pool_begin" phaseAcquireLock = "acquire_lock" phaseValidateRemoteURLs = "validate_remote_urls" phaseVersionChecks = "version_checks" @@ -177,10 +179,22 @@ func (s *registryServiceImpl) CreateServer(ctx context.Context, req *apiv0.Serve // before invoking this callback, so the gap between these two timestamps is // exactly how long the publish waited for a free connection. beforeBegin := time.Now() - return database.InTransactionT(ctx, s.db, func(ctx context.Context, tx pgx.Tx) (*apiv0.ServerResponse, error) { + gotConnection := false + resp, err = database.InTransactionT(ctx, s.db, func(ctx context.Context, tx pgx.Tx) (*apiv0.ServerResponse, error) { + gotConnection = true timings.poolWaitMs = time.Since(beforeBegin).Milliseconds() return s.createServerInTransaction(ctx, tx, req, timings) }) + if !gotConnection { + // Begin never handed us a connection, so the callback above did not run and + // no phase inside the transaction was reached. This is the shape the worst + // starvation takes — Begin blocking until the context deadline — so record + // the whole elapsed wait and name the phase, rather than reporting zero + // against an empty phase for the exact case this instrumentation targets. + timings.poolWaitMs = time.Since(beforeBegin).Milliseconds() + timings.failedPhase = phasePoolBegin + } + return resp, err } // createServerInTransaction contains the actual CreateServer logic that needs the diff --git a/internal/service/registry_service_publish_transaction_test.go b/internal/service/registry_service_publish_transaction_test.go index 3327889c4..7ddd5b507 100644 --- a/internal/service/registry_service_publish_transaction_test.go +++ b/internal/service/registry_service_publish_transaction_test.go @@ -200,6 +200,46 @@ func TestCreateServerReportsTimeWaitingForAPoolConnection(t *testing.T) { "pool_wait_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 pool_wait_ms=0 with no failing phase for precisely the case +// this instrumentation exists to catch. +func TestCreateServerReportsPoolWaitWhenBeginNeverSucceeds(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() + poolWaitMs, ok := phaseValueFromLog(t, out, "pool_wait_ms") + require.True(t, ok, "the publish log must report pool_wait_ms") + assert.GreaterOrEqual(t, poolWaitMs, beginDelay.Milliseconds(), + "the wait before Begin failed must still be reported, otherwise the worst starvation logs as zero") + assert.Contains(t, out, "failed_phase="+phasePoolBegin, + "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. From b9987914a01971f0ea110d0f42cd32b835592139 Mon Sep 17 00:00:00 2001 From: Radoslav Dimitrov Date: Sat, 22 Aug 2026 01:32:32 +0300 Subject: [PATCH 3/4] feat(publish): attribute the publish log to the API version it arrived on MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit /v0 and /v0.1 register the same publish handler against the same service instance, so nothing below the handler could tell them apart. The log's existing "version" field is the server's own semver, which reads as an API version but is not one — so a slow publish could not be attributed to a route at all. Tag the API version prefix onto the request context in the handler and report it as api_version. Callers that do not tag it — the importer, tests — report "unknown" rather than an empty value that reads as missing data. Only /v0/publish carries real traffic today (/v0.1/publish is at 0 req/s), but that changes as publishers migrate, and pool_wait_ms is only useful if we can say which route the slow publishes came in on. Co-Authored-By: Claude Opus 5 (1M context) --- internal/api/handlers/v0/publish.go | 5 ++ .../handlers/v0/publish_api_version_test.go | 84 +++++++++++++++++++ internal/service/registry_service.go | 23 +++++ ...gistry_service_publish_transaction_test.go | 35 ++++++++ 4 files changed, 147 insertions(+) create mode 100644 internal/api/handlers/v0/publish_api_version_test.go 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 27f108b13..5b45bc22c 100644 --- a/internal/service/registry_service.go +++ b/internal/service/registry_service.go @@ -74,6 +74,7 @@ func (t *publishTimings) log(ctx context.Context, serverJSON apiv0.ServerJSON, t attrs := []any{ "server_name", serverJSON.Name, "version", serverJSON.Version, + "api_version", apiVersionFromContext(ctx), "total_ms", totalMs, "validate_ms", t.validateMs, "pool_wait_ms", t.poolWaitMs, @@ -91,6 +92,28 @@ func (t *publishTimings) log(ctx context.Context, serverJSON apiv0.ServerJSON, t } } +// 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 diff --git a/internal/service/registry_service_publish_transaction_test.go b/internal/service/registry_service_publish_transaction_test.go index 7ddd5b507..57307d501 100644 --- a/internal/service/registry_service_publish_transaction_test.go +++ b/internal/service/registry_service_publish_transaction_test.go @@ -290,3 +290,38 @@ func TestCreateServerLogsValidationFailureWithPhase(t *testing.T) { 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") +} From 036e871856adab71d5a7cda53ae4219ad91310c6 Mon Sep 17 00:00:00 2001 From: Radoslav Dimitrov Date: Sat, 22 Aug 2026 02:26:14 +0300 Subject: [PATCH 4/4] fix(publish): name what tx_begin_ms measures and attribute commit failures MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three findings from review, all valid. pool_wait_ms claimed to isolate the wait for a free connection, but pgxpool's BeginTx is Acquire followed by Exec("BEGIN") — so the figure was acquisition plus one database round-trip. A slow database or network inflated it and looked like pool starvation. Renamed to tx_begin_ms, with the composition spelled out: BEGIN is normally sub-millisecond so a large value still points at acquisition, but it is not proof on its own. Isolating the two would mean timing pool.Acquire in the database layer. A commit failure left the publish log reporting a failure with no phase. InTransaction returns a commit error after the callback has already succeeded, so every timed phase passes and failedPhase stays empty — the same unattributed-failure gap as a failed Begin, at the other end of the transaction. Attributed to a new commit phase. The edit-path comment told maintainers to watch pool_wait_ms there, which was impossible: only CreateServer builds publishTimings, so the edit path emits no phase log at all. Points at the HTTP request-duration metric for the edit routes instead. Co-Authored-By: Claude Opus 5 (1M context) --- internal/service/registry_service.go | 62 +++++++++++------- ...gistry_service_publish_transaction_test.go | 63 +++++++++++++++---- 2 files changed, 90 insertions(+), 35 deletions(-) diff --git a/internal/service/registry_service.go b/internal/service/registry_service.go index 5b45bc22c..2c9bf074a 100644 --- a/internal/service/registry_service.go +++ b/internal/service/registry_service.go @@ -19,36 +19,43 @@ import ( const maxServerVersionsPerServer = 10000 // Publish phase names emitted on the structured "publish complete"/"publish failed" -// log. validate and pool_begin happen in CreateServer, before or while acquiring the +// 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" - phasePoolBegin = "pool_begin" + 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 opens -// (validate) plus the wait for a pool connection; 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 waiting for a free 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. -// pool_wait_ms is the missing time. +// 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 - poolWaitMs int64 + txBeginMs int64 lockMs int64 remotesMs int64 versionChecksMs int64 @@ -77,7 +84,7 @@ func (t *publishTimings) log(ctx context.Context, serverJSON apiv0.ServerJSON, t "api_version", apiVersionFromContext(ctx), "total_ms", totalMs, "validate_ms", t.validateMs, - "pool_wait_ms", t.poolWaitMs, + "tx_begin_ms", t.txBeginMs, "lock_ms", t.lockMs, "remotes_ms", t.remotesMs, "version_checks_ms", t.versionChecksMs, @@ -200,22 +207,29 @@ func (s *registryServiceImpl) CreateServer(ctx context.Context, req *apiv0.Serve // Everything from here needs the database. InTransaction calls pool.Begin // before invoking this callback, so the gap between these two timestamps is - // exactly how long the publish waited for a free connection. + // 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.poolWaitMs = time.Since(beforeBegin).Milliseconds() + timings.txBeginMs = time.Since(beforeBegin).Milliseconds() return s.createServerInTransaction(ctx, tx, req, timings) }) - if !gotConnection { - // Begin never handed us a connection, so the callback above did not run and - // no phase inside the transaction was reached. This is the shape the worst - // starvation takes — Begin blocking until the context deadline — so record - // the whole elapsed wait and name the phase, rather than reporting zero - // against an empty phase for the exact case this instrumentation targets. - timings.poolWaitMs = time.Since(beforeBegin).Milliseconds() - timings.failedPhase = phasePoolBegin + 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 } @@ -424,7 +438,9 @@ func (s *registryServiceImpl) updateServerInTransaction(ctx context.Context, tx // 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; revisit if it shows up in pool_wait_ms. + // 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 index 57307d501..cc5f8d7b2 100644 --- a/internal/service/registry_service_publish_transaction_test.go +++ b/internal/service/registry_service_publish_transaction_test.go @@ -180,8 +180,8 @@ func (d *slowBeginDB) CreateServer(_ context.Context, _ pgx.Tx, serverJSON *apiv // 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. pool_wait_ms has to account for it. -func TestCreateServerReportsTimeWaitingForAPoolConnection(t *testing.T) { +// 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} @@ -194,10 +194,10 @@ func TestCreateServerReportsTimeWaitingForAPoolConnection(t *testing.T) { }) require.NoError(t, err) - poolWaitMs, ok := phaseValueFromLog(t, logs(), "pool_wait_ms") - require.True(t, ok, "the publish log must report pool_wait_ms") - assert.GreaterOrEqual(t, poolWaitMs, beginDelay.Milliseconds(), - "pool_wait_ms must cover the time pool.Begin blocked, otherwise a starved pool is invisible in the logs") + 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 @@ -216,9 +216,9 @@ func (d *failingBeginDB) InTransaction(context.Context, func(context.Context, pg // 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 pool_wait_ms=0 with no failing phase for precisely the case +// body would report tx_begin_ms=0 with no failing phase for precisely the case // this instrumentation exists to catch. -func TestCreateServerReportsPoolWaitWhenBeginNeverSucceeds(t *testing.T) { +func TestCreateServerReportsBeginLatencyWhenBeginNeverSucceeds(t *testing.T) { logs := capturePublishLogs(t) const beginDelay = 150 * time.Millisecond db := &failingBeginDB{beginDelay: beginDelay, beginErr: context.DeadlineExceeded} @@ -232,11 +232,11 @@ func TestCreateServerReportsPoolWaitWhenBeginNeverSucceeds(t *testing.T) { require.Error(t, err) out := logs() - poolWaitMs, ok := phaseValueFromLog(t, out, "pool_wait_ms") - require.True(t, ok, "the publish log must report pool_wait_ms") - assert.GreaterOrEqual(t, poolWaitMs, beginDelay.Milliseconds(), + 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="+phasePoolBegin, + 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") } @@ -325,3 +325,42 @@ func TestCreateServerLogsUnknownAPIVersionWhenUntagged(t *testing.T) { 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") +}