fix: reset leader session state and withdraw backlog gauge on shutdown (ENG-1670)

cmd/api/app.go

@@ -42,6 +42,9 @@ type App struct {
 	tracerProvider *sdktrace.TracerProvider
 	metrics        *observability.Metrics
 	taxonomyRepo   *repository.TaxonomyRepository
+	// enrichmentBacklogDone closes when the enrichment-backlog poller goroutine has returned,
+	// including its deferred cleanup. Nil when the poller was never started (metrics disabled).
+	enrichmentBacklogDone chan struct{}
 }
 
 var (
@@ -609,14 +612,24 @@ func (a *App) Run(ctx context.Context) error {
 	}
 
 	if a.metrics != nil && a.metrics.EnrichmentBacklog != nil {
-		go runEnrichmentBacklogPoller(ctx, a.db, a.metrics.EnrichmentBacklog, enrichmentBacklogPollConfig{
-			// Trim to stay consistent with NewEnrichmentStatusService (config already canonicalizes
-			// this, so it's defensive symmetry) — the endpoint and the gauge resolve the same target.
-			defaultLang:           strings.TrimSpace(a.cfg.Translation.DefaultLanguage),
-			translationConfigured: a.cfg.Translation.Provider != "" && a.cfg.Translation.Model != "",
-			sentimentConfigured:   a.cfg.Sentiment.Enabled(),
-			emotionsConfigured:    a.cfg.Emotions.Enabled(),
-		})
+		// Tracked, unlike the pollers above, because this one has cleanup that Shutdown must not
+		// race: it withdraws the backlog series and releases the leader's advisory lock on the way
+		// out. See App.awaitEnrichmentBacklogPoller.
+		done := make(chan struct{})
+		a.enrichmentBacklogDone = done
+
+		go func() {
+			defer close(done)
+
+			runEnrichmentBacklogPoller(ctx, a.db, a.metrics.EnrichmentBacklog, enrichmentBacklogPollConfig{
+				// Trim to stay consistent with NewEnrichmentStatusService (config already canonicalizes
+				// this, so it's defensive symmetry) — the endpoint and the gauge resolve the same target.
+				defaultLang:           strings.TrimSpace(a.cfg.Translation.DefaultLanguage),
+				translationConfigured: a.cfg.Translation.Provider != "" && a.cfg.Translation.Model != "",
+				sentimentConfigured:   a.cfg.Sentiment.Enabled(),
+				emotionsConfigured:    a.cfg.Emotions.Enabled(),
+			})
+		}()
 	}
 
 	// Reap taxonomy runs orphaned in a non-terminal state, but only when the taxonomy service is wired
@@ -673,6 +686,12 @@ func runEnrichmentBacklogPoller(
 	leader := repository.NewEnrichmentBacklogLeader(db)
 	defer leader.Close(ctx)
 
+	// Withdraw the series whenever this process stops being the leader, including the most common
+	// case of all: shutdown. Close releases the advisory lock, so without this the meter provider's
+	// final collect-and-export on shutdown would publish one last reading for a backlog this
+	// process no longer owns.
+	defer backlog.ClearEnrichmentPending()
+
 	ticker := time.NewTicker(enrichmentBacklogInterval)
 	defer ticker.Stop()
 
@@ -697,13 +716,13 @@ func runEnrichmentBacklogPoller(
 			// than leave it frozen at the last good reading while the new leader publishes its own.
 			backlog.ClearEnrichmentPending()
 
-			// Always count the failure so a stale gauge is alertable, then escalate the log from
-			// warn to error once failures persist: a single blip is noise, a run of them means the
-			// gauge is frozen at its last value and silently lying about the backlog.
+			// Count the failure so the gap is alertable, then escalate the log from warn to error
+			// once failures persist: a single blip is noise, but a run of them means nobody is
+			// publishing the backlog at all.
 			backlog.RecordPollError(ctx)
 
 			if consecutiveFailures >= enrichmentBacklogFailuresBeforeError {
-				slog.ErrorContext(ctx, "enrichment backlog poll failing repeatedly; gauge is stale",
+				slog.ErrorContext(ctx, "enrichment backlog poll failing repeatedly; gauge is not being published",
 					"error", err, "consecutive_failures", consecutiveFailures)
 			} else {
 				slog.WarnContext(ctx, "enrichment backlog poll failed",
@@ -954,6 +973,13 @@ func (a *App) Shutdown(ctx context.Context) (err error) {
 		}
 	}()
 
+	// Registered AFTER the observability defer so LIFO runs it FIRST: the meter provider's final
+	// collect-and-export must not happen until the poller has withdrawn the backlog series, or
+	// shutdown publishes one last reading for a backlog this process no longer owns -- the exact
+	// thing the poller's ClearEnrichmentPending exists to prevent. A defer (not an inline call)
+	// so the early-return error paths below are covered too.
+	defer a.awaitEnrichmentBacklogPoller(ctx)
+
 	if err = a.server.Shutdown(ctx); err != nil && !errors.Is(err, http.ErrServerClosed) {
 		if stopErr := a.river.Stop(ctx); stopErr != nil {
 			slog.Error("river stop during server shutdown", "error", stopErr)
@@ -968,3 +994,28 @@ func (a *App) Shutdown(ctx context.Context) (err error) {
 
 	return nil
 }
+
+// awaitEnrichmentBacklogPoller blocks until the enrichment-backlog poller goroutine has returned,
+// bounded by the shutdown context.
+//
+// Run returns as soon as its context is cancelled and does not join the poller, so without this the
+// two race: Shutdown could reach the meter provider's final collect first and export a backlog
+// reading for a process that is no longer the leader. Waiting also lets the leader's advisory-lock
+// release and idle-timeout reset finish, which previously raced process exit with nothing awaiting
+// them.
+//
+// On deadline this gives up rather than hanging shutdown past its budget -- the stale sample is a
+// cosmetic gauge artifact, and a wedged shutdown is not. Postgres releases the session lock when the
+// socket closes either way.
+func (a *App) awaitEnrichmentBacklogPoller(ctx context.Context) {
+	if a.enrichmentBacklogDone == nil {
+		return
+	}
+
+	select {
+	case <-a.enrichmentBacklogDone:
+	case <-ctx.Done():
+		slog.Warn("enrichment backlog poller did not finish before the shutdown deadline; " +
+			"its final gauge reading may still be exported")
+	}
+}

cmd/api/app_test.go

@@ -7,13 +7,16 @@ import (
 	"net/http/httptest"
 	"os"
 	"path/filepath"
+	"slices"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
 	"github.com/riverqueue/river"
 	"github.com/riverqueue/river/riverdriver/riverpgxv5"
 	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
 	sdktrace "go.opentelemetry.io/otel/sdk/trace"
 
 	"github.com/formbricks/hub/internal/api/handlers"
@@ -182,6 +185,159 @@ func TestShutdownObservabilityWithProviders(t *testing.T) {
 	}
 }
 
+// lifecycleRecorder orders the shutdown steps a test cares about. Both the exporter below and the
+// test goroutine append to it, so it has to be mutex-guarded.
+type lifecycleRecorder struct {
+	mu     sync.Mutex
+	events []string
+}
+
+func (r *lifecycleRecorder) record(event string) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	r.events = append(r.events, event)
+}
+
+func (r *lifecycleRecorder) snapshot() []string {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	return slices.Clone(r.events)
+}
+
+// recordingExporter notes when the meter provider tears down. Its Shutdown is the step that must not
+// happen before the backlog poller has withdrawn its gauge; Export is recorded too so the final
+// collect is visible when instruments carry data.
+type recordingExporter struct {
+	recorder *lifecycleRecorder
+}
+
+func (e recordingExporter) Temporality(kind sdkmetric.InstrumentKind) metricdata.Temporality {
+	return sdkmetric.DefaultTemporalitySelector(kind)
+}
+
+func (e recordingExporter) Aggregation(kind sdkmetric.InstrumentKind) sdkmetric.Aggregation {
+	return sdkmetric.DefaultAggregationSelector(kind)
+}
+
+func (e recordingExporter) Export(_ context.Context, _ *metricdata.ResourceMetrics) error {
+	e.recorder.record("metrics-exported")
+
+	return nil
+}
+
+func (e recordingExporter) ForceFlush(context.Context) error { return nil }
+
+func (e recordingExporter) Shutdown(context.Context) error {
+	e.recorder.record("metrics-shutdown")
+
+	return nil
+}
+
+// TestShutdownWaitsForEnrichmentBacklogPoller pins the lifecycle ordering that makes the poller's
+// deferred ClearEnrichmentPending observable at all.
+//
+// Run returns the moment its context is cancelled without joining the poller, so Shutdown used to
+// race it: the meter provider's final collect-and-export could publish a backlog reading for a
+// process that had already lost leadership -- the exact stale series the clear exists to prevent.
+// The poller is slowed here so an unsynchronized Shutdown reliably loses the race. Verified by
+// deleting the awaitEnrichmentBacklogPoller defer: the recorded events become
+// [metrics-exported metrics-shutdown], i.e. the gauge is exported and the provider torn down before
+// the clear runs at all.
+func TestShutdownWaitsForEnrichmentBacklogPoller(t *testing.T) {
+	recorder := &lifecycleRecorder{}
+
+	riverClient, err := river.NewClient(riverpgxv5.New(nil), &river.Config{})
+	if err != nil {
+		t.Fatalf("river.NewClient() error = %v, want nil", err)
+	}
+
+	done := make(chan struct{})
+
+	app := &App{
+		cfg:     &config.Config{Server: config.ServerConfig{Port: "0"}},
+		server:  &http.Server{Addr: "127.0.0.1:0", ReadHeaderTimeout: time.Second},
+		river:   riverClient,
+		message: service.NewMessagePublisherManager(1, time.Second, nil),
+		meterProvider: sdkmetric.NewMeterProvider(
+			sdkmetric.WithReader(sdkmetric.NewPeriodicReader(recordingExporter{recorder: recorder})),
+		),
+		enrichmentBacklogDone: done,
+	}
+
+	// Stands in for the poller returning: it withdraws the gauge, then signals. The delay is what
+	// makes the ordering assertion meaningful rather than incidentally true.
+	go func() {
+		time.Sleep(50 * time.Millisecond)
+		recorder.record("backlog-cleared")
+		close(done)
+	}()
+
+	shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	if err := app.Shutdown(shutdownCtx); err != nil {
+		t.Fatalf("Shutdown() error = %v, want nil", err)
+	}
+
+	events := recorder.snapshot()
+
+	clearedAt := slices.Index(events, "backlog-cleared")
+	if clearedAt < 0 {
+		t.Fatalf("Shutdown() events = %v, want the backlog clear to have happened", events)
+	}
+
+	shutdownAt := slices.Index(events, "metrics-shutdown")
+	if shutdownAt < 0 {
+		t.Fatalf("Shutdown() events = %v, want the meter provider to have been shut down", events)
+	}
+
+	if clearedAt > shutdownAt {
+		t.Fatalf("Shutdown() events = %v, want the backlog clear before the meter provider teardown", events)
+	}
+}
+
+// TestShutdownWithoutEnrichmentBacklogPollerDoesNotBlock covers the metrics-disabled path, where the
+// poller never started and there is nothing to join.
+func TestShutdownWithoutEnrichmentBacklogPollerDoesNotBlock(t *testing.T) {
+	riverClient, err := river.NewClient(riverpgxv5.New(nil), &river.Config{})
+	if err != nil {
+		t.Fatalf("river.NewClient() error = %v, want nil", err)
+	}
+
+	app := &App{
+		cfg:     &config.Config{Server: config.ServerConfig{Port: "0"}},
+		server:  &http.Server{Addr: "127.0.0.1:0", ReadHeaderTimeout: time.Second},
+		river:   riverClient,
+		message: service.NewMessagePublisherManager(1, time.Second, nil),
+	}
+
+	shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	if err := app.Shutdown(shutdownCtx); err != nil {
+		t.Fatalf("Shutdown() error = %v, want nil", err)
+	}
+}
+
+// TestAwaitEnrichmentBacklogPollerGivesUpOnDeadline pins the bound: a poller stuck in its leader
+// cleanup must not hold shutdown past its budget.
+func TestAwaitEnrichmentBacklogPollerGivesUpOnDeadline(t *testing.T) {
+	app := &App{enrichmentBacklogDone: make(chan struct{})} // never closed
+
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
+	defer cancel()
+
+	start := time.Now()
+
+	app.awaitEnrichmentBacklogPoller(ctx)
+
+	if elapsed := time.Since(start); elapsed > time.Second {
+		t.Fatalf("awaitEnrichmentBacklogPoller() blocked for %v, want it to give up on the deadline", elapsed)
+	}
+}
+
 func TestAppRunReturnsServerError(t *testing.T) {
 	app := &App{
 		cfg: &config.Config{

internal/observability/enrichment_backlog.go

@@ -31,9 +31,10 @@ type EnrichmentBacklogMetrics interface {
 	// would look like a permanently stuck backlog. Callers must clear as soon as they are no longer
 	// the leader. Exporting nothing is the honest state — absence is visible, a stale value is not.
 	ClearEnrichmentPending()
-	// RecordPollError counts a failed refresh. The gauge holds its last value when a poll fails,
-	// which is indistinguishable from a healthy steady backlog on a dashboard — this counter is
-	// the signal to alert on (rate > 0 means the gauge is going stale).
+	// RecordPollError counts a failed refresh. A failed poll costs this process its leadership and
+	// withdraws the series (see ClearEnrichmentPending), so the symptom is a MISSING gauge rather
+	// than a stale value: alert on this counter, or on the absence of the gauge — a value-based
+	// staleness rule will not catch it.
 	RecordPollError(ctx context.Context)
 }
 
@@ -84,9 +85,10 @@ func NewEnrichmentBacklogMetrics(meter metric.Meter) (EnrichmentBacklogMetrics,
 	pollErrors, err := meter.Int64Counter(
 		MetricNameEnrichmentBacklogPollErrs,
 		metric.WithDescription(
-			"Failed refreshes of the enrichment backlog gauge. A non-zero rate means the gauge is "+
-				"stale (holding its last value), which on a dashboard is indistinguishable from a "+
-				"steady backlog — alert on this rather than trusting a flat gauge.",
+			"Failed refreshes of the enrichment backlog gauge. A failed poll costs the process its "+
+				"leadership and withdraws the series, so the symptom is a MISSING "+
+				MetricNameEnrichmentPendingRecords+" rather than a stale value — alert on this "+
+				"counter, or on absence of that gauge.",
 		),
 		metric.WithUnit("1"),
 	)

internal/observability/enrichment_backlog_test.go

@@ -60,12 +60,25 @@ func TestEnrichmentBacklogMetricsClearWithdrawsSeries(t *testing.T) {
 	var collected metricdata.ResourceMetrics
 	require.NoError(t, reader.Collect(context.Background(), &collected))
 
+	// Count rather than iterate-and-assert: after the clear the SDK drops the empty metric and then
+	// the empty scope, so a loop over ScopeMetrics runs zero times and would pass vacuously (it
+	// would also "pass" if the metric name were wrong, or if Collect returned nothing at all).
+	points := 0
+
 	for _, scope := range collected.ScopeMetrics {
 		for _, m := range scope.Metrics {
-			assert.NotEqual(t, MetricNameEnrichmentPendingRecords, m.Name,
-				"a cleared gauge must export no data points at all, not a stale value")
+			if m.Name != MetricNameEnrichmentPendingRecords {
+				continue
+			}
+
+			gauge, ok := m.Data.(metricdata.Gauge[int64])
+			require.True(t, ok, "expected Gauge[int64] for %s", MetricNameEnrichmentPendingRecords)
+
+			points += len(gauge.DataPoints)
 		}
 	}
+
+	assert.Zero(t, points, "a cleared gauge must export no data points at all, not a stale value")
 }
 
 // TestEnrichmentBacklogMetricsPollErrors verifies failed refreshes are counted, so a gauge frozen