From ed88be8555d482e3fcc06c9909d0a5153ab5ac95 Mon Sep 17 00:00:00 2001 From: Tiago Farto Date: Fri, 24 Jul 2026 16:22:03 +0000 Subject: [PATCH 01/13] feat(enrichment): add GET /v1/enrichment-status endpoint (ENG-1670) Expose per-tenant, data-derived enrichment progress so the UI can surface an "in progress" indicator, mirroring the embeddings UX (eligible vs done counts, not queue depth). GET /v1/enrichment-status?tenant_id= returns {enabled, eligible, done} per enrichment (translation, sentiment, emotions) counted over feedback_records: - eligibility is text-with-content (field_type='text', btrim charset matches the workers' TrimSpace), gated by the per-tenant switch (sentiment/emotions) or a resolvable target language (translation); - "done" for translation means the stored lang key equals the effective target (stale translations are not done); - the deployment-level provider/model gate is overlaid in the service. tenant_id is required and validated; the response is counts-only (no record ids/content); the query is parameterized and strictly tenant-scoped. --- cmd/api/app.go | 16 +- cmd/api/app_test.go | 1 + .../api/handlers/enrichment_status_handler.go | 47 ++++++ .../enrichment_status_handler_test.go | 66 ++++++++ internal/models/enrichment_status.go | 23 +++ .../enrichment_status_repository.go | 91 ++++++++++ internal/service/enrichment_status_service.go | 104 ++++++++++++ .../service/enrichment_status_service_test.go | 155 ++++++++++++++++++ tests/enrichment_status_test.go | 153 +++++++++++++++++ 9 files changed, 655 insertions(+), 1 deletion(-) create mode 100644 internal/api/handlers/enrichment_status_handler.go create mode 100644 internal/api/handlers/enrichment_status_handler_test.go create mode 100644 internal/models/enrichment_status.go create mode 100644 internal/repository/enrichment_status_repository.go create mode 100644 internal/service/enrichment_status_service.go create mode 100644 internal/service/enrichment_status_service_test.go create mode 100644 tests/enrichment_status_test.go diff --git a/cmd/api/app.go b/cmd/api/app.go index f837f235..749b8592 100644 --- a/cmd/api/app.go +++ b/cmd/api/app.go @@ -545,6 +545,17 @@ func NewApp(cfg *config.Config, db *pgxpool.Pool) (*App, error) { taxonomyHandler := handlers.NewTaxonomyHandler(taxonomyService) feedbackRecordsHandler := handlers.NewFeedbackRecordsHandler(feedbackRecordsService) taxonomyInternalHandler := handlers.NewTaxonomyInternalHandler(taxonomyService) + + enrichmentStatusService := service.NewEnrichmentStatusService(service.NewEnrichmentStatusServiceParams{ + Repo: repository.NewEnrichmentStatusRepository(db), + Settings: tenantSettingsService, + DefaultLang: cfg.Translation.DefaultLanguage, + TranslationConfigured: cfg.Translation.Provider != "" && cfg.Translation.Model != "", + SentimentConfigured: cfg.Sentiment.Enabled(), + EmotionsConfigured: cfg.Emotions.Enabled(), + }) + enrichmentStatusHandler := handlers.NewEnrichmentStatusHandler(enrichmentStatusService) + healthHandler := handlers.NewHealthHandler() openapiHandler, err := handlers.NewOpenAPIHandler(handlers.ResolveOpenAPISpecPath(), cfg.Server.PublicBaseURL) @@ -557,7 +568,7 @@ func NewApp(cfg *config.Config, db *pgxpool.Pool) (*App, error) { server := newHTTPServer( cfg, healthHandler, openapiHandler, feedbackRecordsHandler, webhooksHandler, tenantDataHandler, tenantSettingsHandler, searchHandler, - taxonomyHandler, taxonomyInternalHandler, + taxonomyHandler, taxonomyInternalHandler, enrichmentStatusHandler, meterProvider, tracerProvider, ) @@ -588,6 +599,7 @@ func newHTTPServer( search *handlers.SearchHandler, taxonomy *handlers.TaxonomyHandler, taxonomyInternal *handlers.TaxonomyInternalHandler, + enrichmentStatus *handlers.EnrichmentStatusHandler, meterProvider *sdkmetric.MeterProvider, tracerProvider *sdktrace.TracerProvider, ) *http.Server { @@ -615,6 +627,8 @@ func newHTTPServer( protected.HandleFunc("PUT /v1/tenants/{tenant_id}/settings", tenantSettings.Update) protected.HandleFunc("PATCH /v1/tenants/{tenant_id}/settings", tenantSettings.Patch) + protected.HandleFunc("GET /v1/enrichment-status", enrichmentStatus.GetStatus) + // Search endpoints are always registered; when embeddings are disabled, the handler returns 503. protected.HandleFunc("POST /v1/feedback-records/search/semantic", search.SemanticSearch) protected.HandleFunc("GET /v1/feedback-records/{id}/similar", search.SimilarFeedback) diff --git a/cmd/api/app_test.go b/cmd/api/app_test.go index e2337745..075f4c83 100644 --- a/cmd/api/app_test.go +++ b/cmd/api/app_test.go @@ -380,6 +380,7 @@ func newTestHTTPServerWithConfig(t *testing.T, publicBaseURL string, taxonomy co handlers.NewSearchHandler(nil), handlers.NewTaxonomyHandler(nil), handlers.NewTaxonomyInternalHandler(), + handlers.NewEnrichmentStatusHandler(nil), nil, nil, ) diff --git a/internal/api/handlers/enrichment_status_handler.go b/internal/api/handlers/enrichment_status_handler.go new file mode 100644 index 00000000..47885fe9 --- /dev/null +++ b/internal/api/handlers/enrichment_status_handler.go @@ -0,0 +1,47 @@ +package handlers + +import ( + "context" + "net/http" + + "github.com/formbricks/hub/internal/api/response" + "github.com/formbricks/hub/internal/models" +) + +// EnrichmentStatusService is the application service used by the enrichment status handler. +type EnrichmentStatusService interface { + GetEnrichmentStatus(ctx context.Context, tenantID string) (*models.EnrichmentStatusResponse, error) +} + +// EnrichmentStatusHandler hosts the public enrichment-status endpoint. +type EnrichmentStatusHandler struct { + service EnrichmentStatusService +} + +// NewEnrichmentStatusHandler creates an enrichment status handler. +func NewEnrichmentStatusHandler(service EnrichmentStatusService) *EnrichmentStatusHandler { + return &EnrichmentStatusHandler{service: service} +} + +// GetStatus handles GET /v1/enrichment-status?tenant_id=. It returns per-enrichment +// eligible/done counts for the tenant. tenant_id is required; a missing/blank value yields 400. +func (h *EnrichmentStatusHandler) GetStatus(w http.ResponseWriter, r *http.Request) { + if h.service == nil { + response.RespondServiceUnavailable(w, r, "Enrichment status is not available.") + + return + } + + tenantID := r.URL.Query().Get("tenant_id") + + result, err := h.service.GetEnrichmentStatus(r.Context(), tenantID) + if err != nil { + // RespondError maps a missing/blank tenant_id (huberrors.ValidationError) to 400 and + // any repository/DB failure to a generic 500 — no internals leak to the client. + response.RespondError(w, r, err) + + return + } + + response.RespondJSON(w, http.StatusOK, result) +} diff --git a/internal/api/handlers/enrichment_status_handler_test.go b/internal/api/handlers/enrichment_status_handler_test.go new file mode 100644 index 00000000..f8cc0885 --- /dev/null +++ b/internal/api/handlers/enrichment_status_handler_test.go @@ -0,0 +1,66 @@ +package handlers + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/formbricks/hub/internal/huberrors" + "github.com/formbricks/hub/internal/models" +) + +type fakeEnrichmentStatusService struct { + resp *models.EnrichmentStatusResponse + err error +} + +func (f *fakeEnrichmentStatusService) GetEnrichmentStatus( + _ context.Context, _ string, +) (*models.EnrichmentStatusResponse, error) { + return f.resp, f.err +} + +func TestEnrichmentStatusHandler_GetStatus(t *testing.T) { + t.Run("service unavailable when not configured", func(t *testing.T) { + h := NewEnrichmentStatusHandler(nil) + rec := httptest.NewRecorder() + h.GetStatus(rec, httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/v1/enrichment-status?tenant_id=t1", nil)) + + assert.Equal(t, http.StatusServiceUnavailable, rec.Code) + }) + + t.Run("missing tenant_id maps to 400", func(t *testing.T) { + // The service returns a huberrors.ValidationError for a blank tenant_id; the handler must + // surface it as a 400 via response.RespondError, not a 500. + h := NewEnrichmentStatusHandler(&fakeEnrichmentStatusService{ + err: huberrors.NewValidationError("tenant_id", "tenant_id is required and cannot be empty"), + }) + rec := httptest.NewRecorder() + h.GetStatus(rec, httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/v1/enrichment-status", nil)) + + assert.Equal(t, http.StatusBadRequest, rec.Code) + }) + + t.Run("returns 200 with the status body", func(t *testing.T) { + want := &models.EnrichmentStatusResponse{ + TenantID: "t1", + Translation: models.EnrichmentTypeStatus{Enabled: true, Eligible: 10, Done: 4}, + Sentiment: models.EnrichmentTypeStatus{Enabled: true, Eligible: 8, Done: 8}, + Emotions: models.EnrichmentTypeStatus{Enabled: false}, + } + h := NewEnrichmentStatusHandler(&fakeEnrichmentStatusService{resp: want}) + rec := httptest.NewRecorder() + h.GetStatus(rec, httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/v1/enrichment-status?tenant_id=t1", nil)) + + require.Equal(t, http.StatusOK, rec.Code) + + var got models.EnrichmentStatusResponse + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &got)) + assert.Equal(t, *want, got) + }) +} diff --git a/internal/models/enrichment_status.go b/internal/models/enrichment_status.go new file mode 100644 index 00000000..6ca3bc53 --- /dev/null +++ b/internal/models/enrichment_status.go @@ -0,0 +1,23 @@ +package models + +// EnrichmentTypeStatus is one tenant's progress for a single record-level enrichment +// (translation, sentiment, or emotions). Enabled reports whether the enrichment is both +// deployment-configured and switched on for the tenant (for translation: has a resolvable +// target language). Eligible is the number of feedback records that qualify for the +// enrichment; Done is how many have been enriched. The UI derives "in progress" as +// Eligible - Done. When Enabled is false, Eligible and Done are zero (no work will run). +type EnrichmentTypeStatus struct { + Enabled bool `json:"enabled"` + Eligible int64 `json:"eligible"` + Done int64 `json:"done"` +} + +// EnrichmentStatusResponse reports a tenant's enrichment progress across the record-level +// enrichments. Counts are directory-level totals; the response never includes record +// identifiers or feedback content. +type EnrichmentStatusResponse struct { + TenantID string `json:"tenant_id"` + Translation EnrichmentTypeStatus `json:"translation"` + Sentiment EnrichmentTypeStatus `json:"sentiment"` + Emotions EnrichmentTypeStatus `json:"emotions"` +} diff --git a/internal/repository/enrichment_status_repository.go b/internal/repository/enrichment_status_repository.go new file mode 100644 index 00000000..c1e6dd8a --- /dev/null +++ b/internal/repository/enrichment_status_repository.go @@ -0,0 +1,91 @@ +package repository + +import ( + "context" + "fmt" + + "github.com/jackc/pgx/v5/pgxpool" +) + +// EnrichmentStatusRepository computes data-derived enrichment progress counts over +// feedback_records. It is a read-only sibling of FeedbackRecordsRepository, kept separate +// so the status/observability concern doesn't grow the primary records repository. +type EnrichmentStatusRepository struct { + db *pgxpool.Pool +} + +// NewEnrichmentStatusRepository creates an enrichment status repository. +func NewEnrichmentStatusRepository(db *pgxpool.Pool) *EnrichmentStatusRepository { + return &EnrichmentStatusRepository{db: db} +} + +// EnrichmentStatusCounts holds the raw per-enrichment eligible/done counts for a tenant. +// The counts already reflect the per-tenant gates: sentiment/emotions counts include only +// tenants with the enrichment switched on, translation counts only records with a resolvable +// effective target language. The deployment-level (provider/model) gate is applied by the caller. +type EnrichmentStatusCounts struct { + TranslationEligible int64 + TranslationDone int64 + SentimentEligible int64 + SentimentDone int64 + EmotionsEligible int64 + EmotionsDone int64 +} + +// enrichmentEligibleText is the data-level eligibility predicate: an open-text field with +// content. It mirrors the backfill eligibility (see classifyBackfillEligibleSQL / +// translationBackfillSelectSQL) but uses the fuller btrim charset E' \t\r\n' so a whitespace-only +// value_text ("\t", "\n") is treated as empty — matching the workers' Go strings.TrimSpace content +// gate (HasOpenText). field_type = 'text' is load-bearing: matrix/multi-choice expansion writes +// value_text on categorical/number rows that are not enrichable. +const enrichmentEligibleText = `fr.field_type = 'text' AND fr.value_text IS NOT NULL AND btrim(fr.value_text, E' \t\r\n') <> ''` + +// enrichmentEffectiveTarget resolves a tenant's effective translation target: its own +// target_language, falling back to the deployment default ($N). An empty result means translation +// is not enabled for the tenant. Mirrors translationBackfillSelectSQL. +// sentimentEnabled / emotionsEnabled read the tri-state per-directory switch, defaulting to +// enabled when the key is absent — matching EnrichmentSettings.SentimentEnrichmentEnabled / +// EmotionsEnrichmentEnabled (parity is covered by test). +const ( + enrichmentEffectiveTarget = `COALESCE(NULLIF(ts.settings->>'target_language', ''), $2)` + enrichmentSentimentOn = `COALESCE((ts.settings->>'sentiment_enabled')::boolean, true)` + enrichmentEmotionsOn = `COALESCE((ts.settings->>'emotions_enabled')::boolean, true)` +) + +// countEnrichmentStatusSQL counts eligible/done per enrichment for one tenant in a single pass. +// $1 = tenant_id, $2 = deployment default target language. All predicate fragments are static +// constants (never user input); tenant_id is bound. +const countEnrichmentStatusSQL = ` + SELECT + COUNT(*) FILTER (WHERE ` + enrichmentEligibleText + ` AND ` + enrichmentSentimentOn + `), + COUNT(*) FILTER (WHERE ` + enrichmentEligibleText + ` AND ` + enrichmentSentimentOn + ` AND fr.sentiment IS NOT NULL), + COUNT(*) FILTER (WHERE ` + enrichmentEligibleText + ` AND ` + enrichmentEmotionsOn + `), + COUNT(*) FILTER (WHERE ` + enrichmentEligibleText + ` AND ` + enrichmentEmotionsOn + ` AND fr.emotions IS NOT NULL), + COUNT(*) FILTER (WHERE ` + enrichmentEligibleText + ` AND ` + enrichmentEffectiveTarget + ` <> ''), + COUNT(*) FILTER ( + WHERE ` + enrichmentEligibleText + ` + AND ` + enrichmentEffectiveTarget + ` <> '' + AND fr.translation_lang_key = ` + enrichmentEffectiveTarget + `) + FROM feedback_records fr + LEFT JOIN tenant_settings ts ON ts.tenant_id = fr.tenant_id + WHERE fr.tenant_id = $1` + +// CountEnrichmentStatus returns one tenant's eligible/done counts per enrichment. defaultLang is +// the deployment translation fallback ("" disables the fallback, so only tenants with their own +// target language have eligible translation records). Always scoped to the given tenant_id. +func (r *EnrichmentStatusRepository) CountEnrichmentStatus( + ctx context.Context, tenantID, defaultLang string, +) (EnrichmentStatusCounts, error) { + var counts EnrichmentStatusCounts + + err := r.db.QueryRow(ctx, countEnrichmentStatusSQL, tenantID, defaultLang).Scan( + &counts.SentimentEligible, &counts.SentimentDone, + &counts.EmotionsEligible, &counts.EmotionsDone, + &counts.TranslationEligible, &counts.TranslationDone, + ) + if err != nil { + return EnrichmentStatusCounts{}, fmt.Errorf("count enrichment status: %w", err) + } + + return counts, nil +} diff --git a/internal/service/enrichment_status_service.go b/internal/service/enrichment_status_service.go new file mode 100644 index 00000000..2d9af72c --- /dev/null +++ b/internal/service/enrichment_status_service.go @@ -0,0 +1,104 @@ +package service + +import ( + "context" + "fmt" + "strings" + "time" + + "github.com/formbricks/hub/internal/models" + "github.com/formbricks/hub/internal/repository" +) + +// enrichmentStatusQueryTimeout bounds the settings lookup + count query for one request. The +// endpoint is polled and there is no HTTP handler timeout, so a slow count must not pin a DB +// pool connection indefinitely. +const enrichmentStatusQueryTimeout = 5 * time.Second + +// EnrichmentStatusRepository is the count surface the status service needs. +type EnrichmentStatusRepository interface { + CountEnrichmentStatus(ctx context.Context, tenantID, defaultLang string) (repository.EnrichmentStatusCounts, error) +} + +// EnrichmentStatusService reports a tenant's enrichment progress. It resolves the per-tenant +// enable state from settings (via TenantSettingsReader) and overlays the deployment-level +// (provider/model) gate; the repo supplies the counts. +type EnrichmentStatusService struct { + repo EnrichmentStatusRepository + settings TenantSettingsReader + defaultLang string + translationConfigured bool + sentimentConfigured bool + emotionsConfigured bool +} + +// NewEnrichmentStatusServiceParams configures an EnrichmentStatusService. The *Configured flags +// are the deployment-level gates (provider+model set), mirroring how the enrichment providers are +// constructed; defaultLang is TRANSLATION_DEFAULT_LANGUAGE. +type NewEnrichmentStatusServiceParams struct { + Repo EnrichmentStatusRepository + Settings TenantSettingsReader + DefaultLang string + TranslationConfigured bool + SentimentConfigured bool + EmotionsConfigured bool +} + +// NewEnrichmentStatusService creates an enrichment status service. +func NewEnrichmentStatusService(params NewEnrichmentStatusServiceParams) *EnrichmentStatusService { + return &EnrichmentStatusService{ + repo: params.Repo, + settings: params.Settings, + defaultLang: strings.TrimSpace(params.DefaultLang), + translationConfigured: params.TranslationConfigured, + sentimentConfigured: params.SentimentConfigured, + emotionsConfigured: params.EmotionsConfigured, + } +} + +// GetEnrichmentStatus returns the tenant's per-enrichment progress. tenant_id is required and +// validated; the query is scoped to that tenant alone. +func (s *EnrichmentStatusService) GetEnrichmentStatus( + ctx context.Context, tenantID string, +) (*models.EnrichmentStatusResponse, error) { + normalizedTenantID, err := normalizeRequiredTenantIDValue(tenantID) + if err != nil { + return nil, err + } + + ctx, cancel := context.WithTimeout(ctx, enrichmentStatusQueryTimeout) + defer cancel() + + settings, err := s.settings.GetSettings(ctx, normalizedTenantID) + if err != nil { + return nil, fmt.Errorf("resolve tenant settings: %w", err) + } + + counts, err := s.repo.CountEnrichmentStatus(ctx, normalizedTenantID, s.defaultLang) + if err != nil { + return nil, fmt.Errorf("count enrichment status: %w", err) + } + + translationEnabled := s.translationConfigured && + resolveTargetLang(settings.Settings.TargetLanguage, s.defaultLang) != "" + + return &models.EnrichmentStatusResponse{ + TenantID: normalizedTenantID, + Translation: enrichmentTypeStatus(translationEnabled, + counts.TranslationEligible, counts.TranslationDone), + Sentiment: enrichmentTypeStatus(s.sentimentConfigured && settings.Settings.SentimentEnrichmentEnabled(), + counts.SentimentEligible, counts.SentimentDone), + Emotions: enrichmentTypeStatus(s.emotionsConfigured && settings.Settings.EmotionsEnrichmentEnabled(), + counts.EmotionsEligible, counts.EmotionsDone), + }, nil +} + +// enrichmentTypeStatus assembles one enrichment's status, zeroing the counts when it is not +// enabled for the tenant so the API never reports a backlog for work that will never run. +func enrichmentTypeStatus(enabled bool, eligible, done int64) models.EnrichmentTypeStatus { + if !enabled { + return models.EnrichmentTypeStatus{Enabled: false} + } + + return models.EnrichmentTypeStatus{Enabled: true, Eligible: eligible, Done: done} +} diff --git a/internal/service/enrichment_status_service_test.go b/internal/service/enrichment_status_service_test.go new file mode 100644 index 00000000..0193a9fb --- /dev/null +++ b/internal/service/enrichment_status_service_test.go @@ -0,0 +1,155 @@ +package service + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/formbricks/hub/internal/huberrors" + "github.com/formbricks/hub/internal/models" + "github.com/formbricks/hub/internal/repository" +) + +type fakeStatusRepo struct { + counts repository.EnrichmentStatusCounts + err error + gotTenantID string + gotDefaultLang string +} + +func (f *fakeStatusRepo) CountEnrichmentStatus( + _ context.Context, tenantID, defaultLang string, +) (repository.EnrichmentStatusCounts, error) { + f.gotTenantID = tenantID + f.gotDefaultLang = defaultLang + + return f.counts, f.err +} + +type fakeSettingsResolver struct { + settings *models.TenantSettings + err error +} + +func (f *fakeSettingsResolver) GetSettings(_ context.Context, tenantID string) (*models.TenantSettings, error) { + if f.err != nil { + return nil, f.err + } + + if f.settings != nil { + return f.settings, nil + } + + return &models.TenantSettings{TenantID: tenantID}, nil +} + +// fullCounts is a repo result with a distinct value in every bucket, so a test that expects a +// bucket to be zeroed proves the SERVICE zeroed it (not that the repo happened to return 0). +var fullCounts = repository.EnrichmentStatusCounts{ + TranslationEligible: 10, TranslationDone: 4, + SentimentEligible: 8, SentimentDone: 3, + EmotionsEligible: 6, EmotionsDone: 2, +} + +func TestEnrichmentStatusService_GetEnrichmentStatus(t *testing.T) { + emotionsOff := false + + cases := []struct { + name string + params NewEnrichmentStatusServiceParams + settings *models.TenantSettings + wantTrans models.EnrichmentTypeStatus + wantSent models.EnrichmentTypeStatus + wantEmotion models.EnrichmentTypeStatus + }{ + { + name: "all configured and enabled", + params: NewEnrichmentStatusServiceParams{ + DefaultLang: "en-US", TranslationConfigured: true, SentimentConfigured: true, EmotionsConfigured: true, + }, + settings: &models.TenantSettings{Settings: models.EnrichmentSettings{TargetLanguage: "de-DE"}}, + wantTrans: models.EnrichmentTypeStatus{Enabled: true, Eligible: 10, Done: 4}, + wantSent: models.EnrichmentTypeStatus{Enabled: true, Eligible: 8, Done: 3}, + wantEmotion: models.EnrichmentTypeStatus{Enabled: true, Eligible: 6, Done: 2}, + }, + { + name: "sentiment not deployment-configured is zeroed", + params: NewEnrichmentStatusServiceParams{ + DefaultLang: "en-US", TranslationConfigured: true, SentimentConfigured: false, EmotionsConfigured: true, + }, + settings: &models.TenantSettings{Settings: models.EnrichmentSettings{TargetLanguage: "de-DE"}}, + wantTrans: models.EnrichmentTypeStatus{Enabled: true, Eligible: 10, Done: 4}, + wantSent: models.EnrichmentTypeStatus{Enabled: false}, + wantEmotion: models.EnrichmentTypeStatus{Enabled: true, Eligible: 6, Done: 2}, + }, + { + name: "tenant emotions switch off is zeroed", + params: NewEnrichmentStatusServiceParams{ + DefaultLang: "en-US", TranslationConfigured: true, SentimentConfigured: true, EmotionsConfigured: true, + }, + settings: &models.TenantSettings{Settings: models.EnrichmentSettings{ + TargetLanguage: "de-DE", EmotionsEnabled: &emotionsOff, + }}, + wantTrans: models.EnrichmentTypeStatus{Enabled: true, Eligible: 10, Done: 4}, + wantSent: models.EnrichmentTypeStatus{Enabled: true, Eligible: 8, Done: 3}, + wantEmotion: models.EnrichmentTypeStatus{Enabled: false}, + }, + { + name: "translation with no target and no default is zeroed", + params: NewEnrichmentStatusServiceParams{ + DefaultLang: "", TranslationConfigured: true, SentimentConfigured: true, EmotionsConfigured: true, + }, + settings: &models.TenantSettings{Settings: models.EnrichmentSettings{}}, + wantTrans: models.EnrichmentTypeStatus{Enabled: false}, + wantSent: models.EnrichmentTypeStatus{Enabled: true, Eligible: 8, Done: 3}, + wantEmotion: models.EnrichmentTypeStatus{Enabled: true, Eligible: 6, Done: 2}, + }, + { + name: "translation enabled via default-language fallback", + params: NewEnrichmentStatusServiceParams{ + DefaultLang: "en-US", TranslationConfigured: true, SentimentConfigured: true, EmotionsConfigured: true, + }, + settings: &models.TenantSettings{Settings: models.EnrichmentSettings{}}, + wantTrans: models.EnrichmentTypeStatus{Enabled: true, Eligible: 10, Done: 4}, + wantSent: models.EnrichmentTypeStatus{Enabled: true, Eligible: 8, Done: 3}, + wantEmotion: models.EnrichmentTypeStatus{Enabled: true, Eligible: 6, Done: 2}, + }, + } + + for _, testCase := range cases { + t.Run(testCase.name, func(t *testing.T) { + repo := &fakeStatusRepo{counts: fullCounts} + params := testCase.params + params.Repo = repo + params.Settings = &fakeSettingsResolver{settings: testCase.settings} + + svc := NewEnrichmentStatusService(params) + + got, err := svc.GetEnrichmentStatus(context.Background(), " tenant-1 ") + require.NoError(t, err) + + assert.Equal(t, "tenant-1", got.TenantID, "tenant_id is normalized (trimmed)") + assert.Equal(t, "tenant-1", repo.gotTenantID, "repo receives the normalized tenant_id") + assert.Equal(t, testCase.wantTrans, got.Translation) + assert.Equal(t, testCase.wantSent, got.Sentiment) + assert.Equal(t, testCase.wantEmotion, got.Emotions) + }) + } +} + +func TestEnrichmentStatusService_RequiresTenantID(t *testing.T) { + svc := NewEnrichmentStatusService(NewEnrichmentStatusServiceParams{ + Repo: &fakeStatusRepo{}, + Settings: &fakeSettingsResolver{}, + }) + + _, err := svc.GetEnrichmentStatus(context.Background(), " ") + require.Error(t, err) + + // normalizeRequiredTenantIDValue returns a *huberrors.ValidationError, which the response + // layer maps to a client 400 rather than a generic 500. + var validationErr *huberrors.ValidationError + assert.ErrorAs(t, err, &validationErr, "missing tenant_id must be a validation error") +} diff --git a/tests/enrichment_status_test.go b/tests/enrichment_status_test.go new file mode 100644 index 00000000..6abeaf3f --- /dev/null +++ b/tests/enrichment_status_test.go @@ -0,0 +1,153 @@ +package tests + +import ( + "context" + "testing" + + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/formbricks/hub/internal/config" + "github.com/formbricks/hub/internal/models" + "github.com/formbricks/hub/internal/repository" + "github.com/formbricks/hub/pkg/database" +) + +// TestCountEnrichmentStatus exercises the data-derived enrichment-status counts against Postgres, +// locking the gap-analysis edge cases: only text records with real content are eligible +// (field_type gate + whitespace trim), translation "done" means the stored lang key equals the +// effective target (stale != done), the sentiment/emotions per-tenant switch gates eligibility, +// the translation default-language fallback, and strict per-tenant isolation. +func TestCountEnrichmentStatus(t *testing.T) { + ctx := context.Background() + + cfg, err := config.Load() + require.NoError(t, err) + + db, err := database.NewPostgresPool(ctx, cfg.Database.URL, database.WithPoolConfig(cfg.Database.PoolConfig())) + require.NoError(t, err) + + defer db.Close() + + statusRepo := repository.NewEnrichmentStatusRepository(db) + frepo := repository.NewFeedbackRecordsRepository(db) + tsRepo := repository.NewTenantSettingsRepository(db) + + // mkRecord inserts one feedback record for a tenant with an explicit field type + value_text. + mkRecord := func(tenant string, fieldType models.FieldType, valueText string) *models.FeedbackRecord { + vt := valueText + rec, createErr := frepo.Create(ctx, &models.CreateFeedbackRecordRequest{ + SourceType: "formbricks", + FieldID: "q1", + FieldType: fieldType, + ValueText: &vt, + TenantID: tenant, + SubmissionID: testTenantID("sub"), + }) + require.NoError(t, createErr) + + return rec + } + + setSentiment := func(id uuid.UUID) { + label := models.SentimentPositive + score := 1.0 + require.NoError(t, frepo.SetSentiment(ctx, id, &label, &score, nil)) + } + setEmotions := func(id uuid.UUID) { + require.NoError(t, frepo.SetEmotions(ctx, id, []models.EmotionValue{models.EmotionJoy}, nil)) + } + // setTranslationLang writes the translation columns directly to a given lang key, so a test can + // stage both "done" (key == effective target) and "stale" (key != target) states precisely. + setTranslationLang := func(id uuid.UUID, langKey string) { + translated := "translated text" + _, execErr := db.Exec(ctx, + `UPDATE feedback_records SET value_text_translated = $1, translation_lang_key = $2 WHERE id = $3`, + translated, langKey, id) + require.NoError(t, execErr) + } + + t.Run("eligibility, done buckets, and translation staleness", func(t *testing.T) { + tenant := testTenantID("enrich-status-main") + _, err := tsRepo.Upsert(ctx, tenant, models.EnrichmentSettings{TargetLanguage: "de-DE"}) + require.NoError(t, err) + + // Three eligible text records with content. The first stays fully un-enriched (pending in + // every bucket); the other two are enriched below. + mkRecord(tenant, models.FieldTypeText, "great product, would recommend") + doneAll := mkRecord(tenant, models.FieldTypeText, "fully enriched record") + staleTrans := mkRecord(tenant, models.FieldTypeText, "sentiment set, translation stale") + + // doneAll: sentiment + emotions set, translated to the effective target. + setSentiment(doneAll.ID) + setEmotions(doneAll.ID) + setTranslationLang(doneAll.ID, "de-DE") + + // staleTrans: sentiment set (done), emotions NULL (pending), translated to a DIFFERENT + // language than the tenant's target — so translation is NOT done. + setSentiment(staleTrans.ID) + setTranslationLang(staleTrans.ID, "fr-FR") + + // Ineligible rows that must NOT be counted: + mkRecord(tenant, models.FieldTypeCategorical, "some choice") // non-text carries value_text + mkRecord(tenant, models.FieldTypeText, "\t\n ") // whitespace-only content + mkRecord(tenant, models.FieldTypeText, "") // empty content + + counts, err := statusRepo.CountEnrichmentStatus(ctx, tenant, "") + require.NoError(t, err) + + assert.Equal(t, int64(3), counts.SentimentEligible, "3 text records with content are eligible") + assert.Equal(t, int64(2), counts.SentimentDone, "doneAll + staleTrans have sentiment") + assert.Equal(t, int64(3), counts.EmotionsEligible) + assert.Equal(t, int64(1), counts.EmotionsDone, "only doneAll has emotions") + assert.Equal(t, int64(3), counts.TranslationEligible, "all 3 have the effective target de-DE") + assert.Equal(t, int64(1), counts.TranslationDone, "only doneAll's lang key matches; fr-FR is stale") + }) + + t.Run("sentiment switch off zeroes sentiment eligibility, emotions unaffected", func(t *testing.T) { + tenant := testTenantID("enrich-status-switch") + off := false + _, err := tsRepo.Upsert(ctx, tenant, models.EnrichmentSettings{SentimentEnabled: &off}) + require.NoError(t, err) + + mkRecord(tenant, models.FieldTypeText, "a") + mkRecord(tenant, models.FieldTypeText, "b") + + counts, err := statusRepo.CountEnrichmentStatus(ctx, tenant, "") + require.NoError(t, err) + + assert.Equal(t, int64(0), counts.SentimentEligible, "sentiment disabled → not eligible") + assert.Equal(t, int64(2), counts.EmotionsEligible, "emotions default-enabled → still eligible") + }) + + t.Run("translation eligibility follows the effective target", func(t *testing.T) { + tenant := testTenantID("enrich-status-trans") + // No tenant settings row at all → no own target language. + mkRecord(tenant, models.FieldTypeText, "x") + mkRecord(tenant, models.FieldTypeText, "y") + + noDefault, err := statusRepo.CountEnrichmentStatus(ctx, tenant, "") + require.NoError(t, err) + assert.Equal(t, int64(0), noDefault.TranslationEligible, "no target + no default → not eligible") + + withDefault, err := statusRepo.CountEnrichmentStatus(ctx, tenant, "en-US") + require.NoError(t, err) + assert.Equal(t, int64(2), withDefault.TranslationEligible, "default-language fallback makes records eligible") + }) + + t.Run("counts are strictly per-tenant", func(t *testing.T) { + tenantA := testTenantID("enrich-status-iso-a") + tenantB := testTenantID("enrich-status-iso-b") + + mkRecord(tenantA, models.FieldTypeText, "a1") + + for range 5 { + mkRecord(tenantB, models.FieldTypeText, "b") + } + + counts, err := statusRepo.CountEnrichmentStatus(ctx, tenantA, "") + require.NoError(t, err) + assert.Equal(t, int64(1), counts.SentimentEligible, "tenant A sees only its own record, not tenant B's") + }) +} From fffe83e3fc6a728d38860d8d8bccfe2780b85fdb Mon Sep 17 00:00:00 2001 From: Tiago Farto Date: Fri, 24 Jul 2026 16:29:09 +0000 Subject: [PATCH 02/13] feat(observability): add aggregate enrichment-backlog gauge (ENG-1670) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Emit hub_enrichment_pending_records{enrichment}, an async gauge of eligible-but-unenriched feedback records per enrichment type, summed across all tenants. Unlike the transient River queue-depth gauge this is a data-derived completeness signal that persists across queue drains. A background poller in the API process refreshes it on a conservative interval with a per-iteration query timeout (the aggregate is a full-table scan), gated on metrics being enabled and reporting only deployment-configured enrichments. The per-tenant enable gates still apply in SQL, so a disabled tenant never inflates the backlog. The gauge is labeled by enrichment type only — never tenant_id — to keep cardinality bounded; metrics are OTLP-push, so it never reaches the public API. --- cmd/api/app.go | 75 +++++++++++++++ internal/observability/aggregate.go | 24 +++-- internal/observability/enrichment_backlog.go | 74 +++++++++++++++ .../observability/enrichment_backlog_test.go | 71 ++++++++++++++ internal/observability/names.go | 5 + .../enrichment_status_repository.go | 42 +++++++++ tests/enrichment_status_test.go | 94 ++++++++++++++++--- 7 files changed, 365 insertions(+), 20 deletions(-) create mode 100644 internal/observability/enrichment_backlog.go create mode 100644 internal/observability/enrichment_backlog_test.go diff --git a/cmd/api/app.go b/cmd/api/app.go index 749b8592..242e1b79 100644 --- a/cmd/api/app.go +++ b/cmd/api/app.go @@ -52,6 +52,12 @@ var ( const ( riverQueueDepthInterval = 15 * time.Second startupCleanupTimeout = 5 * time.Second + // enrichmentBacklogInterval is deliberately slower than the River depth poll: the backlog query + // is a full-table aggregate over feedback_records, not a queue-scoped scan. + enrichmentBacklogInterval = 60 * time.Second + // enrichmentBacklogQueryTimeout bounds each aggregate scan so a slow query cannot pin a pool + // connection or stall the ticker. + enrichmentBacklogQueryTimeout = 30 * time.Second ) // embeddingProviderAndModel returns (provider, model) when embeddings are enabled: both EMBEDDING_PROVIDER @@ -706,6 +712,15 @@ func (a *App) Run(ctx context.Context) error { go runRiverQueueDepthPoller(ctx, a.db, a.metrics.Events) } + if a.metrics != nil && a.metrics.EnrichmentBacklog != nil { + go runEnrichmentBacklogPoller(ctx, a.db, a.metrics.EnrichmentBacklog, enrichmentBacklogPollConfig{ + defaultLang: 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 // (no runs exist otherwise, so the sweep would be pointless). if a.taxonomyRepo != nil && (a.cfg.Taxonomy.ServiceURL != "" || a.cfg.Taxonomy.ServiceToken != "") { @@ -744,6 +759,66 @@ var riverDepthQueues = []string{ service.EmotionsQueueName, } +// enrichmentBacklogPollConfig configures runEnrichmentBacklogPoller: the deployment default target +// language and which enrichments are deployment-configured (only those emit a gauge). +type enrichmentBacklogPollConfig struct { + defaultLang string + translationConfigured bool + sentimentConfigured bool + emotionsConfigured bool +} + +// runEnrichmentBacklogPoller periodically refreshes the aggregate enrichment-backlog gauge +// (eligible-but-unenriched records per enrichment, summed across all tenants) — a durable +// completeness signal complementing the transient River queue-depth gauge. Only +// deployment-configured enrichments are reported, and each scan is bounded by its own timeout. +func runEnrichmentBacklogPoller( + ctx context.Context, + db *pgxpool.Pool, + backlog observability.EnrichmentBacklogMetrics, + cfg enrichmentBacklogPollConfig, +) { + repo := repository.NewEnrichmentStatusRepository(db) + + ticker := time.NewTicker(enrichmentBacklogInterval) + defer ticker.Stop() + + update := func() { + queryCtx, cancel := context.WithTimeout(ctx, enrichmentBacklogQueryTimeout) + defer cancel() + + counts, err := repo.CountEnrichmentBacklogAggregate(queryCtx, cfg.defaultLang) + if err != nil { + slog.WarnContext(ctx, "enrichment backlog poll failed", "error", err) + + return + } + + if cfg.translationConfigured { + backlog.SetEnrichmentPending("translation", counts.TranslationEligible-counts.TranslationDone) + } + + if cfg.sentimentConfigured { + backlog.SetEnrichmentPending("sentiment", counts.SentimentEligible-counts.SentimentDone) + } + + if cfg.emotionsConfigured { + backlog.SetEnrichmentPending("emotions", counts.EmotionsEligible-counts.EmotionsDone) + } + } + + update() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + update() + } + } +} + // runRiverQueueDepthPoller periodically updates the per-queue River backlog gauge. Covering // every declared queue (not just default) means a provider outage or a backfill piling tens of // thousands of jobs into an enrichment queue is visible in metrics before users notice the lag. diff --git a/internal/observability/aggregate.go b/internal/observability/aggregate.go index 1ffc4459..0194bd0b 100644 --- a/internal/observability/aggregate.go +++ b/internal/observability/aggregate.go @@ -20,6 +20,8 @@ type Metrics struct { Cache CacheMetrics // EnrichmentClear counts enrichment outputs nulled by an edit's eager-clear. EnrichmentClear EnrichmentClearMetrics + // EnrichmentBacklog gauges the aggregate eligible-but-unenriched record count per enrichment. + EnrichmentBacklog EnrichmentBacklogMetrics } // NewMetrics creates EventMetrics, WebhookMetrics, EmbeddingMetrics, TranslationMetrics, and CacheMetrics from the given meter. @@ -70,14 +72,20 @@ func NewMetrics(meter metric.Meter) (*Metrics, error) { return nil, fmt.Errorf("enrichment clear metrics: %w", err) } + enrichmentBacklog, err := NewEnrichmentBacklogMetrics(meter) + if err != nil { + return nil, fmt.Errorf("enrichment backlog metrics: %w", err) + } + return &Metrics{ - Events: events, - Webhooks: webhooks, - Embeddings: embeddings, - Translation: translation, - Sentiment: sentiment, - Emotions: emotions, - Cache: cache, - EnrichmentClear: enrichmentClear, + Events: events, + Webhooks: webhooks, + Embeddings: embeddings, + Translation: translation, + Sentiment: sentiment, + Emotions: emotions, + Cache: cache, + EnrichmentClear: enrichmentClear, + EnrichmentBacklog: enrichmentBacklog, }, nil } diff --git a/internal/observability/enrichment_backlog.go b/internal/observability/enrichment_backlog.go new file mode 100644 index 00000000..6c6f9234 --- /dev/null +++ b/internal/observability/enrichment_backlog.go @@ -0,0 +1,74 @@ +package observability + +import ( + "context" + "fmt" + "sync" + + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" +) + +// EnrichmentBacklogMetrics reports the aggregate (cross-tenant) count of eligible-but-unenriched +// feedback records per enrichment type — a data-derived "how far behind is enrichment" gauge that +// complements the transient River queue-depth gauge. A background poller refreshes the values; an +// async gauge observes them. Labeled by enrichment type only (a fixed, bounded set); tenant_id is +// deliberately NOT a label, so per-tenant detail stays in the API and metric cardinality is bounded. +type EnrichmentBacklogMetrics interface { + SetEnrichmentPending(enrichment string, count int64) +} + +// enrichmentBacklogMetrics implements EnrichmentBacklogMetrics. The latest per-enrichment value is +// stored under mu and read by the gauge callback. +type enrichmentBacklogMetrics struct { + mu sync.Mutex + pending map[string]int64 + gauge metric.Int64ObservableGauge +} + +// NewEnrichmentBacklogMetrics registers the pending-records gauge. Returns (nil, nil) when meter is +// nil (metrics disabled); the caller translates that into a nil interface. +func NewEnrichmentBacklogMetrics(meter metric.Meter) (EnrichmentBacklogMetrics, error) { + if meter == nil { + //nolint:nilnil // intentional: callers use "if metrics != nil" when metrics disabled + return nil, nil + } + + m := &enrichmentBacklogMetrics{pending: make(map[string]int64)} + + gauge, err := meter.Int64ObservableGauge( + MetricNameEnrichmentPendingRecords, + metric.WithDescription( + "Eligible-but-unenriched feedback records per enrichment type (translation, sentiment, "+ + "emotions), aggregated across all tenants. A data-derived backlog/completeness signal; "+ + "unlike the River queue depth it persists across queue drains.", + ), + metric.WithUnit("1"), + metric.WithInt64Callback(func(_ context.Context, observer metric.Int64Observer) error { + m.mu.Lock() + defer m.mu.Unlock() + + for enrichment, count := range m.pending { + observer.Observe(count, metric.WithAttributes(attribute.String(AttrEnrichment, enrichment))) + } + + return nil + }), + ) + if err != nil { + return nil, fmt.Errorf("create %s: %w", MetricNameEnrichmentPendingRecords, err) + } + + m.gauge = gauge + + return m, nil +} + +// SetEnrichmentPending stores the latest backlog count for an enrichment type; the registered gauge +// callback reports it on the next collection. +func (m *enrichmentBacklogMetrics) SetEnrichmentPending(enrichment string, count int64) { + m.mu.Lock() + defer m.mu.Unlock() + + m.pending[enrichment] = count +} diff --git a/internal/observability/enrichment_backlog_test.go b/internal/observability/enrichment_backlog_test.go new file mode 100644 index 00000000..eb35a011 --- /dev/null +++ b/internal/observability/enrichment_backlog_test.go @@ -0,0 +1,71 @@ +package observability + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/attribute" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" +) + +func TestNewEnrichmentBacklogMetricsNilMeterDisabled(t *testing.T) { + metrics, err := NewEnrichmentBacklogMetrics(nil) + require.NoError(t, err) + assert.Nil(t, metrics, "a nil meter disables metrics") +} + +// TestEnrichmentBacklogMetricsGauge verifies the async gauge reports the latest per-enrichment +// value under the enrichment label, and that a later Set overwrites (a gauge, not a counter). +func TestEnrichmentBacklogMetricsGauge(t *testing.T) { + reader := sdkmetric.NewManualReader() + provider := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)) + + metrics, err := NewEnrichmentBacklogMetrics(provider.Meter("test")) + require.NoError(t, err) + require.NotNil(t, metrics) + + metrics.SetEnrichmentPending("translation", 12) + metrics.SetEnrichmentPending("sentiment", 5) + metrics.SetEnrichmentPending("emotions", 0) + + assert.Equal(t, int64(12), backlogGaugeValue(t, reader, "translation")) + assert.Equal(t, int64(5), backlogGaugeValue(t, reader, "sentiment")) + assert.Equal(t, int64(0), backlogGaugeValue(t, reader, "emotions")) + + // A gauge reports the latest value, not a running sum. + metrics.SetEnrichmentPending("translation", 3) + assert.Equal(t, int64(3), backlogGaugeValue(t, reader, "translation")) +} + +// backlogGaugeValue collects metrics and returns the pending-records gauge value for one enrichment. +func backlogGaugeValue(t *testing.T, reader sdkmetric.Reader, enrichment string) int64 { + t.Helper() + + var collected metricdata.ResourceMetrics + require.NoError(t, reader.Collect(context.Background(), &collected)) + + for _, scope := range collected.ScopeMetrics { + for _, m := range scope.Metrics { + if m.Name != MetricNameEnrichmentPendingRecords { + continue + } + + gauge, ok := m.Data.(metricdata.Gauge[int64]) + require.True(t, ok, "expected Gauge[int64] for %s", MetricNameEnrichmentPendingRecords) + + for _, point := range gauge.DataPoints { + if value, present := point.Attributes.Value(attribute.Key(AttrEnrichment)); present && + value.AsString() == enrichment { + return point.Value + } + } + } + } + + t.Fatalf("gauge %q has no data point for enrichment %q", MetricNameEnrichmentPendingRecords, enrichment) + + return 0 +} diff --git a/internal/observability/names.go b/internal/observability/names.go index e21b6454..ecc87108 100644 --- a/internal/observability/names.go +++ b/internal/observability/names.go @@ -15,6 +15,7 @@ const ( MetricNameProviderPanics = "hub_provider_panics_total" MetricNameHNSWIterativeScanDegraded = "hub_hnsw_iterative_scan_degraded" MetricNameEnrichmentOutputsCleared = "hub_enrichment_outputs_cleared_total" + MetricNameEnrichmentPendingRecords = "hub_enrichment_pending_records" MetricNameWebhookJobsEnqueued = "hub_webhook_jobs_enqueued_total" MetricNameWebhookProviderErrors = "hub_webhook_provider_errors_total" MetricNameWebhookDeliveries = "hub_webhook_deliveries_total" @@ -65,6 +66,10 @@ const ( // AttrQueue labels the River queue-depth gauge; values come from the poller's fixed queue // set, so cardinality is bounded. AttrQueue = "queue" + // AttrEnrichment labels the enrichment-backlog gauge; values are the fixed enrichment types + // (translation, sentiment, emotions). tenant_id is deliberately NOT a label — the gauge is + // aggregated across all tenants to keep cardinality bounded. + AttrEnrichment = "enrichment" ) // AllowedEventTypes returns event type strings allowed for metric attributes (bounded cardinality). diff --git a/internal/repository/enrichment_status_repository.go b/internal/repository/enrichment_status_repository.go index c1e6dd8a..24afcb23 100644 --- a/internal/repository/enrichment_status_repository.go +++ b/internal/repository/enrichment_status_repository.go @@ -89,3 +89,45 @@ func (r *EnrichmentStatusRepository) CountEnrichmentStatus( return counts, nil } + +// enrichmentEffectiveTargetAgg mirrors enrichmentEffectiveTarget but binds the default target to +// $1 — the aggregate query's only parameter (no tenant filter). +const enrichmentEffectiveTargetAgg = `COALESCE(NULLIF(ts.settings->>'target_language', ''), $1)` + +// countEnrichmentBacklogAggregateSQL is countEnrichmentStatusSQL without the tenant filter: it sums +// eligible/done per enrichment across ALL tenants (for the observability gauge). $1 = deployment +// default target language. The per-tenant enable gates still apply so a tenant that switched an +// enrichment off, or has no resolvable target, never inflates the backlog. +const countEnrichmentBacklogAggregateSQL = ` + SELECT + COUNT(*) FILTER (WHERE ` + enrichmentEligibleText + ` AND ` + enrichmentSentimentOn + `), + COUNT(*) FILTER (WHERE ` + enrichmentEligibleText + ` AND ` + enrichmentSentimentOn + ` AND fr.sentiment IS NOT NULL), + COUNT(*) FILTER (WHERE ` + enrichmentEligibleText + ` AND ` + enrichmentEmotionsOn + `), + COUNT(*) FILTER (WHERE ` + enrichmentEligibleText + ` AND ` + enrichmentEmotionsOn + ` AND fr.emotions IS NOT NULL), + COUNT(*) FILTER (WHERE ` + enrichmentEligibleText + ` AND ` + enrichmentEffectiveTargetAgg + ` <> ''), + COUNT(*) FILTER ( + WHERE ` + enrichmentEligibleText + ` + AND ` + enrichmentEffectiveTargetAgg + ` <> '' + AND fr.translation_lang_key = ` + enrichmentEffectiveTargetAgg + `) + FROM feedback_records fr + LEFT JOIN tenant_settings ts ON ts.tenant_id = fr.tenant_id` + +// CountEnrichmentBacklogAggregate returns eligible/done counts per enrichment summed across all +// tenants. defaultLang is the deployment translation fallback. Used by the observability poller; +// the result carries no tenant dimension. +func (r *EnrichmentStatusRepository) CountEnrichmentBacklogAggregate( + ctx context.Context, defaultLang string, +) (EnrichmentStatusCounts, error) { + var counts EnrichmentStatusCounts + + err := r.db.QueryRow(ctx, countEnrichmentBacklogAggregateSQL, defaultLang).Scan( + &counts.SentimentEligible, &counts.SentimentDone, + &counts.EmotionsEligible, &counts.EmotionsDone, + &counts.TranslationEligible, &counts.TranslationDone, + ) + if err != nil { + return EnrichmentStatusCounts{}, fmt.Errorf("count enrichment backlog aggregate: %w", err) + } + + return counts, nil +} diff --git a/tests/enrichment_status_test.go b/tests/enrichment_status_test.go index 6abeaf3f..cac1be44 100644 --- a/tests/enrichment_status_test.go +++ b/tests/enrichment_status_test.go @@ -36,18 +36,7 @@ func TestCountEnrichmentStatus(t *testing.T) { // mkRecord inserts one feedback record for a tenant with an explicit field type + value_text. mkRecord := func(tenant string, fieldType models.FieldType, valueText string) *models.FeedbackRecord { - vt := valueText - rec, createErr := frepo.Create(ctx, &models.CreateFeedbackRecordRequest{ - SourceType: "formbricks", - FieldID: "q1", - FieldType: fieldType, - ValueText: &vt, - TenantID: tenant, - SubmissionID: testTenantID("sub"), - }) - require.NoError(t, createErr) - - return rec + return seedEnrichmentRecord(t, frepo, tenant, fieldType, valueText) } setSentiment := func(id uuid.UUID) { @@ -151,3 +140,84 @@ func TestCountEnrichmentStatus(t *testing.T) { assert.Equal(t, int64(1), counts.SentimentEligible, "tenant A sees only its own record, not tenant B's") }) } + +// seedEnrichmentRecord inserts one feedback record and returns it. Shared by the per-tenant and +// aggregate enrichment-status tests. +func seedEnrichmentRecord( + t *testing.T, frepo *repository.FeedbackRecordsRepository, tenant string, fieldType models.FieldType, valueText string, +) *models.FeedbackRecord { + t.Helper() + + vt := valueText + rec, err := frepo.Create(context.Background(), &models.CreateFeedbackRecordRequest{ + SourceType: "formbricks", + FieldID: "q1", + FieldType: fieldType, + ValueText: &vt, + TenantID: tenant, + SubmissionID: testTenantID("sub"), + }) + require.NoError(t, err) + + return rec +} + +// TestCountEnrichmentBacklogAggregate covers the cross-tenant aggregate query that feeds the +// backlog gauge. The shared test DB holds records from other tests, so it asserts on the DELTA +// around a fresh seed rather than absolute totals — and verifies the per-tenant enable gate still +// applies (a sentiment-off tenant contributes to emotions but not sentiment). +func TestCountEnrichmentBacklogAggregate(t *testing.T) { + ctx := context.Background() + + cfg, err := config.Load() + require.NoError(t, err) + + db, err := database.NewPostgresPool(ctx, cfg.Database.URL, database.WithPoolConfig(cfg.Database.PoolConfig())) + require.NoError(t, err) + + defer db.Close() + + statusRepo := repository.NewEnrichmentStatusRepository(db) + frepo := repository.NewFeedbackRecordsRepository(db) + tsRepo := repository.NewTenantSettingsRepository(db) + + const defaultLang = "en-US" + + pending := func(c repository.EnrichmentStatusCounts) (sentiment, emotions, translation int64) { + return c.SentimentEligible - c.SentimentDone, + c.EmotionsEligible - c.EmotionsDone, + c.TranslationEligible - c.TranslationDone + } + + before, err := statusRepo.CountEnrichmentBacklogAggregate(ctx, defaultLang) + require.NoError(t, err) + + beforeSent, beforeEmo, beforeTrans := pending(before) + + // Tenant with everything enabled (no settings row → sentiment/emotions default-on; translation + // via the en-US default): 3 un-enriched text records. + enabled := testTenantID("agg-enabled") + for range 3 { + seedEnrichmentRecord(t, frepo, enabled, models.FieldTypeText, "pending record") + } + + // Tenant with sentiment switched OFF: 2 un-enriched text records. These must add to emotions and + // translation backlog but NOT sentiment. + sentimentOff := testTenantID("agg-sentiment-off") + off := false + _, err = tsRepo.Upsert(ctx, sentimentOff, models.EnrichmentSettings{SentimentEnabled: &off}) + require.NoError(t, err) + + for range 2 { + seedEnrichmentRecord(t, frepo, sentimentOff, models.FieldTypeText, "pending record") + } + + after, err := statusRepo.CountEnrichmentBacklogAggregate(ctx, defaultLang) + require.NoError(t, err) + + afterSent, afterEmo, afterTrans := pending(after) + + assert.Equal(t, int64(3), afterSent-beforeSent, "only the sentiment-enabled tenant's 3 records add to sentiment backlog") + assert.Equal(t, int64(5), afterEmo-beforeEmo, "emotions default-enabled for both tenants → 3+2") + assert.Equal(t, int64(5), afterTrans-beforeTrans, "en-US default makes all 5 records translation-pending") +} From 92fdeb02329ecd5c3deec6c5126270e9a501849c Mon Sep 17 00:00:00 2001 From: Tiago Farto Date: Fri, 24 Jul 2026 16:30:50 +0000 Subject: [PATCH 03/13] docs(openapi): add /v1/enrichment-status to the API contract (ENG-1670) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Register the endpoint (tag "Enrichment Status", operationId get-enrichment-status → enrichmentStatus.retrieve) plus the EnrichmentStatusOutputBody / EnrichmentTypeStatus schemas, so Stainless regenerates the SDK and API reference. Validated with spectral (make lint-openapi). --- openapi.yaml | 90 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 90 insertions(+) diff --git a/openapi.yaml b/openapi.yaml index ffc87700..ed953636 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -54,6 +54,8 @@ tags: description: Tenant-scoped enrichment settings - name: Taxonomy description: Automatic topic/subtopic taxonomy generation, run history, tree browsing, and node edits + - name: Enrichment Status + description: Tenant-scoped enrichment progress (translation, sentiment, emotions) security: - ApiKeyAuth: [] paths: @@ -1380,6 +1382,56 @@ paths: application/problem+json: schema: $ref: '#/components/schemas/ErrorModel' + /v1/enrichment-status: + get: + tags: + - Enrichment Status + summary: Get tenant enrichment status + description: | + Returns a tenant's enrichment progress across the record-level enrichments + (translation, sentiment, emotions). For each, `enabled` reports whether the enrichment + is active for the tenant (deployment-configured and switched on / with a resolvable + target language), and `eligible`/`done` are directory-level counts of feedback records + that qualify and that have been enriched — the UI derives "in progress" as + `eligible - done`. When an enrichment is not enabled its counts are zero. The response + contains counts only (no record identifiers or content). + operationId: get-enrichment-status + parameters: + - name: tenant_id + in: query + required: true + description: Tenant whose enrichment status should be returned. + schema: + type: string + minLength: 1 + maxLength: 255 + pattern: '^[^\x00]*$' + example: "org-123" + responses: + "200": + description: Tenant enrichment status + content: + application/json: + schema: + $ref: '#/components/schemas/EnrichmentStatusOutputBody' + "400": + description: Bad Request (e.g. missing or invalid tenant_id) + content: + application/problem+json: + schema: + $ref: '#/components/schemas/ErrorModel' + "401": + description: Unauthorized (missing or invalid API key) + content: + application/problem+json: + schema: + $ref: '#/components/schemas/ErrorModel' + default: + description: Error + content: + application/problem+json: + schema: + $ref: '#/components/schemas/ErrorModel' /v1/taxonomy/fields: get: tags: @@ -3362,6 +3414,44 @@ components: $ref: '#/components/schemas/TaxonomyFieldOption' required: - data + EnrichmentTypeStatus: + type: object + additionalProperties: false + description: One enrichment's progress for a tenant. When `enabled` is false, `eligible` and `done` are zero. + properties: + enabled: + type: boolean + description: Whether the enrichment is active for the tenant (deployment-configured and switched on / with a resolvable target language). + eligible: + type: integer + format: int64 + description: Feedback records that qualify for this enrichment. + done: + type: integer + format: int64 + description: Eligible records that have been enriched. + required: + - enabled + - eligible + - done + EnrichmentStatusOutputBody: + type: object + additionalProperties: false + description: A tenant's enrichment progress across the record-level enrichments. Counts are directory-level totals. + properties: + tenant_id: + type: string + translation: + $ref: '#/components/schemas/EnrichmentTypeStatus' + sentiment: + $ref: '#/components/schemas/EnrichmentTypeStatus' + emotions: + $ref: '#/components/schemas/EnrichmentTypeStatus' + required: + - tenant_id + - translation + - sentiment + - emotions CreateTaxonomyRunInputBody: type: object additionalProperties: false From a907e4c6736e26b92f669816f412e64291854128 Mon Sep 17 00:00:00 2001 From: Tiago Farto Date: Fri, 24 Jul 2026 16:47:44 +0000 Subject: [PATCH 04/13] refactor(enrichment): address pre-PR review nits (ENG-1670) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - DRY the count SQL: the per-tenant and aggregate queries now share a single enrichmentCountSelect / enrichmentCountFrom so the six FILTER predicates can't drift; the effective-target placeholder is unified to $1 in both. - Document that the per-tenant scan is served by idx_feedback_records_tenant_field_type (tenant_id, field_type) — the existing composite index already covers it, so no new index is needed. - Name the metric-label values (observability.EnrichmentType{Translation,Sentiment, Emotions}) instead of string literals in the poller. --- cmd/api/app.go | 6 +- internal/observability/enrichment_backlog.go | 8 ++ .../enrichment_status_repository.go | 75 +++++++++---------- 3 files changed, 46 insertions(+), 43 deletions(-) diff --git a/cmd/api/app.go b/cmd/api/app.go index 242e1b79..55986e21 100644 --- a/cmd/api/app.go +++ b/cmd/api/app.go @@ -795,15 +795,15 @@ func runEnrichmentBacklogPoller( } if cfg.translationConfigured { - backlog.SetEnrichmentPending("translation", counts.TranslationEligible-counts.TranslationDone) + backlog.SetEnrichmentPending(observability.EnrichmentTypeTranslation, counts.TranslationEligible-counts.TranslationDone) } if cfg.sentimentConfigured { - backlog.SetEnrichmentPending("sentiment", counts.SentimentEligible-counts.SentimentDone) + backlog.SetEnrichmentPending(observability.EnrichmentTypeSentiment, counts.SentimentEligible-counts.SentimentDone) } if cfg.emotionsConfigured { - backlog.SetEnrichmentPending("emotions", counts.EmotionsEligible-counts.EmotionsDone) + backlog.SetEnrichmentPending(observability.EnrichmentTypeEmotions, counts.EmotionsEligible-counts.EmotionsDone) } } diff --git a/internal/observability/enrichment_backlog.go b/internal/observability/enrichment_backlog.go index 6c6f9234..928f69db 100644 --- a/internal/observability/enrichment_backlog.go +++ b/internal/observability/enrichment_backlog.go @@ -9,6 +9,14 @@ import ( "go.opentelemetry.io/otel/metric" ) +// Enrichment type label values for MetricNameEnrichmentPendingRecords — a fixed, bounded set that +// keeps the gauge's cardinality low. Callers pass these to SetEnrichmentPending. +const ( + EnrichmentTypeTranslation = "translation" + EnrichmentTypeSentiment = "sentiment" + EnrichmentTypeEmotions = "emotions" +) + // EnrichmentBacklogMetrics reports the aggregate (cross-tenant) count of eligible-but-unenriched // feedback records per enrichment type — a data-derived "how far behind is enrichment" gauge that // complements the transient River queue-depth gauge. A background poller refreshes the values; an diff --git a/internal/repository/enrichment_status_repository.go b/internal/repository/enrichment_status_repository.go index 24afcb23..f9620d7c 100644 --- a/internal/repository/enrichment_status_repository.go +++ b/internal/repository/enrichment_status_repository.go @@ -19,10 +19,10 @@ func NewEnrichmentStatusRepository(db *pgxpool.Pool) *EnrichmentStatusRepository return &EnrichmentStatusRepository{db: db} } -// EnrichmentStatusCounts holds the raw per-enrichment eligible/done counts for a tenant. -// The counts already reflect the per-tenant gates: sentiment/emotions counts include only -// tenants with the enrichment switched on, translation counts only records with a resolvable -// effective target language. The deployment-level (provider/model) gate is applied by the caller. +// EnrichmentStatusCounts holds the raw per-enrichment eligible/done counts. The counts already +// reflect the per-tenant gates: sentiment/emotions include only tenants with the enrichment +// switched on, translation only records with a resolvable effective target language. The +// deployment-level (provider/model) gate is applied by the caller. type EnrichmentStatusCounts struct { TranslationEligible int64 TranslationDone int64 @@ -41,22 +41,23 @@ type EnrichmentStatusCounts struct { const enrichmentEligibleText = `fr.field_type = 'text' AND fr.value_text IS NOT NULL AND btrim(fr.value_text, E' \t\r\n') <> ''` // enrichmentEffectiveTarget resolves a tenant's effective translation target: its own -// target_language, falling back to the deployment default ($N). An empty result means translation +// target_language, falling back to the deployment default ($1). An empty result means translation // is not enabled for the tenant. Mirrors translationBackfillSelectSQL. -// sentimentEnabled / emotionsEnabled read the tri-state per-directory switch, defaulting to -// enabled when the key is absent — matching EnrichmentSettings.SentimentEnrichmentEnabled / -// EmotionsEnrichmentEnabled (parity is covered by test). +// enrichmentSentimentOn / enrichmentEmotionsOn read the tri-state per-directory switch, defaulting +// to enabled when the key is absent — matching EnrichmentSettings.SentimentEnrichmentEnabled / +// EmotionsEnrichmentEnabled (parity is covered by test). $1 is the deployment default target in +// both the per-tenant and aggregate queries. const ( - enrichmentEffectiveTarget = `COALESCE(NULLIF(ts.settings->>'target_language', ''), $2)` + enrichmentEffectiveTarget = `COALESCE(NULLIF(ts.settings->>'target_language', ''), $1)` enrichmentSentimentOn = `COALESCE((ts.settings->>'sentiment_enabled')::boolean, true)` enrichmentEmotionsOn = `COALESCE((ts.settings->>'emotions_enabled')::boolean, true)` ) -// countEnrichmentStatusSQL counts eligible/done per enrichment for one tenant in a single pass. -// $1 = tenant_id, $2 = deployment default target language. All predicate fragments are static -// constants (never user input); tenant_id is bound. -const countEnrichmentStatusSQL = ` - SELECT +// enrichmentCountSelect is the shared six-column SELECT list — {sentiment, emotions, translation} × +// {eligible, done} — used by both the per-tenant and aggregate queries so the predicates can't +// drift between them. Column order must match the EnrichmentStatusCounts scan order below. All +// fragments are static constants (never user input). +const enrichmentCountSelect = ` COUNT(*) FILTER (WHERE ` + enrichmentEligibleText + ` AND ` + enrichmentSentimentOn + `), COUNT(*) FILTER (WHERE ` + enrichmentEligibleText + ` AND ` + enrichmentSentimentOn + ` AND fr.sentiment IS NOT NULL), COUNT(*) FILTER (WHERE ` + enrichmentEligibleText + ` AND ` + enrichmentEmotionsOn + `), @@ -65,10 +66,24 @@ const countEnrichmentStatusSQL = ` COUNT(*) FILTER ( WHERE ` + enrichmentEligibleText + ` AND ` + enrichmentEffectiveTarget + ` <> '' - AND fr.translation_lang_key = ` + enrichmentEffectiveTarget + `) + AND fr.translation_lang_key = ` + enrichmentEffectiveTarget + `)` + +const enrichmentCountFrom = ` FROM feedback_records fr - LEFT JOIN tenant_settings ts ON ts.tenant_id = fr.tenant_id - WHERE fr.tenant_id = $1` + LEFT JOIN tenant_settings ts ON ts.tenant_id = fr.tenant_id` + +// countEnrichmentStatusSQL counts eligible/done per enrichment for ONE tenant. $1 = deployment +// default target language, $2 = tenant_id. The tenant-scoped scan is served by +// idx_feedback_records_tenant_field_type (tenant_id, field_type): only the tenant's text rows are +// visited, so cost scales with the tenant's text-record count, not the whole table. +const countEnrichmentStatusSQL = `SELECT ` + enrichmentCountSelect + enrichmentCountFrom + ` + WHERE fr.tenant_id = $2` + +// countEnrichmentBacklogAggregateSQL is the same SELECT without the tenant filter: it sums +// eligible/done per enrichment across ALL tenants (for the observability gauge). $1 = deployment +// default target language. The per-tenant enable gates still apply, so a tenant that switched an +// enrichment off, or has no resolvable target, never inflates the backlog. +const countEnrichmentBacklogAggregateSQL = `SELECT ` + enrichmentCountSelect + enrichmentCountFrom // CountEnrichmentStatus returns one tenant's eligible/done counts per enrichment. defaultLang is // the deployment translation fallback ("" disables the fallback, so only tenants with their own @@ -78,7 +93,8 @@ func (r *EnrichmentStatusRepository) CountEnrichmentStatus( ) (EnrichmentStatusCounts, error) { var counts EnrichmentStatusCounts - err := r.db.QueryRow(ctx, countEnrichmentStatusSQL, tenantID, defaultLang).Scan( + // Scan order matches enrichmentCountSelect. + err := r.db.QueryRow(ctx, countEnrichmentStatusSQL, defaultLang, tenantID).Scan( &counts.SentimentEligible, &counts.SentimentDone, &counts.EmotionsEligible, &counts.EmotionsDone, &counts.TranslationEligible, &counts.TranslationDone, @@ -90,28 +106,6 @@ func (r *EnrichmentStatusRepository) CountEnrichmentStatus( return counts, nil } -// enrichmentEffectiveTargetAgg mirrors enrichmentEffectiveTarget but binds the default target to -// $1 — the aggregate query's only parameter (no tenant filter). -const enrichmentEffectiveTargetAgg = `COALESCE(NULLIF(ts.settings->>'target_language', ''), $1)` - -// countEnrichmentBacklogAggregateSQL is countEnrichmentStatusSQL without the tenant filter: it sums -// eligible/done per enrichment across ALL tenants (for the observability gauge). $1 = deployment -// default target language. The per-tenant enable gates still apply so a tenant that switched an -// enrichment off, or has no resolvable target, never inflates the backlog. -const countEnrichmentBacklogAggregateSQL = ` - SELECT - COUNT(*) FILTER (WHERE ` + enrichmentEligibleText + ` AND ` + enrichmentSentimentOn + `), - COUNT(*) FILTER (WHERE ` + enrichmentEligibleText + ` AND ` + enrichmentSentimentOn + ` AND fr.sentiment IS NOT NULL), - COUNT(*) FILTER (WHERE ` + enrichmentEligibleText + ` AND ` + enrichmentEmotionsOn + `), - COUNT(*) FILTER (WHERE ` + enrichmentEligibleText + ` AND ` + enrichmentEmotionsOn + ` AND fr.emotions IS NOT NULL), - COUNT(*) FILTER (WHERE ` + enrichmentEligibleText + ` AND ` + enrichmentEffectiveTargetAgg + ` <> ''), - COUNT(*) FILTER ( - WHERE ` + enrichmentEligibleText + ` - AND ` + enrichmentEffectiveTargetAgg + ` <> '' - AND fr.translation_lang_key = ` + enrichmentEffectiveTargetAgg + `) - FROM feedback_records fr - LEFT JOIN tenant_settings ts ON ts.tenant_id = fr.tenant_id` - // CountEnrichmentBacklogAggregate returns eligible/done counts per enrichment summed across all // tenants. defaultLang is the deployment translation fallback. Used by the observability poller; // the result carries no tenant dimension. @@ -120,6 +114,7 @@ func (r *EnrichmentStatusRepository) CountEnrichmentBacklogAggregate( ) (EnrichmentStatusCounts, error) { var counts EnrichmentStatusCounts + // Scan order matches enrichmentCountSelect. err := r.db.QueryRow(ctx, countEnrichmentBacklogAggregateSQL, defaultLang).Scan( &counts.SentimentEligible, &counts.SentimentDone, &counts.EmotionsEligible, &counts.EmotionsDone, From 978862267c86ba4f2bc6d64609cdf8da227ba1f3 Mon Sep 17 00:00:00 2001 From: Tiago Farto Date: Fri, 24 Jul 2026 17:01:21 +0000 Subject: [PATCH 05/13] perf(enrichment): run the backlog poller every 5m instead of 60s (ENG-1670) The aggregate backlog gauge is a slow-moving trend signal, so a full-table scan every 60s over the high-write feedback_records table is more often than needed. Widening to 5m cuts the shared-DB load 5x and keeps the MVCC snapshot the scan holds short-lived relative to VACUUM. The poller still only runs when metrics are enabled, and each scan is bounded by a 30s timeout. --- cmd/api/app.go | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/cmd/api/app.go b/cmd/api/app.go index 55986e21..97108f94 100644 --- a/cmd/api/app.go +++ b/cmd/api/app.go @@ -52,11 +52,14 @@ var ( const ( riverQueueDepthInterval = 15 * time.Second startupCleanupTimeout = 5 * time.Second - // enrichmentBacklogInterval is deliberately slower than the River depth poll: the backlog query - // is a full-table aggregate over feedback_records, not a queue-scoped scan. - enrichmentBacklogInterval = 60 * time.Second + // enrichmentBacklogInterval is deliberately much slower than the River depth poll: the backlog + // query is a full-table aggregate over feedback_records (a high-write table), so it runs + // infrequently to minimize shared-DB load and keep the MVCC snapshot it holds short-lived + // relative to VACUUM. Backlog is a slow-moving trend signal, so 5-minute resolution is ample. + // The poller only runs at all when metrics are enabled (see App.Run). + enrichmentBacklogInterval = 5 * time.Minute // enrichmentBacklogQueryTimeout bounds each aggregate scan so a slow query cannot pin a pool - // connection or stall the ticker. + // connection, stall the ticker, or hold a long snapshot that delays VACUUM on feedback_records. enrichmentBacklogQueryTimeout = 30 * time.Second ) From 92151a6c7f6bab5e20227328d242295d93c1fedd Mon Sep 17 00:00:00 2001 From: Tiago Farto Date: Fri, 24 Jul 2026 17:16:30 +0000 Subject: [PATCH 06/13] chore(deps): bump golang.org/x/text to v0.39.0 (GO-2026-5970) govulncheck flags GO-2026-5970 (infinite loop on invalid input in golang.org/x/text, fixed in v0.39.0), failing the Code Quality gate. The vulnerable paths are pre-existing (webhook sender, pgx pool, embedding input) and unrelated to this PR, but the scan runs module-wide. Bump to v0.39.0; govulncheck is clean afterwards. go mod tidy also pulled x/sync v0.20.0 -> v0.21.0. --- go.mod | 4 ++-- go.sum | 12 ++++++------ 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/go.mod b/go.mod index 2e267b88..85a4966d 100644 --- a/go.mod +++ b/go.mod @@ -27,8 +27,8 @@ require ( go.opentelemetry.io/otel/sdk v1.43.0 go.opentelemetry.io/otel/sdk/metric v1.43.0 go.opentelemetry.io/otel/trace v1.43.0 - golang.org/x/sync v0.20.0 - golang.org/x/text v0.37.0 + golang.org/x/sync v0.21.0 + golang.org/x/text v0.39.0 google.golang.org/genai v1.54.0 ) diff --git a/go.sum b/go.sum index f447f872..4a587470 100644 --- a/go.sum +++ b/go.sum @@ -187,14 +187,14 @@ golang.org/x/net v0.55.0 h1:bcvxaJn3e1U6InsFWt1JUq1aSjnRxLzT2rtD2KfkDF8= golang.org/x/net v0.55.0/go.mod h1:L5U2KuzuOe1lY7Z+aWVIKK6qEeJXnXV9yzGA+WCHJww= golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs= golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q= -golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= -golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sync v0.21.0 h1:HLII4xRRTtCRkxYp4HNFF0Js/Og6q2i++KXbg0gHCwM= +golang.org/x/sync v0.21.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY= golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= -golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc= -golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38= -golang.org/x/tools v0.44.0 h1:UP4ajHPIcuMjT1GqzDWRlalUEoY+uzoZKnhOjbIPD2c= -golang.org/x/tools v0.44.0/go.mod h1:KA0AfVErSdxRZIsOVipbv3rQhVXTnlU6UhKxHd1seDI= +golang.org/x/text v0.39.0 h1:UbZz4pLOvn600D6Oh6GGEI6VAmndrEBLv8/6BEXzyus= +golang.org/x/text v0.39.0/go.mod h1:3UwRclnC2g0TU9x8PZiyfOajCd1zaUNHF9cvqcQZ+ZM= +golang.org/x/tools v0.47.0 h1:7Kn5x/d1svx/PzryTsqeoZN4TZwqeH5pGWjefhLi/1Q= +golang.org/x/tools v0.47.0/go.mod h1:dFHnyTvFWY212G+h7ZY4Vsp/K3U4/7W9TyVaAul8uCA= gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= google.golang.org/api v0.276.0 h1:nVArUtfLEihtW+b0DdcqRGK1xoEm2+ltAihyztq7MKY= From c23c28505694563fc0ef0ac4a0ffa47b4f10c67e Mon Sep 17 00:00:00 2001 From: Tiago Farto Date: Fri, 24 Jul 2026 18:08:04 +0000 Subject: [PATCH 07/13] refactor(enrichment): address review comments (ENG-1670) - repository: hoist `field_type = 'text'` into the outer WHERE of both count queries so the (tenant_id, field_type) index narrows the scan to text rows instead of scanning all rows and discarding non-text inside the FILTERs. Redundant with the per-column predicate, so no count changes (EXPLAIN confirms index access; integration tests still green). - api: trim the poller's fallback language for symmetry with NewEnrichmentStatusService so the endpoint and the backlog gauge resolve the same target (config already canonicalizes it; defensive). - openapi: document the 503 the handler returns when the service is unwired, matching the other feature-gated endpoints. --- cmd/api/app.go | 5 ++++- .../enrichment_status_repository.go | 20 ++++++++++++------- openapi.yaml | 8 ++++++++ 3 files changed, 25 insertions(+), 8 deletions(-) diff --git a/cmd/api/app.go b/cmd/api/app.go index 97108f94..efc27ac8 100644 --- a/cmd/api/app.go +++ b/cmd/api/app.go @@ -6,6 +6,7 @@ import ( "fmt" "log/slog" "net/http" + "strings" "time" lru "github.com/hashicorp/golang-lru/v2" @@ -717,7 +718,9 @@ 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{ - defaultLang: a.cfg.Translation.DefaultLanguage, + // 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(), diff --git a/internal/repository/enrichment_status_repository.go b/internal/repository/enrichment_status_repository.go index f9620d7c..e2283e96 100644 --- a/internal/repository/enrichment_status_repository.go +++ b/internal/repository/enrichment_status_repository.go @@ -72,18 +72,24 @@ const enrichmentCountFrom = ` FROM feedback_records fr LEFT JOIN tenant_settings ts ON ts.tenant_id = fr.tenant_id` +// The `fr.field_type = 'text'` predicate in the outer WHERE below is redundant with +// enrichmentEligibleText inside every FILTER (so it can never change a count), but hoisting it out +// lets the planner use idx_feedback_records_tenant_field_type (tenant_id, field_type) to visit only +// text rows instead of scanning every row and discarding non-text inside the FILTERs. + // countEnrichmentStatusSQL counts eligible/done per enrichment for ONE tenant. $1 = deployment -// default target language, $2 = tenant_id. The tenant-scoped scan is served by -// idx_feedback_records_tenant_field_type (tenant_id, field_type): only the tenant's text rows are -// visited, so cost scales with the tenant's text-record count, not the whole table. +// default target language, $2 = tenant_id. The (tenant_id, field_type) predicate is served by +// idx_feedback_records_tenant_field_type, so cost scales with the tenant's text-record count. const countEnrichmentStatusSQL = `SELECT ` + enrichmentCountSelect + enrichmentCountFrom + ` - WHERE fr.tenant_id = $2` + WHERE fr.tenant_id = $2 AND fr.field_type = 'text'` // countEnrichmentBacklogAggregateSQL is the same SELECT without the tenant filter: it sums // eligible/done per enrichment across ALL tenants (for the observability gauge). $1 = deployment -// default target language. The per-tenant enable gates still apply, so a tenant that switched an -// enrichment off, or has no resolvable target, never inflates the backlog. -const countEnrichmentBacklogAggregateSQL = `SELECT ` + enrichmentCountSelect + enrichmentCountFrom +// default target language. The field_type = 'text' predicate narrows the scan to text rows; the +// per-tenant enable gates still apply, so a tenant that switched an enrichment off, or has no +// resolvable target, never inflates the backlog. +const countEnrichmentBacklogAggregateSQL = `SELECT ` + enrichmentCountSelect + enrichmentCountFrom + ` + WHERE fr.field_type = 'text'` // CountEnrichmentStatus returns one tenant's eligible/done counts per enrichment. defaultLang is // the deployment translation fallback ("" disables the fallback, so only tenants with their own diff --git a/openapi.yaml b/openapi.yaml index ed953636..7a2a44f6 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -1426,6 +1426,14 @@ paths: application/problem+json: schema: $ref: '#/components/schemas/ErrorModel' + "503": + description: | + Service Unavailable – the enrichment status service is not available + (code `service_unavailable`). + content: + application/problem+json: + schema: + $ref: '#/components/schemas/ErrorModel' default: description: Error content: From bc4fb989addc3e9b9adf11ec7726b88cc0122320 Mon Sep 17 00:00:00 2001 From: Tiago Farto Date: Sat, 25 Jul 2026 11:41:08 +0000 Subject: [PATCH 08/13] refactor(enrichment): honest content-trim + emotions gate test (ENG-1670) Follow-ups from a final re-review: - Broaden the eligibility trim to the full ASCII whitespace set (add VT/FF -> E' \t\n\v\f\r') and correct the comment, which previously overclaimed an exact match with the workers' Go strings.TrimSpace gate. TrimSpace also strips exotic Unicode whitespace (NBSP, U+3000, ...), so a value composed entirely of those remains a rare eligible-but-never-done edge -- now documented and accepted, matching the existing backfill queries rather than pretending otherwise. - Add a DB-level test that the emotions per-tenant switch gates eligibility (previously only the sentiment switch was covered against Postgres). --- .../repository/enrichment_status_repository.go | 17 ++++++++++------- tests/enrichment_status_test.go | 16 ++++++++++++++++ 2 files changed, 26 insertions(+), 7 deletions(-) diff --git a/internal/repository/enrichment_status_repository.go b/internal/repository/enrichment_status_repository.go index e2283e96..cb30bf17 100644 --- a/internal/repository/enrichment_status_repository.go +++ b/internal/repository/enrichment_status_repository.go @@ -32,13 +32,16 @@ type EnrichmentStatusCounts struct { EmotionsDone int64 } -// enrichmentEligibleText is the data-level eligibility predicate: an open-text field with -// content. It mirrors the backfill eligibility (see classifyBackfillEligibleSQL / -// translationBackfillSelectSQL) but uses the fuller btrim charset E' \t\r\n' so a whitespace-only -// value_text ("\t", "\n") is treated as empty — matching the workers' Go strings.TrimSpace content -// gate (HasOpenText). field_type = 'text' is load-bearing: matrix/multi-choice expansion writes -// value_text on categorical/number rows that are not enrichable. -const enrichmentEligibleText = `fr.field_type = 'text' AND fr.value_text IS NOT NULL AND btrim(fr.value_text, E' \t\r\n') <> ''` +// enrichmentEligibleText is the data-level eligibility predicate: an open-text field with content. +// It mirrors the backfill eligibility (classifyBackfillEligibleSQL / translationBackfillSelectSQL), +// trimming the full ASCII whitespace set (space, tab, VT, FF, CR, LF) before the emptiness check. +// This approximates the workers' HasOpenText gate (Go strings.TrimSpace, which additionally strips +// exotic Unicode whitespace such as NBSP U+00A0 or the ideographic space U+3000); a value composed +// ENTIRELY of exotic Unicode whitespace is a rare edge that would read as eligible-but-never-done +// here — an accepted approximation, consistent with the existing backfill queries. field_type = +// 'text' is load-bearing: matrix/multi-choice expansion writes value_text on categorical/number +// rows that are not enrichable. +const enrichmentEligibleText = `fr.field_type = 'text' AND fr.value_text IS NOT NULL AND btrim(fr.value_text, E' \t\n\v\f\r') <> ''` // enrichmentEffectiveTarget resolves a tenant's effective translation target: its own // target_language, falling back to the deployment default ($1). An empty result means translation diff --git a/tests/enrichment_status_test.go b/tests/enrichment_status_test.go index cac1be44..221b2f24 100644 --- a/tests/enrichment_status_test.go +++ b/tests/enrichment_status_test.go @@ -110,6 +110,22 @@ func TestCountEnrichmentStatus(t *testing.T) { assert.Equal(t, int64(2), counts.EmotionsEligible, "emotions default-enabled → still eligible") }) + t.Run("emotions switch off zeroes emotions eligibility, sentiment unaffected", func(t *testing.T) { + tenant := testTenantID("enrich-status-emotions-off") + off := false + _, err := tsRepo.Upsert(ctx, tenant, models.EnrichmentSettings{EmotionsEnabled: &off}) + require.NoError(t, err) + + mkRecord(tenant, models.FieldTypeText, "a") + mkRecord(tenant, models.FieldTypeText, "b") + + counts, err := statusRepo.CountEnrichmentStatus(ctx, tenant, "") + require.NoError(t, err) + + assert.Equal(t, int64(0), counts.EmotionsEligible, "emotions disabled → not eligible") + assert.Equal(t, int64(2), counts.SentimentEligible, "sentiment default-enabled → still eligible") + }) + t.Run("translation eligibility follows the effective target", func(t *testing.T) { tenant := testTenantID("enrich-status-trans") // No tenant settings row at all → no own target language. From b8d3e8b32f7eac4cdc04f13a1f4b83bb7673fece Mon Sep 17 00:00:00 2001 From: Tiago Farto Date: Mon, 27 Jul 2026 08:40:06 +0000 Subject: [PATCH 09/13] fix(enrichment): single-flight backlog poll, alertable failures, honest plan note (ENG-1670) Review follow-ups: - Multi-replica: production runs several API replicas per region and each was repeating the same cross-tenant scan every tick and publishing the same global gauge (multiplied DB load, duplicate series a sum would over-count). The poll now runs under a non-blocking transaction-scoped advisory lock, so exactly one replica scans per tick; losing the race is a normal skip, not an error. Reuses the pg_try_advisory_xact_lock idiom already used elsewhere in the package. - Alerting: a failing poll left the gauge frozen at its last value, which reads as a healthy steady backlog on a dashboard. Failures now increment hub_enrichment_backlog_poll_errors_total and escalate warn -> error after three consecutive misses. - Correct the aggregate query's plan comment: it canNOT use the (tenant_id, field_type) index (tenant_id is the leading column and the aggregate has no tenant predicate), so it seq-scans by design. Documented why that is the right trade-off rather than adding a partial index that would tax the ingest path. Adds DB-backed coverage for the leader election and the poll-error counter. --- cmd/api/app.go | 30 ++++- internal/observability/enrichment_backlog.go | 31 +++++- .../observability/enrichment_backlog_test.go | 37 +++++++ internal/observability/names.go | 1 + .../enrichment_status_repository.go | 103 +++++++++++++----- tests/enrichment_status_test.go | 49 +++++++++ 6 files changed, 218 insertions(+), 33 deletions(-) diff --git a/cmd/api/app.go b/cmd/api/app.go index efc27ac8..dbaf3bc8 100644 --- a/cmd/api/app.go +++ b/cmd/api/app.go @@ -62,6 +62,9 @@ const ( // enrichmentBacklogQueryTimeout bounds each aggregate scan so a slow query cannot pin a pool // connection, stall the ticker, or hold a long snapshot that delays VACUUM on feedback_records. enrichmentBacklogQueryTimeout = 30 * time.Second + // enrichmentBacklogFailuresBeforeError is how many consecutive failed refreshes escalate the + // log from warn to error (a transient blip is expected; a sustained run means a stale gauge). + enrichmentBacklogFailuresBeforeError = 3 ) // embeddingProviderAndModel returns (provider, model) when embeddings are enabled: both EMBEDDING_PROVIDER @@ -789,14 +792,37 @@ func runEnrichmentBacklogPoller( ticker := time.NewTicker(enrichmentBacklogInterval) defer ticker.Stop() + consecutiveFailures := 0 + update := func() { queryCtx, cancel := context.WithTimeout(ctx, enrichmentBacklogQueryTimeout) defer cancel() - counts, err := repo.CountEnrichmentBacklogAggregate(queryCtx, cfg.defaultLang) + // Only the replica that wins the advisory lock scans; the others skip this tick. + counts, leader, err := repo.CountEnrichmentBacklogAggregateIfLeader(queryCtx, cfg.defaultLang) if err != nil { - slog.WarnContext(ctx, "enrichment backlog poll failed", "error", err) + consecutiveFailures++ + + // 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. + backlog.RecordPollError(ctx) + + if consecutiveFailures >= enrichmentBacklogFailuresBeforeError { + slog.ErrorContext(ctx, "enrichment backlog poll failing repeatedly; gauge is stale", + "error", err, "consecutive_failures", consecutiveFailures) + } else { + slog.WarnContext(ctx, "enrichment backlog poll failed", + "error", err, "consecutive_failures", consecutiveFailures) + } + + return + } + + consecutiveFailures = 0 + if !leader { + // Another replica refreshed the gauge this tick. return } diff --git a/internal/observability/enrichment_backlog.go b/internal/observability/enrichment_backlog.go index 928f69db..1da5dc57 100644 --- a/internal/observability/enrichment_backlog.go +++ b/internal/observability/enrichment_backlog.go @@ -24,14 +24,19 @@ const ( // deliberately NOT a label, so per-tenant detail stays in the API and metric cardinality is bounded. type EnrichmentBacklogMetrics interface { SetEnrichmentPending(enrichment string, count int64) + // 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(ctx context.Context) } // enrichmentBacklogMetrics implements EnrichmentBacklogMetrics. The latest per-enrichment value is // stored under mu and read by the gauge callback. type enrichmentBacklogMetrics struct { - mu sync.Mutex - pending map[string]int64 - gauge metric.Int64ObservableGauge + mu sync.Mutex + pending map[string]int64 + gauge metric.Int64ObservableGauge + pollErrors metric.Int64Counter } // NewEnrichmentBacklogMetrics registers the pending-records gauge. Returns (nil, nil) when meter is @@ -69,9 +74,29 @@ func NewEnrichmentBacklogMetrics(meter metric.Meter) (EnrichmentBacklogMetrics, m.gauge = gauge + 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.", + ), + metric.WithUnit("1"), + ) + if err != nil { + return nil, fmt.Errorf("create %s: %w", MetricNameEnrichmentBacklogPollErrs, err) + } + + m.pollErrors = pollErrors + return m, nil } +// RecordPollError counts one failed backlog refresh. +func (m *enrichmentBacklogMetrics) RecordPollError(ctx context.Context) { + m.pollErrors.Add(ctx, 1) +} + // SetEnrichmentPending stores the latest backlog count for an enrichment type; the registered gauge // callback reports it on the next collection. func (m *enrichmentBacklogMetrics) SetEnrichmentPending(enrichment string, count int64) { diff --git a/internal/observability/enrichment_backlog_test.go b/internal/observability/enrichment_backlog_test.go index eb35a011..037249fc 100644 --- a/internal/observability/enrichment_backlog_test.go +++ b/internal/observability/enrichment_backlog_test.go @@ -40,6 +40,43 @@ func TestEnrichmentBacklogMetricsGauge(t *testing.T) { assert.Equal(t, int64(3), backlogGaugeValue(t, reader, "translation")) } +// TestEnrichmentBacklogMetricsPollErrors verifies failed refreshes are counted, so a gauge frozen +// at its last value (which looks like a healthy steady backlog) is still alertable. +func TestEnrichmentBacklogMetricsPollErrors(t *testing.T) { + reader := sdkmetric.NewManualReader() + provider := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)) + + metrics, err := NewEnrichmentBacklogMetrics(provider.Meter("test")) + require.NoError(t, err) + require.NotNil(t, metrics) + + ctx := context.Background() + metrics.RecordPollError(ctx) + metrics.RecordPollError(ctx) + + var collected metricdata.ResourceMetrics + require.NoError(t, reader.Collect(ctx, &collected)) + + var total int64 + + for _, scope := range collected.ScopeMetrics { + for _, m := range scope.Metrics { + if m.Name != MetricNameEnrichmentBacklogPollErrs { + continue + } + + sum, ok := m.Data.(metricdata.Sum[int64]) + require.True(t, ok, "expected Sum[int64] for %s", MetricNameEnrichmentBacklogPollErrs) + + for _, point := range sum.DataPoints { + total += point.Value + } + } + } + + assert.Equal(t, int64(2), total, "each failed poll increments the error counter") +} + // backlogGaugeValue collects metrics and returns the pending-records gauge value for one enrichment. func backlogGaugeValue(t *testing.T, reader sdkmetric.Reader, enrichment string) int64 { t.Helper() diff --git a/internal/observability/names.go b/internal/observability/names.go index ecc87108..f78c6c96 100644 --- a/internal/observability/names.go +++ b/internal/observability/names.go @@ -16,6 +16,7 @@ const ( MetricNameHNSWIterativeScanDegraded = "hub_hnsw_iterative_scan_degraded" MetricNameEnrichmentOutputsCleared = "hub_enrichment_outputs_cleared_total" MetricNameEnrichmentPendingRecords = "hub_enrichment_pending_records" + MetricNameEnrichmentBacklogPollErrs = "hub_enrichment_backlog_poll_errors_total" MetricNameWebhookJobsEnqueued = "hub_webhook_jobs_enqueued_total" MetricNameWebhookProviderErrors = "hub_webhook_provider_errors_total" MetricNameWebhookDeliveries = "hub_webhook_deliveries_total" diff --git a/internal/repository/enrichment_status_repository.go b/internal/repository/enrichment_status_repository.go index cb30bf17..6152d749 100644 --- a/internal/repository/enrichment_status_repository.go +++ b/internal/repository/enrichment_status_repository.go @@ -4,6 +4,7 @@ import ( "context" "fmt" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" ) @@ -75,62 +76,108 @@ const enrichmentCountFrom = ` FROM feedback_records fr LEFT JOIN tenant_settings ts ON ts.tenant_id = fr.tenant_id` -// The `fr.field_type = 'text'` predicate in the outer WHERE below is redundant with -// enrichmentEligibleText inside every FILTER (so it can never change a count), but hoisting it out -// lets the planner use idx_feedback_records_tenant_field_type (tenant_id, field_type) to visit only -// text rows instead of scanning every row and discarding non-text inside the FILTERs. +// The `fr.field_type = 'text'` predicate in the outer WHERE of both queries below is redundant +// with enrichmentEligibleText inside every FILTER (so it can never change a count); it is hoisted +// out so the planner can use it as an access-path predicate rather than only as a per-row filter. // countEnrichmentStatusSQL counts eligible/done per enrichment for ONE tenant. $1 = deployment -// default target language, $2 = tenant_id. The (tenant_id, field_type) predicate is served by +// default target language, $2 = tenant_id. The (tenant_id, field_type) pair matches // idx_feedback_records_tenant_field_type, so cost scales with the tenant's text-record count. const countEnrichmentStatusSQL = `SELECT ` + enrichmentCountSelect + enrichmentCountFrom + ` WHERE fr.tenant_id = $2 AND fr.field_type = 'text'` // countEnrichmentBacklogAggregateSQL is the same SELECT without the tenant filter: it sums // eligible/done per enrichment across ALL tenants (for the observability gauge). $1 = deployment -// default target language. The field_type = 'text' predicate narrows the scan to text rows; the -// per-tenant enable gates still apply, so a tenant that switched an enrichment off, or has no -// resolvable target, never inflates the backlog. +// default target language. The per-tenant enable gates still apply, so a tenant that switched an +// enrichment off, or has no resolvable target, never inflates the backlog. +// +// NOTE: unlike the per-tenant query this one canNOT use idx_feedback_records_tenant_field_type -- +// tenant_id is that index's leading column, and this query has no tenant predicate -- so Postgres +// plans a sequential scan of feedback_records (confirmed via EXPLAIN). That is deliberate: the +// aggregate must read every text row anyway, so an index scan over most of the table would not be +// cheaper, and a dedicated partial index would add write amplification to the high-throughput +// ingest path we are protecting. The scan is instead bounded by running it infrequently +// (enrichmentBacklogInterval), under a statement timeout, and on ONE replica at a time via the +// advisory lock below. const countEnrichmentBacklogAggregateSQL = `SELECT ` + enrichmentCountSelect + enrichmentCountFrom + ` WHERE fr.field_type = 'text'` +// enrichmentBacklogLockKey names the advisory lock that elects a single backlog-poller run across +// API replicas. Hashed with hashtextextended like the other advisory locks in this package. +const enrichmentBacklogLockKey = "hub:enrichment-backlog-poller" + +// tryAdvisoryXactLockSQL takes a transaction-scoped advisory lock without blocking; it returns +// false when another session (replica) already holds it. Transaction scope means the lock is +// always released on commit/rollback, so a crashed poller cannot wedge the others. +const tryAdvisoryXactLockSQL = `SELECT pg_try_advisory_xact_lock(hashtextextended($1, 0))` + // CountEnrichmentStatus returns one tenant's eligible/done counts per enrichment. defaultLang is // the deployment translation fallback ("" disables the fallback, so only tenants with their own // target language have eligible translation records). Always scoped to the given tenant_id. func (r *EnrichmentStatusRepository) CountEnrichmentStatus( ctx context.Context, tenantID, defaultLang string, ) (EnrichmentStatusCounts, error) { - var counts EnrichmentStatusCounts - - // Scan order matches enrichmentCountSelect. - err := r.db.QueryRow(ctx, countEnrichmentStatusSQL, defaultLang, tenantID).Scan( - &counts.SentimentEligible, &counts.SentimentDone, - &counts.EmotionsEligible, &counts.EmotionsDone, - &counts.TranslationEligible, &counts.TranslationDone, - ) - if err != nil { - return EnrichmentStatusCounts{}, fmt.Errorf("count enrichment status: %w", err) - } - - return counts, nil + return scanEnrichmentCounts( + r.db.QueryRow(ctx, countEnrichmentStatusSQL, defaultLang, tenantID), "count enrichment status") } // CountEnrichmentBacklogAggregate returns eligible/done counts per enrichment summed across all -// tenants. defaultLang is the deployment translation fallback. Used by the observability poller; -// the result carries no tenant dimension. +// tenants. defaultLang is the deployment translation fallback. The result carries no tenant +// dimension. Prefer CountEnrichmentBacklogAggregateIfLeader from the poller so only one replica +// runs the scan. func (r *EnrichmentStatusRepository) CountEnrichmentBacklogAggregate( ctx context.Context, defaultLang string, ) (EnrichmentStatusCounts, error) { + return scanEnrichmentCounts( + r.db.QueryRow(ctx, countEnrichmentBacklogAggregateSQL, defaultLang), "count enrichment backlog aggregate") +} + +// CountEnrichmentBacklogAggregateIfLeader runs the cross-tenant aggregate only if this process wins +// a non-blocking advisory lock, and reports whether it did. Production runs several API replicas +// per region, and each would otherwise repeat the same full-table scan on every tick and publish +// the same global gauge — multiplying DB load and producing duplicate series that a naive sum would +// over-count. Losing the race is normal, not an error: the winner refreshes the gauge for everyone. +// The lock is transaction-scoped, so it is released even if this process dies mid-scan. +func (r *EnrichmentStatusRepository) CountEnrichmentBacklogAggregateIfLeader( + ctx context.Context, defaultLang string, +) (EnrichmentStatusCounts, bool, error) { + backlogTx, err := r.db.Begin(ctx) + if err != nil { + return EnrichmentStatusCounts{}, false, fmt.Errorf("begin enrichment backlog tx: %w", err) + } + + // Read-only work: always roll back (releasing the advisory lock); a rollback after success is + // equivalent to a commit here and avoids leaking the tx on any early return. + defer func() { _ = backlogTx.Rollback(ctx) }() + + var acquired bool + if err := backlogTx.QueryRow(ctx, tryAdvisoryXactLockSQL, enrichmentBacklogLockKey).Scan(&acquired); err != nil { + return EnrichmentStatusCounts{}, false, fmt.Errorf("try enrichment backlog advisory lock: %w", err) + } + + if !acquired { + return EnrichmentStatusCounts{}, false, nil + } + + counts, err := scanEnrichmentCounts( + backlogTx.QueryRow(ctx, countEnrichmentBacklogAggregateSQL, defaultLang), "count enrichment backlog aggregate") + if err != nil { + return EnrichmentStatusCounts{}, false, err + } + + return counts, true, nil +} + +// scanEnrichmentCounts reads the six-column count row; the scan order matches enrichmentCountSelect. +func scanEnrichmentCounts(row pgx.Row, what string) (EnrichmentStatusCounts, error) { var counts EnrichmentStatusCounts - // Scan order matches enrichmentCountSelect. - err := r.db.QueryRow(ctx, countEnrichmentBacklogAggregateSQL, defaultLang).Scan( + if err := row.Scan( &counts.SentimentEligible, &counts.SentimentDone, &counts.EmotionsEligible, &counts.EmotionsDone, &counts.TranslationEligible, &counts.TranslationDone, - ) - if err != nil { - return EnrichmentStatusCounts{}, fmt.Errorf("count enrichment backlog aggregate: %w", err) + ); err != nil { + return EnrichmentStatusCounts{}, fmt.Errorf("%s: %w", what, err) } return counts, nil diff --git a/tests/enrichment_status_test.go b/tests/enrichment_status_test.go index 221b2f24..754feabd 100644 --- a/tests/enrichment_status_test.go +++ b/tests/enrichment_status_test.go @@ -237,3 +237,52 @@ func TestCountEnrichmentBacklogAggregate(t *testing.T) { assert.Equal(t, int64(5), afterEmo-beforeEmo, "emotions default-enabled for both tenants → 3+2") assert.Equal(t, int64(5), afterTrans-beforeTrans, "en-US default makes all 5 records translation-pending") } + +// TestCountEnrichmentBacklogAggregateIfLeader covers the single-flight advisory lock that stops +// every API replica from repeating the same cross-tenant scan: one caller wins and gets the counts, +// a concurrent caller is denied (not an error) and skips its tick, and the lock is released +// afterwards so the next tick can win again. +func TestCountEnrichmentBacklogAggregateIfLeader(t *testing.T) { + ctx := context.Background() + + cfg, err := config.Load() + require.NoError(t, err) + + // Two independent pools stand in for two API replicas: a transaction-scoped advisory lock is + // held per session, so the contention is only observable across separate connections. + dbLeader, err := database.NewPostgresPool(ctx, cfg.Database.URL, database.WithPoolConfig(cfg.Database.PoolConfig())) + require.NoError(t, err) + + defer dbLeader.Close() + + dbRival, err := database.NewPostgresPool(ctx, cfg.Database.URL, database.WithPoolConfig(cfg.Database.PoolConfig())) + require.NoError(t, err) + + defer dbRival.Close() + + leaderRepo := repository.NewEnrichmentStatusRepository(dbLeader) + rivalRepo := repository.NewEnrichmentStatusRepository(dbRival) + + // Hold the lock by starting a scan on one "replica" and blocking before it commits: emulate by + // taking the lock in an explicit transaction on the leader pool. + holderTx, err := dbLeader.Begin(ctx) + require.NoError(t, err) + + var held bool + require.NoError(t, holderTx.QueryRow(ctx, + `SELECT pg_try_advisory_xact_lock(hashtextextended('hub:enrichment-backlog-poller', 0))`).Scan(&held)) + require.True(t, held, "test must hold the poller lock to simulate a busy replica") + + // While it is held, the other replica must be denied — and that is a normal skip, not an error. + _, acquired, err := rivalRepo.CountEnrichmentBacklogAggregateIfLeader(ctx, "") + require.NoError(t, err, "losing the leader race is not an error") + assert.False(t, acquired, "a second replica must not run the aggregate concurrently") + + // Releasing the transaction releases the lock, so the next tick can win. + require.NoError(t, holderTx.Rollback(ctx)) + + counts, acquired, err := leaderRepo.CountEnrichmentBacklogAggregateIfLeader(ctx, "") + require.NoError(t, err) + assert.True(t, acquired, "lock is free again after the holder's transaction ends") + assert.GreaterOrEqual(t, counts.SentimentEligible, int64(0), "leader gets real counts") +} From ef4af84df8ed1871f42adcb79603b5d678d61d7c Mon Sep 17 00:00:00 2001 From: Tiago Farto Date: Mon, 27 Jul 2026 09:31:07 +0000 Subject: [PATCH 10/13] fix(enrichment): emotions completion marker + sticky poller leadership (ENG-1670) Emotions could never reach "done". A successful classification that detects no emotion is stored as NULL (the 015 CHECK rejects the empty array), which is the same value that means "never classified", so those records counted as pending forever and the backlog gauge could not drain. Adds emotions_classified_at (migration 020) to record completion independently of the labels found: - SetEmotions stamps it (an empty result is still a completed classification); the new ClearEmotions clears both, for the content-is-gone path. - The eager-clear on a value_text edit drops the marker with the labels. - "done" is (emotions_classified_at IS NOT NULL OR emotions IS NOT NULL); the second arm covers rows enriched before the column existed, so the migration needs no bulk UPDATE of the primary write table. - The classify backfill now skips already-classified rows instead of re-sending every classified-empty record to the provider on each run. The marker is processing state and stays out of the API surface, so no response shape, OpenAPI or SDK change. Sentiment and translation are unaffected: their successful results are single-valued and always non-NULL (verified). Also replaces the previous backlog-poller advisory lock, which did not do what it claimed. It was transaction-scoped, so it was held only for the ~seconds a scan takes, while replicas tick on independent unsynchronized schedules -- they almost never collided, so neither the duplicated DB load nor the duplicate series was actually prevented, and a collision on a replica's first poll left it exporting no series at all until the next tick. Leadership is now a sticky session-scoped lock held on a dedicated connection for the process lifetime: exactly one replica scans and exports, the series stays put instead of flapping, followers never pin a connection, and Close hands over promptly on shutdown. Also stops counting a shutdown-cancelled scan as a poll error, which would otherwise fire the staleness alert on every rolling deploy. --- cmd/api/app.go | 17 +- .../enrichment_status_repository.go | 181 ++++++++++++++---- .../repository/feedback_records_repository.go | 97 +++++++--- internal/service/feedback_records_service.go | 25 ++- .../service/feedback_records_service_test.go | 13 +- internal/workers/feedback_emotions.go | 13 +- internal/workers/feedback_emotions_test.go | 30 ++- migrations/020_add_emotions_classified_at.sql | 28 +++ tests/enrichment_status_test.go | 90 +++++++-- 9 files changed, 385 insertions(+), 109 deletions(-) create mode 100644 migrations/020_add_emotions_classified_at.sql diff --git a/cmd/api/app.go b/cmd/api/app.go index dbaf3bc8..a1c578fd 100644 --- a/cmd/api/app.go +++ b/cmd/api/app.go @@ -787,7 +787,8 @@ func runEnrichmentBacklogPoller( backlog observability.EnrichmentBacklogMetrics, cfg enrichmentBacklogPollConfig, ) { - repo := repository.NewEnrichmentStatusRepository(db) + leader := repository.NewEnrichmentBacklogLeader(db) + defer leader.Close(ctx) ticker := time.NewTicker(enrichmentBacklogInterval) defer ticker.Stop() @@ -798,9 +799,15 @@ func runEnrichmentBacklogPoller( queryCtx, cancel := context.WithTimeout(ctx, enrichmentBacklogQueryTimeout) defer cancel() - // Only the replica that wins the advisory lock scans; the others skip this tick. - counts, leader, err := repo.CountEnrichmentBacklogAggregateIfLeader(queryCtx, cfg.defaultLang) + // Exactly one replica holds leadership and scans; the rest skip until it goes away. + counts, isLeader, err := leader.CountIfLeader(queryCtx, cfg.defaultLang) if err != nil { + // Shutdown cancels the scan mid-flight. That is not a poll failure, and counting it + // would fire the very alert this counter exists for on every rolling deploy. + if ctx.Err() != nil { + return + } + consecutiveFailures++ // Always count the failure so a stale gauge is alertable, then escalate the log from @@ -821,8 +828,8 @@ func runEnrichmentBacklogPoller( consecutiveFailures = 0 - if !leader { - // Another replica refreshed the gauge this tick. + if !isLeader { + // Another replica owns this gauge and exports the single global series for it. return } diff --git a/internal/repository/enrichment_status_repository.go b/internal/repository/enrichment_status_repository.go index 6152d749..0a580efa 100644 --- a/internal/repository/enrichment_status_repository.go +++ b/internal/repository/enrichment_status_repository.go @@ -3,6 +3,7 @@ package repository import ( "context" "fmt" + "time" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" @@ -34,14 +35,22 @@ type EnrichmentStatusCounts struct { } // enrichmentEligibleText is the data-level eligibility predicate: an open-text field with content. -// It mirrors the backfill eligibility (classifyBackfillEligibleSQL / translationBackfillSelectSQL), -// trimming the full ASCII whitespace set (space, tab, VT, FF, CR, LF) before the emptiness check. -// This approximates the workers' HasOpenText gate (Go strings.TrimSpace, which additionally strips -// exotic Unicode whitespace such as NBSP U+00A0 or the ideographic space U+3000); a value composed -// ENTIRELY of exotic Unicode whitespace is a rare edge that would read as eligible-but-never-done -// here — an accepted approximation, consistent with the existing backfill queries. field_type = -// 'text' is load-bearing: matrix/multi-choice expansion writes value_text on categorical/number -// rows that are not enrichable. +// +// It trims the full ASCII whitespace set (space, tab, VT, FF, CR, LF). This deliberately does NOT +// match the backfill queries (classifyBackfillEligibleSQL / translationBackfillSelectSQL), whose +// bare btrim() strips spaces only: a value of "\t\n" is enqueued by those but counted ineligible +// here. That asymmetry is intentional -- what matters for a progress count is agreeing with the +// WORKER, which gates on Go strings.TrimSpace (HasOpenText) and would clear such a record rather +// than enrich it. Counting it eligible would leave it pending forever. Do not "restore parity" by +// weakening this to bare btrim. +// +// It remains an approximation in one direction: strings.TrimSpace also strips exotic Unicode +// whitespace (NBSP U+00A0, ideographic space U+3000, ...), so a value composed ENTIRELY of those is +// still counted eligible while the worker treats it as empty. Rare enough to accept; expressing the +// full Unicode set here would mean embedding invisible characters in this source file. +// +// field_type = 'text' is load-bearing: matrix/multi-choice expansion writes value_text on +// categorical/number rows that are not enrichable. const enrichmentEligibleText = `fr.field_type = 'text' AND fr.value_text IS NOT NULL AND btrim(fr.value_text, E' \t\n\v\f\r') <> ''` // enrichmentEffectiveTarget resolves a tenant's effective translation target: its own @@ -57,6 +66,16 @@ const ( enrichmentEmotionsOn = `COALESCE((ts.settings->>'emotions_enabled')::boolean, true)` ) +// enrichmentEmotionsDone reports that emotion classification has completed for a record. It cannot +// be `emotions IS NOT NULL` alone: a successful classification that detects no emotion is stored as +// NULL (the 015 CHECK rejects the empty array), so such records would count as pending forever and +// the backlog would never drain. emotions_classified_at (migration 020) records completion +// independently of the labels found. The `emotions IS NOT NULL` arm covers rows classified BEFORE +// that column existed -- a non-NULL label set is itself proof of completion -- which is why the +// migration needs no bulk backfill; only historical classified-empty rows remain unresolved until +// the classify backfill re-processes them. +const enrichmentEmotionsDone = `(fr.emotions_classified_at IS NOT NULL OR fr.emotions IS NOT NULL)` + // enrichmentCountSelect is the shared six-column SELECT list — {sentiment, emotions, translation} × // {eligible, done} — used by both the per-tenant and aggregate queries so the predicates can't // drift between them. Column order must match the EnrichmentStatusCounts scan order below. All @@ -65,7 +84,7 @@ const enrichmentCountSelect = ` COUNT(*) FILTER (WHERE ` + enrichmentEligibleText + ` AND ` + enrichmentSentimentOn + `), COUNT(*) FILTER (WHERE ` + enrichmentEligibleText + ` AND ` + enrichmentSentimentOn + ` AND fr.sentiment IS NOT NULL), COUNT(*) FILTER (WHERE ` + enrichmentEligibleText + ` AND ` + enrichmentEmotionsOn + `), - COUNT(*) FILTER (WHERE ` + enrichmentEligibleText + ` AND ` + enrichmentEmotionsOn + ` AND fr.emotions IS NOT NULL), + COUNT(*) FILTER (WHERE ` + enrichmentEligibleText + ` AND ` + enrichmentEmotionsOn + ` AND ` + enrichmentEmotionsDone + `), COUNT(*) FILTER (WHERE ` + enrichmentEligibleText + ` AND ` + enrichmentEffectiveTarget + ` <> ''), COUNT(*) FILTER ( WHERE ` + enrichmentEligibleText + ` @@ -92,24 +111,32 @@ const countEnrichmentStatusSQL = `SELECT ` + enrichmentCountSelect + enrichmentC // enrichment off, or has no resolvable target, never inflates the backlog. // // NOTE: unlike the per-tenant query this one canNOT use idx_feedback_records_tenant_field_type -- -// tenant_id is that index's leading column, and this query has no tenant predicate -- so Postgres -// plans a sequential scan of feedback_records (confirmed via EXPLAIN). That is deliberate: the -// aggregate must read every text row anyway, so an index scan over most of the table would not be -// cheaper, and a dedicated partial index would add write amplification to the high-throughput -// ingest path we are protecting. The scan is instead bounded by running it infrequently -// (enrichmentBacklogInterval), under a statement timeout, and on ONE replica at a time via the -// advisory lock below. +// tenant_id is that index's leading column and this query has no tenant predicate -- so Postgres +// plans a sequential scan of feedback_records (confirmed via EXPLAIN). Accepted rather than fixed +// with a new index: the aggregate has to read every text row regardless, so an index scan covering +// most of the table would not be cheaper. (Migration 016 does already maintain partial indexes over +// unenriched text rows, so a further index is not unthinkable -- it is simply unlikely to pay off +// for a whole-table aggregate.) The cost is instead bounded by running the scan infrequently +// (enrichmentBacklogInterval), under a statement timeout, and on exactly ONE replica via the +// leader election below. const countEnrichmentBacklogAggregateSQL = `SELECT ` + enrichmentCountSelect + enrichmentCountFrom + ` WHERE fr.field_type = 'text'` -// enrichmentBacklogLockKey names the advisory lock that elects a single backlog-poller run across -// API replicas. Hashed with hashtextextended like the other advisory locks in this package. +// enrichmentBacklogLockKey names the advisory lock that elects the single backlog-poller process +// across API replicas. Hashed with hashtextextended like the other advisory locks in this package. const enrichmentBacklogLockKey = "hub:enrichment-backlog-poller" -// tryAdvisoryXactLockSQL takes a transaction-scoped advisory lock without blocking; it returns -// false when another session (replica) already holds it. Transaction scope means the lock is -// always released on commit/rollback, so a crashed poller cannot wedge the others. -const tryAdvisoryXactLockSQL = `SELECT pg_try_advisory_xact_lock(hashtextextended($1, 0))` +// enrichmentBacklogUnlockTimeout bounds the best-effort advisory unlock on shutdown. +const enrichmentBacklogUnlockTimeout = 5 * time.Second + +const ( + // trySessionLockSQL takes a SESSION-scoped advisory lock without blocking. Session scope is + // deliberate -- see EnrichmentBacklogLeader for why a transaction-scoped lock cannot work here. + trySessionLockSQL = `SELECT pg_try_advisory_lock(hashtextextended($1, 0))` + // sessionUnlockSQL releases it. A session lock outlives returning the connection to the pool, + // so it must be released explicitly. + sessionUnlockSQL = `SELECT pg_advisory_unlock(hashtextextended($1, 0))` +) // CountEnrichmentStatus returns one tenant's eligible/done counts per enrichment. defaultLang is // the deployment translation fallback ("" disables the fallback, so only tenants with their own @@ -132,40 +159,110 @@ func (r *EnrichmentStatusRepository) CountEnrichmentBacklogAggregate( r.db.QueryRow(ctx, countEnrichmentBacklogAggregateSQL, defaultLang), "count enrichment backlog aggregate") } -// CountEnrichmentBacklogAggregateIfLeader runs the cross-tenant aggregate only if this process wins -// a non-blocking advisory lock, and reports whether it did. Production runs several API replicas -// per region, and each would otherwise repeat the same full-table scan on every tick and publish -// the same global gauge — multiplying DB load and producing duplicate series that a naive sum would -// over-count. Losing the race is normal, not an error: the winner refreshes the gauge for everyone. -// The lock is transaction-scoped, so it is released even if this process dies mid-scan. -func (r *EnrichmentStatusRepository) CountEnrichmentBacklogAggregateIfLeader( +// EnrichmentBacklogLeader elects ONE process to refresh the cross-tenant backlog gauge. +// +// Production runs several API replicas per region. Without election each replica repeats the same +// full-table aggregate every tick and exports its own copy of a value that is global by definition: +// N times the DB work, and N identical series that a dashboard summing them silently over-counts. +// +// Leadership is a SESSION-scoped advisory lock held on a dedicated pooled connection for the +// process lifetime, NOT a lock taken around each scan. That distinction is the whole point: +// replicas tick on independent, unsynchronized schedules (each ticker starts at its own boot), and +// a scan-scoped lock is held for only a couple of seconds out of every interval, so replicas would +// virtually never collide -- suppressing nothing, while occasionally blanking one replica's series +// when they did collide. Sticky leadership instead means exactly one replica scans and exports, and +// the series stays put instead of flapping between replicas. +// +// A non-leader never holds a connection: it acquires one, loses the race, and hands it straight +// back. If the leader's connection dies its backend session ends and Postgres drops the lock +// automatically, so the next tick re-elects; the process also drops leadership itself whenever a +// scan fails, so it cannot keep believing it is the leader after losing the session. +type EnrichmentBacklogLeader struct { + pool *pgxpool.Pool + conn *pgxpool.Conn // non-nil only while this process holds leadership +} + +// NewEnrichmentBacklogLeader creates a leader-elected reader for the aggregate backlog counts. +func NewEnrichmentBacklogLeader(pool *pgxpool.Pool) *EnrichmentBacklogLeader { + return &EnrichmentBacklogLeader{pool: pool} +} + +// CountIfLeader returns the cross-tenant counts when this process holds (or wins) leadership, and +// reports whether it did. Not being the leader is the normal steady state for all but one replica, +// so it is signalled by a false second return rather than an error. +func (l *EnrichmentBacklogLeader) CountIfLeader( ctx context.Context, defaultLang string, ) (EnrichmentStatusCounts, bool, error) { - backlogTx, err := r.db.Begin(ctx) + if l.conn == nil { + acquired, err := l.tryAcquire(ctx) + if err != nil || !acquired { + return EnrichmentStatusCounts{}, false, err + } + } + + // Deliberately run on the leader connection: a broken session then surfaces here as a scan + // error rather than silently leaving this process convinced it still holds the lock. + counts, err := scanEnrichmentCounts( + l.conn.QueryRow(ctx, countEnrichmentBacklogAggregateSQL, defaultLang), "count enrichment backlog aggregate") if err != nil { - return EnrichmentStatusCounts{}, false, fmt.Errorf("begin enrichment backlog tx: %w", err) + l.release(ctx) + + return EnrichmentStatusCounts{}, false, err } - // Read-only work: always roll back (releasing the advisory lock); a rollback after success is - // equivalent to a commit here and avoids leaking the tx on any early return. - defer func() { _ = backlogTx.Rollback(ctx) }() + return counts, true, nil +} + +// Close relinquishes leadership so another replica can take over promptly instead of waiting for +// this process's session to time out. Safe to call when not the leader. +func (l *EnrichmentBacklogLeader) Close(ctx context.Context) { + l.release(ctx) +} + +// tryAcquire takes a connection and attempts the session lock, keeping the connection only on +// success -- a non-leader must not pin a pooled connection it will not use. +func (l *EnrichmentBacklogLeader) tryAcquire(ctx context.Context) (bool, error) { + conn, err := l.pool.Acquire(ctx) + if err != nil { + return false, fmt.Errorf("acquire enrichment backlog leader connection: %w", err) + } var acquired bool - if err := backlogTx.QueryRow(ctx, tryAdvisoryXactLockSQL, enrichmentBacklogLockKey).Scan(&acquired); err != nil { - return EnrichmentStatusCounts{}, false, fmt.Errorf("try enrichment backlog advisory lock: %w", err) + if err := conn.QueryRow(ctx, trySessionLockSQL, enrichmentBacklogLockKey).Scan(&acquired); err != nil { + conn.Release() + + return false, fmt.Errorf("try enrichment backlog advisory lock: %w", err) } if !acquired { - return EnrichmentStatusCounts{}, false, nil + conn.Release() + + return false, nil } - counts, err := scanEnrichmentCounts( - backlogTx.QueryRow(ctx, countEnrichmentBacklogAggregateSQL, defaultLang), "count enrichment backlog aggregate") - if err != nil { - return EnrichmentStatusCounts{}, false, err + l.conn = conn + + return true, nil +} + +// release unlocks and returns the leader connection. The unlock uses a context detached from the +// caller's, so shutdown (whose context is already cancelled) still releases the lock rather than +// leaving it held until the backend session is reaped. +func (l *EnrichmentBacklogLeader) release(ctx context.Context) { + if l.conn == nil { + return } - return counts, true, nil + unlockCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), enrichmentBacklogUnlockTimeout) + defer cancel() + + // Best effort: if this fails the connection is broken, and ending that session releases the + // lock anyway. Returning the connection WITHOUT unlocking would be the real leak, since a + // session lock survives being handed back to the pool. + _, _ = l.conn.Exec(unlockCtx, sessionUnlockSQL, enrichmentBacklogLockKey) + + l.conn.Release() + l.conn = nil } // scanEnrichmentCounts reads the six-column count row; the scan order matches enrichmentCountSelect. diff --git a/internal/repository/feedback_records_repository.go b/internal/repository/feedback_records_repository.go index 0585c131..2adf18df 100644 --- a/internal/repository/feedback_records_repository.go +++ b/internal/repository/feedback_records_repository.go @@ -426,36 +426,21 @@ func (r *FeedbackRecordsRepository) SetEmotions( emotionsArg = labels } - return withTenantWritePoolTx(ctx, r.db, nil, func(dbTx tenantWriteTx) error { - // Emotions ride the feedback record's tenant boundary; resolve and lock it so the write - // cannot race a tenant data purge. - if _, err := lockFeedbackRecordTenantShared(ctx, dbTx, feedbackRecordID); err != nil { - return err - } - - if err := guardValueTextCurrent(ctx, dbTx, feedbackRecordID, stillCurrent, - huberrors.ErrClassificationSuperseded); err != nil { - return err - } - - tag, err := dbTx.Exec(ctx, ` - UPDATE feedback_records - SET emotions = $2, updated_at = NOW() - WHERE id = $1`, - feedbackRecordID, emotionsArg, - ) - if err != nil { - return fmt.Errorf("set feedback record emotions: %w", err) - } - - // Locked above, so zero rows means the record was deleted between the lock and this write: - // surface NotFound so the worker treats it as a benign skip. - if tag.RowsAffected() == 0 { - return huberrors.NewNotFoundError("feedback record", "feedback record not found") - } + // A classifier result -- even an empty one -- means the record HAS been classified, so stamp + // the completion marker. Without it an empty result is stored as a bare NULL and is + // indistinguishable from "never classified", which pins the record in every pending count + // forever (ENG-1670). Use ClearEmotions for the not-classified transition. + return r.writeEmotions(ctx, feedbackRecordID, emotionsArg, true, stillCurrent) +} - return nil - }) +// ClearEmotions removes a record's emotion enrichment AND its completion marker, returning it to +// the "not classified" state. Used when the source content is gone (an empty-content job clears +// rather than classifies), so the record is correctly excluded from progress counts instead of +// masquerading as classified-with-no-emotions. +func (r *FeedbackRecordsRepository) ClearEmotions( + ctx context.Context, feedbackRecordID uuid.UUID, stillCurrent func(valueText *string) bool, +) error { + return r.writeEmotions(ctx, feedbackRecordID, nil, false, stillCurrent) } // translationBackfillSelectSQL selects feedback records that need (re)translation: text @@ -557,8 +542,14 @@ const sentimentBackfillSelectSQL = classifyBackfillEligibleSQL + ` ORDER BY id LIMIT $2` +// Both NULL checks are required. emotions IS NULL alone would re-send every record whose +// classification legitimately found no emotion (stored as NULL) to the provider on every run; +// emotions_classified_at IS NULL alone would re-classify rows enriched before that column existed +// (migration 020). Together they select exactly the records never classified. The 016 partial index +// covers the emotions IS NULL arm; the marker is an additional filter on top. const emotionsBackfillSelectSQL = classifyBackfillEligibleSQL + ` AND emotions IS NULL + AND emotions_classified_at IS NULL AND id > $1 ORDER BY id LIMIT $2` @@ -907,6 +898,9 @@ func buildUpdateQuery( clearColumnWhen("sentiment", valueTextChanged), clearColumnWhen("sentiment_score", valueTextChanged), clearColumnWhen("emotions", valueTextChanged), + // The completion marker must be cleared with the value it describes, otherwise an + // edited record keeps counting as classified while its emotions are gone (ENG-1670). + clearColumnWhen("emotions_classified_at", valueTextChanged), ) } @@ -1215,3 +1209,48 @@ func (r *FeedbackRecordsRepository) fetchFeedbackRecords( return records, nil } + +// writeEmotions is the shared write path for SetEmotions/ClearEmotions. classified controls the +// completion marker: NOW() for a classifier result, NULL when clearing. +func (r *FeedbackRecordsRepository) writeEmotions( + ctx context.Context, feedbackRecordID uuid.UUID, emotionsArg any, classified bool, + stillCurrent func(valueText *string) bool, +) error { + return withTenantWritePoolTx(ctx, r.db, nil, func(dbTx tenantWriteTx) error { + // Emotions ride the feedback record's tenant boundary; resolve and lock it so the write + // cannot race a tenant data purge. + if _, err := lockFeedbackRecordTenantShared(ctx, dbTx, feedbackRecordID); err != nil { + return err + } + + if err := guardValueTextCurrent(ctx, dbTx, feedbackRecordID, stillCurrent, + huberrors.ErrClassificationSuperseded); err != nil { + return err + } + + // classifiedAt is NULL when clearing and NOW() for a classifier result, so the marker always + // travels with the value it describes. + var classifiedAt any + if classified { + classifiedAt = time.Now() + } + + tag, err := dbTx.Exec(ctx, ` + UPDATE feedback_records + SET emotions = $2, emotions_classified_at = $3, updated_at = NOW() + WHERE id = $1`, + feedbackRecordID, emotionsArg, classifiedAt, + ) + if err != nil { + return fmt.Errorf("set feedback record emotions: %w", err) + } + + // Locked above, so zero rows means the record was deleted between the lock and this write: + // surface NotFound so the worker treats it as a benign skip. + if tag.RowsAffected() == 0 { + return huberrors.NewNotFoundError("feedback record", "feedback record not found") + } + + return nil + }) +} diff --git a/internal/service/feedback_records_service.go b/internal/service/feedback_records_service.go index cfb0f4c8..b4e84975 100644 --- a/internal/service/feedback_records_service.go +++ b/internal/service/feedback_records_service.go @@ -58,6 +58,8 @@ type FeedbackRecordsRepository interface { //nolint:interfacebloat // one cohesi stillCurrent func(valueText *string) bool) error SetEmotions(ctx context.Context, feedbackRecordID uuid.UUID, emotions []models.EmotionValue, stillCurrent func(valueText *string) bool) error + ClearEmotions(ctx context.Context, feedbackRecordID uuid.UUID, + stillCurrent func(valueText *string) bool) error ListTranslationBackfillTargets( ctx context.Context, afterID uuid.UUID, limit int, defaultLang string, ) ([]models.TranslationBackfillTarget, error) @@ -260,16 +262,18 @@ func (s *FeedbackRecordsService) SetSentiment( // (no enrichment loop). stillCurrent (optional) is the repository's content-supersession guard: // it is given the record's current value_text atomically with the write, and a false return skips // the write with huberrors.ErrClassificationSuperseded (nil ⇒ unconditional). Emotions are -// multi-label; an empty (or nil) set clears the column, so "no emotion detected" and "not yet -// enriched" share the same NULL representation. +// multi-label; an empty (or nil) set still records a COMPLETED classification (the labels column +// stays NULL, but the completion marker is stamped), which is what distinguishes "no emotion +// detected" from "not yet enriched" -- use ClearEmotions for the latter. func (s *FeedbackRecordsService) SetEmotions( ctx context.Context, feedbackRecordID uuid.UUID, emotions []models.EmotionValue, stillCurrent func(valueText *string) bool, ) error { - // An empty set clears (stored as NULL, never an empty array). + // An empty set stores NULL labels (never an empty array) but is still a classification result, + // so it goes through SetEmotions -- not ClearEmotions -- to stamp the completion marker. if len(emotions) == 0 { if err := s.repo.SetEmotions(ctx, feedbackRecordID, nil, stillCurrent); err != nil { - return fmt.Errorf("clear feedback record emotions: %w", err) + return fmt.Errorf("set empty feedback record emotions: %w", err) } return nil @@ -288,6 +292,19 @@ func (s *FeedbackRecordsService) SetEmotions( return nil } +// ClearEmotions returns a record to the "not classified" state, dropping both the labels and the +// completion marker. Used when the source content is gone, so the record is excluded from progress +// counts rather than counting as classified-with-no-emotions. +func (s *FeedbackRecordsService) ClearEmotions( + ctx context.Context, feedbackRecordID uuid.UUID, stillCurrent func(valueText *string) bool, +) error { + if err := s.repo.ClearEmotions(ctx, feedbackRecordID, stillCurrent); err != nil { + return fmt.Errorf("clear feedback record emotions: %w", err) + } + + return nil +} + // ListFeedbackRecords retrieves a list of feedback records with optional filters. // Uses cursor-based pagination: omit cursor for first page, use next_cursor for subsequent pages. func (s *FeedbackRecordsService) ListFeedbackRecords( diff --git a/internal/service/feedback_records_service_test.go b/internal/service/feedback_records_service_test.go index 92299cdd..cacccd12 100644 --- a/internal/service/feedback_records_service_test.go +++ b/internal/service/feedback_records_service_test.go @@ -38,8 +38,9 @@ type mockFeedbackRecordsRepo struct { setSentimentLabel *models.SentimentValue setSentimentScore *float64 - setEmotionsCalled bool - setEmotionsLabels []models.EmotionValue + setEmotionsCalled bool + setEmotionsLabels []models.EmotionValue + clearEmotionsCalled bool } func (m *mockFeedbackRecordsRepo) Create( @@ -103,6 +104,14 @@ func (m *mockFeedbackRecordsRepo) SetSentiment( return nil } +func (m *mockFeedbackRecordsRepo) ClearEmotions( + _ context.Context, _ uuid.UUID, _ func(valueText *string) bool, +) error { + m.clearEmotionsCalled = true + + return nil +} + func (m *mockFeedbackRecordsRepo) SetEmotions( _ context.Context, _ uuid.UUID, emotions []models.EmotionValue, _ func(valueText *string) bool, ) error { diff --git a/internal/workers/feedback_emotions.go b/internal/workers/feedback_emotions.go index 653c6454..66ff1e59 100644 --- a/internal/workers/feedback_emotions.go +++ b/internal/workers/feedback_emotions.go @@ -24,6 +24,10 @@ type emotionsWorkerService interface { GetFeedbackRecord(ctx context.Context, id uuid.UUID) (*models.FeedbackRecord, error) SetEmotions(ctx context.Context, feedbackRecordID uuid.UUID, emotions []models.EmotionValue, stillCurrent func(valueText *string) bool) error + // ClearEmotions returns the record to "not classified" (used when the source content is gone), + // as opposed to SetEmotions with an empty result, which is a completed classification. + ClearEmotions(ctx context.Context, feedbackRecordID uuid.UUID, + stillCurrent func(valueText *string) bool) error } // tenantSettingsReader resolves a tenant's enrichment settings for the worker's authoritative @@ -58,11 +62,14 @@ func NewFeedbackEmotionsWorker( }, persist: func(ctx context.Context, record *models.FeedbackRecord, _ service.FeedbackEmotionsArgs, result *service.EmotionsResult) error { // Guard the write against content churn since the Work-time read: a stale job's labels - // (or clear) must not land last over a newer job's write. A nil result (empty content) - // or an empty label set both clear the column: absence is NULL, never an empty array. + // (or clear) must not land last over a newer job's write. Both paths leave the column + // NULL when there are no labels (absence is NULL, never an empty array), but they are + // NOT the same state: a nil result means the content is gone, whereas an empty label + // set is a completed classification. ClearEmotions/SetEmotions record that difference + // in emotions_classified_at so progress counts can tell them apart (ENG-1670). stillCurrent := valueTextStillCurrent(record.ValueText) if result == nil { - return svc.SetEmotions(ctx, record.ID, nil, stillCurrent) + return svc.ClearEmotions(ctx, record.ID, stillCurrent) } return svc.SetEmotions(ctx, record.ID, result.Labels, stillCurrent) diff --git a/internal/workers/feedback_emotions_test.go b/internal/workers/feedback_emotions_test.go index 09426f14..6477259d 100644 --- a/internal/workers/feedback_emotions_test.go +++ b/internal/workers/feedback_emotions_test.go @@ -58,10 +58,11 @@ func (m *countingEmotionsMetrics) RecordEmotionsDuration(_ context.Context, _ ti var _ observability.EmotionsMetrics = (*countingEmotionsMetrics)(nil) type mockEmotionsWorkerService struct { - record *models.FeedbackRecord - getErr error - setErr error - setCalls [][]models.EmotionValue + record *models.FeedbackRecord + getErr error + setErr error + setCalls [][]models.EmotionValue + clearCalls int } func (m *mockEmotionsWorkerService) GetFeedbackRecord(_ context.Context, _ uuid.UUID) (*models.FeedbackRecord, error) { @@ -76,6 +77,16 @@ func (m *mockEmotionsWorkerService) SetEmotions( return m.setErr } +// ClearEmotions is the "content is gone" path, distinct from SetEmotions with an empty result +// (a completed classification). Counted separately so tests can assert which one the worker took. +func (m *mockEmotionsWorkerService) ClearEmotions( + _ context.Context, _ uuid.UUID, _ func(valueText *string) bool, +) error { + m.clearCalls++ + + return m.setErr +} + type stubEmotionsClient struct { result service.EmotionsResult err error @@ -186,8 +197,15 @@ func TestFeedbackEmotionsWorker_EmptyValueTextClears(t *testing.T) { t.Fatalf("Classify calls = %d, want 0 (empty text is not classified)", client.calls) } - if len(svc.setCalls) != 1 || len(svc.setCalls[0]) != 0 { - t.Fatalf("setCalls = %+v, want one clear (empty set)", svc.setCalls) + // Empty content must take the CLEAR path, not a classification with an empty result: + // both leave the labels NULL, but only a classification stamps the completion marker, + // and an unclassifiable record must not be reported as "done" (ENG-1670). + if svc.clearCalls != 1 { + t.Fatalf("clearCalls = %d, want 1 (empty text clears rather than classifies)", svc.clearCalls) + } + + if len(svc.setCalls) != 0 { + t.Fatalf("setCalls = %+v, want none (empty text must not record a classification)", svc.setCalls) } }) } diff --git a/migrations/020_add_emotions_classified_at.sql b/migrations/020_add_emotions_classified_at.sql new file mode 100644 index 00000000..ce63ae08 --- /dev/null +++ b/migrations/020_add_emotions_classified_at.sql @@ -0,0 +1,28 @@ +-- +goose NO TRANSACTION +-- +goose up +-- Emotion completion marker (ENG-1670). `emotions` cannot express "classified, found nothing": +-- the 015 CHECK rejects the empty array, so a successful classification that detects no emotion is +-- persisted as NULL — indistinguishable from "not classified yet". Anything deriving progress from +-- the data (the enrichment-status endpoint, the backlog gauge) therefore counted those records as +-- permanently pending, and the classify backfill re-sent them to the LLM on every run. +-- +-- emotions_classified_at records WHEN the classifier last produced a result for the record, +-- independently of whether that result had any labels. It is processing state, deliberately kept +-- out of the API surface (not in feedbackRecordColumns), so no response shape or SDK type changes. +-- The eager-clear on a value_text edit nulls it alongside `emotions`, so an edited record correctly +-- returns to "not classified". +-- +-- Deliberately NOT backfilled: a bulk UPDATE would rewrite every row of the primary high-write +-- table (WAL + bloat) for no benefit. Readers instead treat a non-NULL `emotions` as proof of +-- completion for pre-existing rows (see the enrichment-status query), which leaves only historical +-- classified-empty rows unresolved; the classify backfill re-processes those once and stamps them. +-- +-- Runs without a transaction like the sibling enrichment migrations. ADD COLUMN of a nullable +-- column with no default is metadata-only (instant, no table rewrite, no long lock), and +-- IF NOT EXISTS keeps the statement re-runnable after an interrupted deploy. +ALTER TABLE feedback_records + ADD COLUMN IF NOT EXISTS emotions_classified_at TIMESTAMPTZ; + +-- +goose down +ALTER TABLE feedback_records + DROP COLUMN IF EXISTS emotions_classified_at; diff --git a/tests/enrichment_status_test.go b/tests/enrichment_status_test.go index 754feabd..53dea39d 100644 --- a/tests/enrichment_status_test.go +++ b/tests/enrichment_status_test.go @@ -126,6 +126,48 @@ func TestCountEnrichmentStatus(t *testing.T) { assert.Equal(t, int64(2), counts.SentimentEligible, "sentiment default-enabled → still eligible") }) + t.Run("emotions completion is the marker, not the labels", func(t *testing.T) { + tenant := testTenantID("enrich-status-emotions-done") + + // A successful classification that detects NO emotion stores NULL labels (the 015 CHECK + // rejects an empty array). Without the completion marker it is indistinguishable from + // "never classified" and would count as pending forever, so the backlog would never drain. + classifiedEmpty := mkRecord(tenant, models.FieldTypeText, "the export ran at 3pm") + require.NoError(t, frepo.SetEmotions(ctx, classifiedEmpty.ID, nil, nil)) + + classifiedWithLabels := mkRecord(tenant, models.FieldTypeText, "I am thrilled") + setEmotions(classifiedWithLabels.ID) + + mkRecord(tenant, models.FieldTypeText, "not processed yet") + + // A pre-020 row: labels present but no marker (the column did not exist when it was + // enriched). Non-NULL labels are themselves proof of completion, which is why the migration + // needs no bulk backfill. + legacy := mkRecord(tenant, models.FieldTypeText, "enriched before the marker existed") + setEmotions(legacy.ID) + _, err := db.Exec(ctx, + `UPDATE feedback_records SET emotions_classified_at = NULL WHERE id = $1`, legacy.ID) + require.NoError(t, err) + + counts, err := statusRepo.CountEnrichmentStatus(ctx, tenant, "") + require.NoError(t, err) + + assert.Equal(t, int64(4), counts.EmotionsEligible, "all four text records are eligible") + assert.Equal(t, int64(3), counts.EmotionsDone, + "classified-empty, classified-with-labels and the legacy row are all done; only the unprocessed one is pending") + + // The eager-clear on a content edit must drop the marker with the labels, or an edited + // record would keep counting as classified while its emotions are gone. + newText := "completely different text now" + _, _, err = frepo.Update(ctx, classifiedWithLabels.ID, + &models.UpdateFeedbackRecordRequest{ValueText: &newText}) + require.NoError(t, err) + + afterEdit, err := statusRepo.CountEnrichmentStatus(ctx, tenant, "") + require.NoError(t, err) + assert.Equal(t, int64(2), afterEdit.EmotionsDone, "editing the text returns that record to pending") + }) + t.Run("translation eligibility follows the effective target", func(t *testing.T) { tenant := testTenantID("enrich-status-trans") // No tenant settings row at all → no own target language. @@ -260,29 +302,41 @@ func TestCountEnrichmentBacklogAggregateIfLeader(t *testing.T) { defer dbRival.Close() - leaderRepo := repository.NewEnrichmentStatusRepository(dbLeader) - rivalRepo := repository.NewEnrichmentStatusRepository(dbRival) + leaderOne := repository.NewEnrichmentBacklogLeader(dbLeader) + leaderTwo := repository.NewEnrichmentBacklogLeader(dbRival) + + // First replica wins leadership and gets real counts. + counts, isLeader, err := leaderOne.CountIfLeader(ctx, "") + require.NoError(t, err) + require.True(t, isLeader, "the first replica to poll becomes the leader") - // Hold the lock by starting a scan on one "replica" and blocking before it commits: emulate by - // taking the lock in an explicit transaction on the leader pool. - holderTx, err := dbLeader.Begin(ctx) + // Prove the leader returns the real aggregate, not a zero value. + want, err := repository.NewEnrichmentStatusRepository(dbLeader).CountEnrichmentBacklogAggregate(ctx, "") require.NoError(t, err) + assert.Equal(t, want.SentimentEligible, counts.SentimentEligible, "leader returns the true aggregate") - var held bool - require.NoError(t, holderTx.QueryRow(ctx, - `SELECT pg_try_advisory_xact_lock(hashtextextended('hub:enrichment-backlog-poller', 0))`).Scan(&held)) - require.True(t, held, "test must hold the poller lock to simulate a busy replica") + // The second replica is denied — a normal skip, not an error — and must NOT publish counts. + zero, isLeader, err := leaderTwo.CountIfLeader(ctx, "") + require.NoError(t, err, "losing the leader election is not an error") + assert.False(t, isLeader, "a second replica must not scan or export the global gauge") + assert.Equal(t, repository.EnrichmentStatusCounts{}, zero, "non-leader returns no counts to publish") - // While it is held, the other replica must be denied — and that is a normal skip, not an error. - _, acquired, err := rivalRepo.CountEnrichmentBacklogAggregateIfLeader(ctx, "") - require.NoError(t, err, "losing the leader race is not an error") - assert.False(t, acquired, "a second replica must not run the aggregate concurrently") + // Leadership is STICKY: unlike a scan-scoped lock, it persists across polls, so the same + // replica keeps exporting the series instead of it flapping between replicas. + _, stillLeader, err := leaderOne.CountIfLeader(ctx, "") + require.NoError(t, err) + assert.True(t, stillLeader, "leadership is held for the process lifetime, not per scan") + + _, stillDenied, err := leaderTwo.CountIfLeader(ctx, "") + require.NoError(t, err) + assert.False(t, stillDenied, "the follower stays a follower while the leader lives") - // Releasing the transaction releases the lock, so the next tick can win. - require.NoError(t, holderTx.Rollback(ctx)) + // Releasing hands leadership over promptly rather than waiting for a session timeout. + leaderOne.Close(ctx) - counts, acquired, err := leaderRepo.CountEnrichmentBacklogAggregateIfLeader(ctx, "") + _, promoted, err := leaderTwo.CountIfLeader(ctx, "") require.NoError(t, err) - assert.True(t, acquired, "lock is free again after the holder's transaction ends") - assert.GreaterOrEqual(t, counts.SentimentEligible, int64(0), "leader gets real counts") + assert.True(t, promoted, "a follower is promoted once the leader releases") + + leaderTwo.Close(ctx) } From 8fbfddf6ac4f7298038e0eb87524fe977a2fff6d Mon Sep 17 00:00:00 2001 From: Tiago Farto Date: Mon, 27 Jul 2026 09:51:17 +0000 Subject: [PATCH 11/13] fix(enrichment): withdraw gauge on demotion, realign backfill index, bound zombie leader (ENG-1670) Three defects found re-reviewing the leader election and the emotions marker: - A demoted leader kept exporting forever. The async gauge re-observes its stored values on every collection and nothing ever cleared them, so after a handover the old leader's frozen reading coexisted with the new leader's live one -- reintroducing exactly the double-count the election was meant to remove, and making the stale copy look like a permanently stuck backlog. The poller now withdraws its series whenever it is not the publishing leader (lost election or failed scan). Exporting nothing is honest; a frozen value is not. - Migration 020 had broken the 016 emotions-backfill index. That index is predicated on `emotions IS NULL` so rows leave it once enriched, keeping it near-empty; classified-empty rows now keep emotions NULL forever, so they would have been retained permanently -- an index growing without bound and a drained backfill forced to scan the whole retained set to find nothing. 020 now rebuilds it with the marker in the predicate, which also removes a post-filter (verified with EXPLAIN) and round-trips cleanly up/down. - A leader lost to node failure or a network partition held the session lock until TCP keepalives reaped the backend (hours by default), with no replica able to take over and no error to alert on. The leader session now sets a 30min idle_session_timeout -- far above the poll interval, so it can only fire on a session that has genuinely stopped polling. Also stamp the marker from the DB clock (CASE WHEN ... THEN NOW()) rather than the pod clock, matching updated_at beside it; extend the eager-clear drift guard to the new column; and release leadership via defer in the leader test so an early failure reports instead of wedging the suite on a checked-out connection. --- cmd/api/app.go | 10 ++++++- internal/observability/enrichment_backlog.go | 16 +++++++++++ .../observability/enrichment_backlog_test.go | 28 +++++++++++++++++++ .../enrichment_status_repository.go | 18 ++++++++++++ .../repository/feedback_records_repository.go | 16 +++++------ .../feedback_records_repository_test.go | 5 +++- migrations/020_add_emotions_classified_at.sql | 23 +++++++++++++++ tests/enrichment_status_test.go | 6 ++++ 8 files changed, 111 insertions(+), 11 deletions(-) diff --git a/cmd/api/app.go b/cmd/api/app.go index a1c578fd..00218ded 100644 --- a/cmd/api/app.go +++ b/cmd/api/app.go @@ -810,6 +810,10 @@ func runEnrichmentBacklogPoller( consecutiveFailures++ + // A failed scan also costs this process its leadership, so withdraw the series rather + // 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. @@ -829,7 +833,11 @@ func runEnrichmentBacklogPoller( consecutiveFailures = 0 if !isLeader { - // Another replica owns this gauge and exports the single global series for it. + // Another replica owns this gauge and exports the single global series for it. Drop + // anything this process exported while it was previously the leader, so a handover + // leaves exactly one series rather than a live one plus a frozen one. + backlog.ClearEnrichmentPending() + return } diff --git a/internal/observability/enrichment_backlog.go b/internal/observability/enrichment_backlog.go index 1da5dc57..3509b1dd 100644 --- a/internal/observability/enrichment_backlog.go +++ b/internal/observability/enrichment_backlog.go @@ -24,6 +24,13 @@ const ( // deliberately NOT a label, so per-tenant detail stays in the API and metric cardinality is bounded. type EnrichmentBacklogMetrics interface { SetEnrichmentPending(enrichment string, count int64) + // ClearEnrichmentPending withdraws every series this process exports for the gauge. The async + // callback re-observes the stored values on EVERY collection, so a process that stops being the + // leader would otherwise keep exporting its final reading forever: the new leader's live series + // and the old leader's frozen one would coexist, a sum would double-count, and the frozen copy + // 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). @@ -92,6 +99,15 @@ func NewEnrichmentBacklogMetrics(meter metric.Meter) (EnrichmentBacklogMetrics, return m, nil } +// ClearEnrichmentPending drops every stored value so the next collection observes nothing and this +// process stops exporting the gauge entirely. +func (m *enrichmentBacklogMetrics) ClearEnrichmentPending() { + m.mu.Lock() + defer m.mu.Unlock() + + clear(m.pending) +} + // RecordPollError counts one failed backlog refresh. func (m *enrichmentBacklogMetrics) RecordPollError(ctx context.Context) { m.pollErrors.Add(ctx, 1) diff --git a/internal/observability/enrichment_backlog_test.go b/internal/observability/enrichment_backlog_test.go index 037249fc..368b815a 100644 --- a/internal/observability/enrichment_backlog_test.go +++ b/internal/observability/enrichment_backlog_test.go @@ -40,6 +40,34 @@ func TestEnrichmentBacklogMetricsGauge(t *testing.T) { assert.Equal(t, int64(3), backlogGaugeValue(t, reader, "translation")) } +// TestEnrichmentBacklogMetricsClearWithdrawsSeries verifies a demoted leader stops exporting +// entirely rather than freezing at its last reading. The async gauge re-observes stored values on +// every collection, so without this a former leader's stale series would coexist with the new +// leader's live one — a sum would double-count and the frozen copy would look like a stuck backlog. +func TestEnrichmentBacklogMetricsClearWithdrawsSeries(t *testing.T) { + reader := sdkmetric.NewManualReader() + provider := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)) + + metrics, err := NewEnrichmentBacklogMetrics(provider.Meter("test")) + require.NoError(t, err) + require.NotNil(t, metrics) + + metrics.SetEnrichmentPending(EnrichmentTypeTranslation, 42) + assert.Equal(t, int64(42), backlogGaugeValue(t, reader, EnrichmentTypeTranslation)) + + metrics.ClearEnrichmentPending() + + var collected metricdata.ResourceMetrics + require.NoError(t, reader.Collect(context.Background(), &collected)) + + 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") + } + } +} + // TestEnrichmentBacklogMetricsPollErrors verifies failed refreshes are counted, so a gauge frozen // at its last value (which looks like a healthy steady backlog) is still alertable. func TestEnrichmentBacklogMetricsPollErrors(t *testing.T) { diff --git a/internal/repository/enrichment_status_repository.go b/internal/repository/enrichment_status_repository.go index 0a580efa..9c744328 100644 --- a/internal/repository/enrichment_status_repository.go +++ b/internal/repository/enrichment_status_repository.go @@ -3,6 +3,7 @@ package repository import ( "context" "fmt" + "log/slog" "time" "github.com/jackc/pgx/v5" @@ -136,6 +137,10 @@ const ( // sessionUnlockSQL releases it. A session lock outlives returning the connection to the pool, // so it must be released explicitly. sessionUnlockSQL = `SELECT pg_advisory_unlock(hashtextextended($1, 0))` + // setLeaderIdleTimeoutSQL caps how long a stalled leader session can hold the lock. Applies to + // this session only (SET, not ALTER ROLE); comfortably above enrichmentBacklogInterval so a + // healthy leader, which queries every interval, is never affected. Requires PG14+. + setLeaderIdleTimeoutSQL = `SET idle_session_timeout = '30min'` ) // CountEnrichmentStatus returns one tenant's eligible/done counts per enrichment. defaultLang is @@ -240,6 +245,19 @@ func (l *EnrichmentBacklogLeader) tryAcquire(ctx context.Context) (bool, error) return false, nil } + // Bound how long a LOST leader can keep the lock. A session lock lives until its backend exits; + // a graceful shutdown releases it via Close, and a killed pod's socket closes, but a node + // failure or network partition leaves a zombie backend holding it until TCP keepalives reap the + // connection -- hours under Linux defaults, during which no replica can take over and the gauge + // just goes absent. An idle timeout on this session alone caps that at one timeout period. The + // leader queries every enrichmentBacklogInterval, so the timeout is set well above it and can + // only fire on a session that has genuinely stopped polling. Best effort: on an older server or + // a restricted role this simply does not apply and behaviour is as before. + if _, err := conn.Exec(ctx, setLeaderIdleTimeoutSQL); err != nil { + slog.WarnContext(ctx, "enrichment backlog: could not bound leader session idle timeout", + "error", err) + } + l.conn = conn return true, nil diff --git a/internal/repository/feedback_records_repository.go b/internal/repository/feedback_records_repository.go index 2adf18df..15b9d6af 100644 --- a/internal/repository/feedback_records_repository.go +++ b/internal/repository/feedback_records_repository.go @@ -1228,18 +1228,16 @@ func (r *FeedbackRecordsRepository) writeEmotions( return err } - // classifiedAt is NULL when clearing and NOW() for a classifier result, so the marker always - // travels with the value it describes. - var classifiedAt any - if classified { - classifiedAt = time.Now() - } - + // The marker is stamped from the DB clock, like updated_at beside it, rather than the pod's + // clock -- every other timestamp on this table is server-generated, and the two should not + // be able to disagree. CASE keeps it NULL when clearing. tag, err := dbTx.Exec(ctx, ` UPDATE feedback_records - SET emotions = $2, emotions_classified_at = $3, updated_at = NOW() + SET emotions = $2, + emotions_classified_at = CASE WHEN $3::boolean THEN NOW() END, + updated_at = NOW() WHERE id = $1`, - feedbackRecordID, emotionsArg, classifiedAt, + feedbackRecordID, emotionsArg, classified, ) if err != nil { return fmt.Errorf("set feedback record emotions: %w", err) diff --git a/internal/repository/feedback_records_repository_test.go b/internal/repository/feedback_records_repository_test.go index f132955c..9e50cc80 100644 --- a/internal/repository/feedback_records_repository_test.go +++ b/internal/repository/feedback_records_repository_test.go @@ -34,7 +34,10 @@ func TestBuildUpdateQuery_ClearsStaleEnrichmentOnContentChange(t *testing.T) { // Enrichment output columns, grouped by what invalidates them. translationCols := []string{"value_text_translated", "translation_lang_key"} - textOnlyCols := []string{"sentiment", "sentiment_score", "emotions"} + // emotions_classified_at must clear with emotions: it is the completion marker, so leaving it + // behind would keep an edited record counting as classified. It must equally NOT clear on a + // language-only edit, which the negative cases below cover. + textOnlyCols := []string{"sentiment", "sentiment_score", "emotions", "emotions_classified_at"} allCols := append(append([]string{}, translationCols...), textOnlyCols...) cases := []struct { diff --git a/migrations/020_add_emotions_classified_at.sql b/migrations/020_add_emotions_classified_at.sql index ce63ae08..04571901 100644 --- a/migrations/020_add_emotions_classified_at.sql +++ b/migrations/020_add_emotions_classified_at.sql @@ -23,6 +23,29 @@ ALTER TABLE feedback_records ADD COLUMN IF NOT EXISTS emotions_classified_at TIMESTAMPTZ; +-- Realign the 016 emotions-backfill index with the new completion semantics. That index was built +-- on `emotions IS NULL` precisely so a row would "leave the index the moment it is enriched", +-- keeping it near-empty once a backfill drains. A classified-empty row now keeps emotions NULL +-- forever (the point of the column above), so under the old predicate it would stay indexed +-- permanently: the index would grow monotonically, and a drained backfill would have to scan the +-- whole retained set and discard every row instead of finishing near-instantly. Adding the marker +-- to the predicate restores the drain property, and also lets the backfill query use the index +-- directly instead of re-checking emotions_classified_at as a post-filter. +-- +-- Built CONCURRENTLY, and DROP-then-CREATE (not IF NOT EXISTS) so an interrupted build leaves an +-- INVALID index that a re-run replaces, matching 016. +DROP INDEX CONCURRENTLY IF EXISTS idx_feedback_records_emotions_backfill; +CREATE INDEX CONCURRENTLY idx_feedback_records_emotions_backfill + ON feedback_records (id) + WHERE field_type = 'text' AND value_text IS NOT NULL + AND emotions IS NULL AND emotions_classified_at IS NULL; + -- +goose down +-- Restore the 016 predicate before dropping the column it references. +DROP INDEX CONCURRENTLY IF EXISTS idx_feedback_records_emotions_backfill; +CREATE INDEX CONCURRENTLY idx_feedback_records_emotions_backfill + ON feedback_records (id) + WHERE field_type = 'text' AND value_text IS NOT NULL AND emotions IS NULL; + ALTER TABLE feedback_records DROP COLUMN IF EXISTS emotions_classified_at; diff --git a/tests/enrichment_status_test.go b/tests/enrichment_status_test.go index 53dea39d..89d5607c 100644 --- a/tests/enrichment_status_test.go +++ b/tests/enrichment_status_test.go @@ -305,6 +305,12 @@ func TestCountEnrichmentBacklogAggregateIfLeader(t *testing.T) { leaderOne := repository.NewEnrichmentBacklogLeader(dbLeader) leaderTwo := repository.NewEnrichmentBacklogLeader(dbRival) + // Release leadership before the pools close. Close is idempotent, and without this an early + // require failure would unwind into pgxpool.Close with the leader's connection still checked + // out, wedging the whole suite until the test timeout instead of reporting the failure. + defer leaderOne.Close(ctx) + defer leaderTwo.Close(ctx) + // First replica wins leadership and gets real counts. counts, isLeader, err := leaderOne.CountIfLeader(ctx, "") require.NoError(t, err) From cb51b5b089400967d109ca3f9fbaa2ffab40d9d1 Mon Sep 17 00:00:00 2001 From: Tiago Farto Date: Mon, 27 Jul 2026 10:01:06 +0000 Subject: [PATCH 12/13] docs(openapi): declare EnrichmentTypeStatus as a Stainless model (ENG-1670) The SDK preview build flagged Model/Recommended for this schema: it is referenced three times (translation, sentiment, emotions), so without a model declaration the generated SDKs emit three structurally identical inline types instead of one shared one. Mirrors the existing taxonomy.run / taxonomy.node declarations. --- openapi.yaml | 3 +++ 1 file changed, 3 insertions(+) diff --git a/openapi.yaml b/openapi.yaml index 7a2a44f6..27b074d4 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -3425,6 +3425,9 @@ components: EnrichmentTypeStatus: type: object additionalProperties: false + # Reused for translation/sentiment/emotions; declare a model so the SDK emits one shared + # type instead of three structurally identical inline copies (Model/Recommended). + x-stainless-model: enrichmentStatus.typeStatus description: One enrichment's progress for a tenant. When `enabled` is false, `eligible` and `done` are zero. properties: enabled: From 9d9c083188fbeb26fd31f3f6388d81a67d1977e9 Mon Sep 17 00:00:00 2001 From: Tiago Farto Date: Mon, 27 Jul 2026 10:05:07 +0000 Subject: [PATCH 13/13] docs(openapi): use snake_case for the EnrichmentTypeStatus model path (ENG-1670) The camelCase model path failed Stainless generation (Name/NotSnakeCase, then a fatal). Model paths are snake_case; the existing taxonomy.run / taxonomy.node declarations are single lowercase words, so the convention was not visible there. --- openapi.yaml | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/openapi.yaml b/openapi.yaml index 27b074d4..b7f7ebd1 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -3427,7 +3427,8 @@ components: additionalProperties: false # Reused for translation/sentiment/emotions; declare a model so the SDK emits one shared # type instead of three structurally identical inline copies (Model/Recommended). - x-stainless-model: enrichmentStatus.typeStatus + # Model paths are snake_case (Name/NotSnakeCase), unlike the resource/method names. + x-stainless-model: enrichment_status.type_status description: One enrichment's progress for a tenant. When `enabled` is false, `eligible` and `done` are zero. properties: enabled: