registry: stop background goroutines leaked across registry rebuilds

registry/handlers/app.go

@@ -5,6 +5,7 @@ import (
 	"crypto/rand"
 	"crypto/tls"
 	"crypto/x509"
+	"errors"
 	"expvar"
 	"fmt"
 	"math"
@@ -61,6 +62,9 @@ const defaultCheckInterval = 10 * time.Second
 type App struct {
 	context.Context
 
+	// cancel stops background goroutines started by NewApp (e.g. the upload purger).
+	cancel context.CancelFunc
+
 	Config *configuration.Configuration
 
 	router           *mux.Router                    // main application router, configured with dispatchers
@@ -92,9 +96,13 @@ type App struct {
 // requests. The app only implements ServeHTTP and can be wrapped in other
 // handlers accordingly.
 func NewApp(ctx context.Context, config *configuration.Configuration) *App {
+	// Make the app's context cancelable so Shutdown can stop background
+	// goroutines (e.g. the upload purger) that would otherwise leak the app.
+	ctx, cancel := context.WithCancel(ctx)
 	app := &App{
 		Config:  config,
 		Context: ctx,
+		cancel:  cancel,
 		router:  v2.RouterWithPrefix(config.HTTP.Prefix),
 		isCache: config.Proxy.RemoteURL != "",
 	}
@@ -449,10 +457,21 @@ func (app *App) RegisterHealthChecks(healthRegistries ...*health.Registry) {
 
 // Shutdown close the underlying registry
 func (app *App) Shutdown() error {
+	if app.cancel != nil {
+		app.cancel()
+	}
+	var errs []error
+	if app.events.sink != nil {
+		if err := app.events.sink.Close(); err != nil {
+			errs = append(errs, err)
+		}
+	}
 	if r, ok := app.registry.(proxy.Closer); ok {
-		return r.Close()
+		if err := r.Close(); err != nil {
+			errs = append(errs, err)
+		}
 	}
-	return nil
+	return errors.Join(errs...)
 }
 
 // register a handler with the application, by route name. The handler will be
@@ -1086,12 +1105,19 @@ func startUploadPurger(ctx context.Context, storageDriver storagedriver.StorageD
 		}
 		jitter := time.Duration(randInt.Int64()%60) * time.Minute
 		log.Infof("Starting upload purge in %s", jitter)
-		time.Sleep(jitter)
+		timer := time.NewTimer(jitter)
+		defer timer.Stop()
 
 		for {
+			select {
+			case <-ctx.Done():
+				log.Info("Upload purger stopped")
+				return
+			case <-timer.C:
+			}
 			storage.PurgeUploads(ctx, storageDriver, time.Now().Add(-purgeAgeDuration), !dryRunBool)
 			log.Infof("Starting upload purge in %s", intervalDuration)
-			time.Sleep(intervalDuration)
+			timer.Reset(intervalDuration)
 		}
 	}()
 }

registry/handlers/app_test.go

@@ -2,11 +2,13 @@ package handlers
 
 import (
 	"encoding/json"
+	"errors"
 	"net/http"
 	"net/http/httptest"
 	"net/url"
 	"reflect"
 	"testing"
+	"time"
 
 	"github.com/distribution/distribution/v3/configuration"
 	"github.com/distribution/distribution/v3/internal/dcontext"
@@ -17,6 +19,7 @@ import (
 	"github.com/distribution/distribution/v3/registry/storage"
 	memorycache "github.com/distribution/distribution/v3/registry/storage/cache/memory"
 	"github.com/distribution/distribution/v3/registry/storage/driver/inmemory"
+	events "github.com/docker/go-events"
 )
 
 // TestAppDispatcher builds an application with a test dispatcher and ensures
@@ -275,3 +278,28 @@ func TestAppendAccessRecords(t *testing.T) {
 		t.Fatal("Actual access record differs from expected")
 	}
 }
+
+// TestAppShutdownCancelsContext ensures Shutdown cancels the app's context so
+// background goroutines started by NewApp (e.g. the upload purger) terminate
+// instead of pinning the app in memory.
+func TestAppShutdownCancelsContext(t *testing.T) {
+	app := NewApp(dcontext.Background(), &configuration.Configuration{
+		Storage: configuration.Storage{
+			"inmemory": nil,
+		},
+	})
+
+	if err := app.Shutdown(); err != nil {
+		t.Fatalf("unexpected error shutting down app: %v", err)
+	}
+
+	select {
+	case <-app.Done():
+	case <-time.After(5 * time.Second):
+		t.Fatal("app context not canceled after Shutdown")
+	}
+
+	if err := app.events.sink.Write(nil); !errors.Is(err, events.ErrSinkClosed) {
+		t.Fatalf("expected event sink to be closed after Shutdown, got %v", err)
+	}
+}

registry/leak_test.go

@@ -0,0 +1,58 @@
+package registry
+
+import (
+	"context"
+	"runtime"
+	"testing"
+	"time"
+
+	"github.com/distribution/distribution/v3/configuration"
+	"github.com/distribution/distribution/v3/internal/dcontext"
+	_ "github.com/distribution/distribution/v3/registry/storage/driver/inmemory"
+)
+
+// TestRebuildDoesNotLeakGoroutines simulates a process that repeatedly tears
+// down and recreates a registry (e.g. on credential rotation) and asserts the
+// old registries' background goroutines (upload purger, event broadcaster) do
+// not accumulate.
+func TestRebuildDoesNotLeakGoroutines(t *testing.T) {
+	config := &configuration.Configuration{
+		Storage: configuration.Storage{"inmemory": configuration.Parameters{}},
+	}
+	config.HTTP.Addr = "127.0.0.1:15999"
+	config.Log.Level = "error"
+
+	rebuild := func() {
+		reg, err := NewRegistry(dcontext.Background(), config)
+		if err != nil {
+			t.Fatalf("NewRegistry: %v", err)
+		}
+		done := make(chan struct{})
+		go func() { _ = reg.ListenAndServe(); close(done) }()
+		time.Sleep(100 * time.Millisecond) // let it bind
+		if err := reg.Shutdown(context.Background()); err != nil {
+			t.Fatalf("Shutdown: %v", err)
+		}
+		<-done
+	}
+
+	rebuild() // warm up process-global state (otel, health)
+	runtime.GC()
+	before := runtime.NumGoroutine()
+
+	const iterations = 10
+	for i := 0; i < iterations; i++ {
+		rebuild()
+	}
+
+	var after int
+	for i := 0; i < 50; i++ { // allow stragglers to exit
+		runtime.GC()
+		after = runtime.NumGoroutine()
+		if after <= before+2 {
+			return
+		}
+		time.Sleep(100 * time.Millisecond)
+	}
+	t.Fatalf("goroutines grew from %d to %d over %d rebuilds", before, after, iterations)
+}

registry/registry.go

@@ -165,6 +165,10 @@ func NewRegistry(ctx context.Context, config *configuration.Configuration) (*Reg
 
 	err = tracing.InitOpenTelemetry(app.Context)
 	if err != nil {
+		// The app is being discarded; stop its background goroutines.
+		if shutdownErr := app.Shutdown(); shutdownErr != nil {
+			err = errors.Join(err, shutdownErr)
+		}
 		return nil, fmt.Errorf("error during open telemetry initialization: %v", err)
 	}
 	if config.HTTP.H2C.Enabled {

tracing/tracing.go

@@ -2,6 +2,7 @@ package tracing
 
 import (
 	"context"
+	"sync"
 
 	"github.com/distribution/distribution/v3/internal/dcontext"
 	"github.com/distribution/distribution/v3/version"
@@ -26,9 +27,33 @@ const (
 	AttributePrefix = "io.cncf.distribution."
 )
 
+var (
+	initMu   sync.Mutex
+	initDone bool
+)
+
 // InitOpenTelemetry initializes OpenTelemetry for the application. This function sets up the
 // necessary components for collecting telemetry data, such as traces.
+//
+// It configures process-global state (the global TracerProvider, error handler, and
+// propagator), so it initializes at most once per process; once it has succeeded,
+// subsequent calls are no-ops (a failed attempt may be retried). This also keeps
+// repeated registry construction (e.g. on config reload) from leaking a
+// BatchSpanProcessor per call.
 func InitOpenTelemetry(ctx context.Context) error {
+	initMu.Lock()
+	defer initMu.Unlock()
+	if initDone {
+		return nil
+	}
+	if err := initOpenTelemetry(ctx); err != nil {
+		return err
+	}
+	initDone = true
+	return nil
+}
+
+func initOpenTelemetry(ctx context.Context) error {
 	res := resource.NewWithAttributes(
 		semconv.SchemaURL,
 		semconv.ServiceNameKey.String(serviceName),